CASSGO-1 CASSGO-30 Native Protocol 5 Support

.github/workflows/main.yml

@@ -37,8 +37,12 @@ jobs:
         go: [ '1.22', '1.23' ]
         cassandra_version: [ '4.0.13', '4.1.6' ]
         auth: [ "false" ]
-        compressor: [ "snappy" ]
+        compressor: [ "no-compression", "snappy", "lz4" ]
         tags: [ "cassandra", "integration", "ccm" ]
+        proto_version: [ "4", "5" ]
+        exclude:
+          - proto_version: "5"
+            compressor: "snappy"
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-go@v2
@@ -102,7 +106,7 @@ jobs:
           ccm status
           ccm node1 nodetool status
 
-          args="-gocql.timeout=60s -runssl -proto=4 -rf=3 -clusterSize=3 -autowait=2000ms -compressor=${{ matrix.compressor }} -gocql.cversion=$VERSION -cluster=$(ccm liveset) ./..."
+          args="-gocql.timeout=60s -runssl -proto=${{ matrix.proto_version }} -rf=3 -clusterSize=3 -autowait=2000ms -compressor=${{ matrix.compressor }} -gocql.cversion=$VERSION -cluster=$(ccm liveset) ./..."
 
           echo "args=$args" >> $GITHUB_ENV
           echo "JVM_EXTRA_OPTS=$JVM_EXTRA_OPTS" >> $GITHUB_ENV
@@ -115,7 +119,7 @@ jobs:
         if: 'failure()'
         uses: actions/upload-artifact@v4
         with:
-          name: ccm-cluster-cassandra-${{ matrix.cassandra_version }}-go-${{ matrix.go }}-tag-${{ matrix.tags }}
+          name: ccm-cluster-cassandra-${{ matrix.cassandra_version }}-go-${{ matrix.go }}-tag-${{ matrix.tags }}-proto-version-${{ matrix.proto_version }}-compressor-${{ matrix.compressor }}
           path: /home/runner/.ccm/test
           retention-days: 5
   integration-auth-cassandra:
@@ -129,9 +133,12 @@ jobs:
       matrix:
         go: [ '1.22', '1.23' ]
         cassandra_version: [ '4.0.13' ]
-        compressor: [ "snappy" ]
+        compressor: [ "no-compression", "snappy", "lz4" ]
         tags: [ "integration" ]
-
+        proto_version: [ "4", "5" ]
+        exclude:
+          - proto_version: "5"
+            compressor: "snappy"
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v4
@@ -193,7 +200,7 @@ jobs:
           ccm status
           ccm node1 nodetool status
 
-          args="-gocql.timeout=60s -runssl -proto=4 -rf=3 -clusterSize=1 -autowait=2000ms -compressor=${{ matrix.compressor }} -gocql.cversion=$VERSION -cluster=$(ccm liveset) ./..."
+          args="-gocql.timeout=60s -runssl -proto=${{ matrix.proto_version }} -rf=3 -clusterSize=1 -autowait=2000ms -compressor=${{ matrix.compressor }} -gocql.cversion=$VERSION -cluster=$(ccm liveset) ./..."
 
           echo "args=$args" >> $GITHUB_ENV
           echo "JVM_EXTRA_OPTS=$JVM_EXTRA_OPTS" >> $GITHUB_ENV

CHANGELOG.md

@@ -11,6 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Support of sending queries to the specific node with Query.SetHostID() (CASSGO-4)
 
+- Support for Native Protocol 5. Following protocol changes exposed new API 
+  Query.SetKeyspace(), Query.WithNowInSeconds(), Batch.SetKeyspace(), Batch.WithNowInSeconds() (CASSGO-1)
+
 ### Changed
 
 - Move lz4 compressor to lz4 package within the gocql module (CASSGO-32)
@@ -41,6 +44,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Refactor HostInfo creation and ConnectAddress() method (CASSGO-45)
 
+- gocql.Compressor interface changes to follow append-like design. Bumped Go version to 1.19 (CASSGO-1)
+
 ### Fixed
 - Cassandra version unmarshal fix (CASSGO-49)
 

batch_test.go

@@ -28,6 +28,7 @@
 package gocql
 
 import (
+	"github.com/stretchr/testify/require"
 	"testing"
 	"time"
 )
@@ -86,3 +87,84 @@ func TestBatch_WithTimestamp(t *testing.T) {
 		t.Errorf("got ts %d, expected %d", storedTs, micros)
 	}
 }
+
+func TestBatch_WithNowInSeconds(t *testing.T) {
+	session := createSession(t)
+	defer session.Close()
+
+	if session.cfg.ProtoVersion < protoVersion5 {
+		t.Skip("Batch now in seconds are only available on protocol >= 5")
+	}
+
+	if err := createTable(session, `CREATE TABLE IF NOT EXISTS batch_now_in_seconds (id int primary key, val text)`); err != nil {
+		t.Fatal(err)
+	}
+
+	b := session.NewBatch(LoggedBatch)
+	b.WithNowInSeconds(0)
+	b.Query("INSERT INTO batch_now_in_seconds (id, val) VALUES (?, ?) USING TTL 20", 1, "val")
+	if err := session.ExecuteBatch(b); err != nil {
+		t.Fatal(err)
+	}
+
+	var remainingTTL int
+	err := session.Query(`SELECT TTL(val) FROM batch_now_in_seconds WHERE id = ?`, 1).
+		WithNowInSeconds(10).
+		Scan(&remainingTTL)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	require.Equal(t, remainingTTL, 10)
+}
+
+func TestBatch_SetKeyspace(t *testing.T) {
+	session := createSession(t)
+	defer session.Close()
+
+	if session.cfg.ProtoVersion < protoVersion5 {
+		t.Skip("keyspace for BATCH message is not supported in protocol < 5")
+	}
+
+	const keyspaceStmt = `
+		CREATE KEYSPACE IF NOT EXISTS gocql_keyspace_override_test 
+		WITH replication = {
+			'class': 'SimpleStrategy', 
+			'replication_factor': '1'
+		};
+`
+
+	err := session.Query(keyspaceStmt).Exec()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = createTable(session, "CREATE TABLE IF NOT EXISTS gocql_keyspace_override_test.batch_keyspace(id int, value text, PRIMARY KEY (id))")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ids := []int{1, 2}
+	texts := []string{"val1", "val2"}
+
+	b := session.NewBatch(LoggedBatch).SetKeyspace("gocql_keyspace_override_test")
+	b.Query("INSERT INTO batch_keyspace(id, value) VALUES (?, ?)", ids[0], texts[0])
+	b.Query("INSERT INTO batch_keyspace(id, value) VALUES (?, ?)", ids[1], texts[1])
+	err = session.ExecuteBatch(b)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var (
+		id   int
+		text string
+	)
+
+	iter := session.Query("SELECT * FROM gocql_keyspace_override_test.batch_keyspace").Iter()
+	defer iter.Close()
+
+	for i := 0; iter.Scan(&id, &text); i++ {
+		require.Equal(t, id, ids[i])
+		require.Equal(t, text, texts[i])
+	}
+}