CASSGO-11 Support vector type

Support marshalling and unmarshalling of vector custom type.

patched by Lukasz Antoniak; reviewed by João Reis, Stanislav Bychkov, Oleksandr Luzhniy, Mykyta Oleksiienko and Bohdan Siryk for CASSGO-11

Author: lukasz-antoniak

.github/workflows/main.yml

@@ -8,7 +8,7 @@ on:
     types: [ opened, synchronize, reopened ]
 
 env:
-  CCM_VERSION: "6e71061146f7ae67b84ccd2b1d90d7319b640e4c"
+  CCM_VERSION: "4621dfee5ad73956b831091a8b863d100d25c610"
 
 jobs:
   build:
@@ -35,7 +35,7 @@ jobs:
       fail-fast: false
       matrix:
         go: [ '1.22', '1.23' ]
-        cassandra_version: [ '4.0.13', '4.1.6' ]
+        cassandra_version: [ '4.1.6', '5.0.3' ]
         auth: [ "false" ]
         compressor: [ "no-compression", "snappy", "lz4" ]
         tags: [ "cassandra", "integration", "ccm" ]

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Support vector type [CASSGO-11](https://issues.apache.org/jira/browse/CASSGO-11)
+
 - Allow SERIAL and LOCAL_SERIAL on SELECT statements [CASSGO-26](https://issues.apache.org/jira/browse/CASSGO-26)
 
 - Support of sending queries to the specific node with Query.SetHostID() (CASSGO-4)

cassandra_test.go

@@ -481,15 +481,15 @@ func TestCAS(t *testing.T) {
 	}
 
 	insertBatch := session.Batch(LoggedBatch)
-	insertBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES ('_foo', 2c3af400-73a4-11e5-9381-29463d90c3f0, DATEOF(NOW()))")
-	insertBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES ('_foo', 3e4ad2f1-73a4-11e5-9381-29463d90c3f0, DATEOF(NOW()))")
+	insertBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES ('_foo', 2c3af400-73a4-11e5-9381-29463d90c3f0, TOTIMESTAMP(NOW()))")
+	insertBatch.Query("INSERT INTO cas_table (title, revid, last_modified) VALUES ('_foo', 3e4ad2f1-73a4-11e5-9381-29463d90c3f0, TOTIMESTAMP(NOW()))")
 	if err := session.ExecuteBatch(insertBatch); err != nil {
 		t.Fatal("insert:", err)
 	}
 
 	failBatch = session.Batch(LoggedBatch)
-	failBatch.Query("UPDATE cas_table SET last_modified = DATEOF(NOW()) WHERE title='_foo' AND revid=2c3af400-73a4-11e5-9381-29463d90c3f0 IF last_modified=DATEOF(NOW());")
-	failBatch.Query("UPDATE cas_table SET last_modified = DATEOF(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified=DATEOF(NOW());")
+	failBatch.Query("UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=2c3af400-73a4-11e5-9381-29463d90c3f0 IF last_modified=TOTIMESTAMP(NOW());")
+	failBatch.Query("UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified=TOTIMESTAMP(NOW());")
 	if applied, iter, err := session.ExecuteBatchCAS(failBatch, &titleCAS, &revidCAS, &modifiedCAS); err != nil {
 		t.Fatal("insert:", err)
 	} else if applied {
@@ -533,21 +533,21 @@ func TestCAS(t *testing.T) {
 	}
 
 	failBatch = session.Batch(LoggedBatch)
-	failBatch.Query("UPDATE cas_table SET last_modified = DATEOF(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified = ?", modified)
+	failBatch.Query("UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified = ?", modified)
 	if _, _, err := session.ExecuteBatchCAS(failBatch, new(bool)); err == nil {
 		t.Fatal("update should have errored")
 	}
 	// make sure MapScanCAS does not panic when MapScan fails
 	casMap = make(map[string]interface{})
 	casMap["last_modified"] = false
-	if _, err := session.Query(`UPDATE cas_table SET last_modified = DATEOF(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified = ?`,
+	if _, err := session.Query(`UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified = ?`,
 		modified).MapScanCAS(casMap); err == nil {
 		t.Fatal("update should hvae errored", err)
 	}
 
 	// make sure MapExecuteBatchCAS does not panic when MapScan fails
 	failBatch = session.Batch(LoggedBatch)
-	failBatch.Query("UPDATE cas_table SET last_modified = DATEOF(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified = ?", modified)
+	failBatch.Query("UPDATE cas_table SET last_modified = TOTIMESTAMP(NOW()) WHERE title='_foo' AND revid=3e4ad2f1-73a4-11e5-9381-29463d90c3f0 IF last_modified = ?", modified)
 	casMap = make(map[string]interface{})
 	casMap["last_modified"] = false
 	if _, _, err := session.MapExecuteBatchCAS(failBatch, casMap); err == nil {
@@ -2516,18 +2516,19 @@ func TestAggregateMetadata(t *testing.T) {
 		t.Fatal("expected two aggregates")
 	}
 
+	protoVer := byte(session.cfg.ProtoVersion)
 	expectedAggregrate := AggregateMetadata{
 		Keyspace:      "gocql_test",
 		Name:          "average",
-		ArgumentTypes: []TypeInfo{NativeType{typ: TypeInt}},
+		ArgumentTypes: []TypeInfo{NativeType{typ: TypeInt, proto: protoVer}},
 		InitCond:      "(0, 0)",
-		ReturnType:    NativeType{typ: TypeDouble},
+		ReturnType:    NativeType{typ: TypeDouble, proto: protoVer},
 		StateType: TupleTypeInfo{
-			NativeType: NativeType{typ: TypeTuple},
+			NativeType: NativeType{typ: TypeTuple, proto: protoVer},
 
 			Elems: []TypeInfo{
-				NativeType{typ: TypeInt},
-				NativeType{typ: TypeBigInt},
+				NativeType{typ: TypeInt, proto: protoVer},
+				NativeType{typ: TypeBigInt, proto: protoVer},
 			},
 		},
 		stateFunc: "avgstate",
@@ -2566,28 +2567,29 @@ func TestFunctionMetadata(t *testing.T) {
 	avgState := functions[1]
 	avgFinal := functions[0]
 
+	protoVer := byte(session.cfg.ProtoVersion)
 	avgStateBody := "if (val !=null) {state.setInt(0, state.getInt(0)+1); state.setLong(1, state.getLong(1)+val.intValue());}return state;"
 	expectedAvgState := FunctionMetadata{
 		Keyspace: "gocql_test",
 		Name:     "avgstate",
 		ArgumentTypes: []TypeInfo{
 			TupleTypeInfo{
-				NativeType: NativeType{typ: TypeTuple},
+				NativeType: NativeType{typ: TypeTuple, proto: protoVer},
 
 				Elems: []TypeInfo{
-					NativeType{typ: TypeInt},
-					NativeType{typ: TypeBigInt},
+					NativeType{typ: TypeInt, proto: protoVer},
+					NativeType{typ: TypeBigInt, proto: protoVer},
 				},
 			},
-			NativeType{typ: TypeInt},
+			NativeType{typ: TypeInt, proto: protoVer},
 		},
 		ArgumentNames: []string{"state", "val"},
 		ReturnType: TupleTypeInfo{
-			NativeType: NativeType{typ: TypeTuple},
+			NativeType: NativeType{typ: TypeTuple, proto: protoVer},
 
 			Elems: []TypeInfo{
-				NativeType{typ: TypeInt},
-				NativeType{typ: TypeBigInt},
+				NativeType{typ: TypeInt, proto: protoVer},
+				NativeType{typ: TypeBigInt, proto: protoVer},
 			},
 		},
 		CalledOnNullInput: true,
@@ -2604,16 +2606,16 @@ func TestFunctionMetadata(t *testing.T) {
 		Name:     "avgfinal",
 		ArgumentTypes: []TypeInfo{
 			TupleTypeInfo{
-				NativeType: NativeType{typ: TypeTuple},
+				NativeType: NativeType{typ: TypeTuple, proto: protoVer},
 
 				Elems: []TypeInfo{
-					NativeType{typ: TypeInt},
-					NativeType{typ: TypeBigInt},
+					NativeType{typ: TypeInt, proto: protoVer},
+					NativeType{typ: TypeBigInt, proto: protoVer},
 				},
 			},
 		},
 		ArgumentNames:     []string{"state"},
-		ReturnType:        NativeType{typ: TypeDouble},
+		ReturnType:        NativeType{typ: TypeDouble, proto: protoVer},
 		CalledOnNullInput: true,
 		Language:          "java",
 		Body:              finalStateBody,
@@ -2717,15 +2719,16 @@ func TestKeyspaceMetadata(t *testing.T) {
 	if flagCassVersion.Before(3, 0, 0) {
 		textType = TypeVarchar
 	}
+	protoVer := byte(session.cfg.ProtoVersion)
 	expectedType := UserTypeMetadata{
 		Keyspace:   "gocql_test",
 		Name:       "basicview",
 		FieldNames: []string{"birthday", "nationality", "weight", "height"},
 		FieldTypes: []TypeInfo{
-			NativeType{typ: TypeTimestamp},
-			NativeType{typ: textType},
-			NativeType{typ: textType},
-			NativeType{typ: textType},
+			NativeType{typ: TypeTimestamp, proto: protoVer},
+			NativeType{typ: textType, proto: protoVer},
+			NativeType{typ: textType, proto: protoVer},
+			NativeType{typ: textType, proto: protoVer},
 		},
 	}
 	if !reflect.DeepEqual(*keyspaceMetadata.UserTypes["basicview"], expectedType) {

common_test.go

@@ -28,6 +28,7 @@ import (
 	"flag"
 	"fmt"
 	"log"
+	"math/rand"
 	"net"
 	"reflect"
 	"strings"
@@ -54,6 +55,10 @@ var (
 	flagCassVersion cassVersion
 )
 
+var seededRand *rand.Rand = rand.New(rand.NewSource(time.Now().UnixNano()))
+
+const randCharset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+
 func init() {
 	flag.Var(&flagCassVersion, "gocql.cversion", "the cassandra version being tested against")
 
@@ -281,6 +286,14 @@ func assertTrue(t *testing.T, description string, value bool) {
 	}
 }
 
+func randomText(size int) string {
+	result := make([]byte, size)
+	for i := range result {
+		result[i] = randCharset[rand.Intn(len(randCharset))]
+	}
+	return string(result)
+}
+
 func assertEqual(t *testing.T, description string, expected, actual interface{}) {
 	t.Helper()
 	if expected != actual {

frame.go

@@ -34,6 +34,7 @@ import (
 	"io/ioutil"
 	"net"
 	"runtime"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -905,6 +906,22 @@ func (f *framer) readTypeInfo() TypeInfo {
 		collection.Elem = f.readTypeInfo()
 
 		return collection
+	case TypeCustom:
+		if strings.HasPrefix(simple.custom, VECTOR_TYPE) {
+			spec := strings.TrimPrefix(simple.custom, VECTOR_TYPE)
+			spec = spec[1 : len(spec)-1] // remove parenthesis
+			idx := strings.LastIndex(spec, ",")
+			typeStr := spec[:idx]
+			dimStr := spec[idx+1:]
+			subType := getCassandraLongType(strings.TrimSpace(typeStr), f.proto, nopLogger{})
+			dim, _ := strconv.Atoi(strings.TrimSpace(dimStr))
+			vector := VectorType{
+				NativeType: simple,
+				SubType:    subType,
+				Dimensions: dim,
+			}
+			return vector
+		}
 	}
 
 	return simple