Repository: gocql/gocql
Pull request #1439 (2020-05-11 16:16:22)

Add Attempt count to ObservedBatch
We have attempt count in ObservedQuery, add it to ObservedBatch as well
for feature parity.


Patch:
diff --git a/.travis.yml b/.travis.yml
index 778a1be..22a2f80 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -31,8 +31,8 @@ env:
       AUTH=false
 
 go:
-  - 1.12.x
-  - 1.13.x
+  - 1.14.x
+  - 1.15.x
 
 install:
   - ./install_test_deps.sh $TRAVIS_REPO_SLUG
diff --git a/README.md b/README.md
index ccb4803..980ca8e 100644
--- a/README.md
+++ b/README.md
@@ -19,8 +19,8 @@ The following matrix shows the versions of Go and Cassandra that are tested with
 
 Go/Cassandra | 2.1.x | 2.2.x | 3.x.x
 -------------| -------| ------| ---------
-1.12 | yes | yes | yes
-1.13 | yes | yes | yes
+1.14 | yes | yes | yes
+1.15 | yes | yes | yes
 
 Gocql has been tested in production against many different versions of Cassandra. Due to limits in our CI setup we only test against the latest 3 major releases, which coincide with the official support from the Apache project.
 
diff --git a/conn.go b/conn.go
index 81210c4..de9e823 100644
--- a/conn.go
+++ b/conn.go
@@ -31,6 +31,7 @@ var (
 		"io.aiven.cassandra.auth.AivenAuthenticator",
 		"com.ericsson.bss.cassandra.ecaudit.auth.AuditPasswordAuthenticator",
 		"com.amazon.helenus.auth.HelenusAuthenticator",
+		"com.ericsson.bss.cassandra.ecaudit.auth.AuditAuthenticator",
 	}
 )
 
@@ -43,8 +44,8 @@ func approve(authenticator string) bool {
 	return false
 }
 
-//JoinHostPort is a utility to return a address string that can be used
-//gocql.Conn to form a connection with a host.
+// JoinHostPort is a utility to return an address string that can be used
+// by `gocql.Conn` to form a connection with a host.
 func JoinHostPort(addr string, port int) string {
 	addr = strings.TrimSpace(addr)
 	if _, _, err := net.SplitHostPort(addr); err != nil {
@@ -124,7 +125,7 @@ func (fn connErrorHandlerFn) HandleError(conn *Conn, err error, closed bool) {
 // which may be serving more queries just fine.
 // Default is 0, should not be changed concurrently with queries.
 //
-// depreciated
+// Deprecated.
 var TimeoutLimit int64 = 0
 
 // Conn is a single connection to a Cassandra node. It can be used to execute
@@ -212,7 +213,6 @@ func (s *Session) dialWithoutObserver(ctx context.Context, host *HostInfo, cfg *
 		dialer = d
 	}
 
-
 	conn, err := dialer.DialContext(ctx, "tcp", host.HostnameAndPort())
 	if err != nil {
 		return nil, err
diff --git a/marshal.go b/marshal.go
index fda5be6..e95c1c8 100644
--- a/marshal.go
+++ b/marshal.go
@@ -1839,6 +1839,11 @@ func marshalTuple(info TypeInfo, value interface{}) ([]byte, error) {
 
 		var buf []byte
 		for i, elem := range v {
+			if elem == nil {
+				buf = appendInt(buf, int32(-1))
+				continue
+			}
+
 			data, err := Marshal(tuple.Elems[i], elem)
 			if err != nil {
 				return nil, err
@@ -1864,7 +1869,14 @@ func marshalTuple(info TypeInfo, value interface{}) ([]byte, error) {
 
 		var buf []byte
 		for i, elem := range tuple.Elems {
-			data, err := Marshal(elem, rv.Field(i).Interface())
+			field := rv.Field(i)
+
+			if field.Kind() == reflect.Ptr && field.IsNil() {
+				buf = appendInt(buf, int32(-1))
+				continue
+			}
+
+			data, err := Marshal(elem, field.Interface())
 			if err != nil {
 				return nil, err
 			}
@@ -1883,7 +1895,14 @@ func marshalTuple(info TypeInfo, value interface{}) ([]byte, error) {
 
 		var buf []byte
 		for i, elem := range tuple.Elems {
-			data, err := Marshal(elem, rv.Index(i).Interface())
+			item := rv.Index(i)
+
+			if item.Kind() == reflect.Ptr && item.IsNil() {
+				buf = appendInt(buf, int32(-1))
+				continue
+			}
+
+			data, err := Marshal(elem, item.Interface())
 			if err != nil {
 				return nil, err
 			}
@@ -1951,16 +1970,22 @@ func unmarshalTuple(info TypeInfo, data []byte, value interface{}) error {
 		}
 
 		for i, elem := range tuple.Elems {
-			m := readInt(data)
-			data = data[4:]
+			var p []byte
+			if len(data) > 4 {
+				p, data = readBytes(data)
+			}
 
 			v := elem.New()
-			if err := Unmarshal(elem, data[:m], v); err != nil {
+			if err := Unmarshal(elem, p, v); err != nil {
 				return err
 			}
-			rv.Field(i).Set(reflect.ValueOf(v).Elem())
 
-			data = data[m:]
+			switch rv.Field(i).Kind() {
+			case reflect.Ptr:
+				rv.Field(i).Set(reflect.ValueOf(v))
+			default:
+				rv.Field(i).Set(reflect.ValueOf(v).Elem())
+			}
 		}
 
 		return nil
@@ -1975,16 +2000,22 @@ func unmarshalTuple(info TypeInfo, data []byte, value interface{}) error {
 		}
 
 		for i, elem := range tuple.Elems {
-			m := readInt(data)
-			data = data[4:]
+			var p []byte
+			if len(data) > 4 {
+				p, data = readBytes(data)
+			}
 
 			v := elem.New()
-			if err := Unmarshal(elem, data[:m], v); err != nil {
+			if err := Unmarshal(elem, p, v); err != nil {
 				return err
 			}
-			rv.Index(i).Set(reflect.ValueOf(v).Elem())
 
-			data = data[m:]
+			switch rv.Index(i).Kind() {
+			case reflect.Ptr:
+				rv.Index(i).Set(reflect.ValueOf(v))
+			default:
+				rv.Index(i).Set(reflect.ValueOf(v).Elem())
+			}
 		}
 
 		return nil
diff --git a/metadata.go b/metadata.go
index 5a17559..e586dd4 100644
--- a/metadata.go
+++ b/metadata.go
@@ -22,7 +22,10 @@ type KeyspaceMetadata struct {
 	Tables          map[string]*TableMetadata
 	Functions       map[string]*FunctionMetadata
 	Aggregates      map[string]*AggregateMetadata
-	Views           map[string]*ViewMetadata
+	// Deprecated: use the MaterializedViews field for views and UserTypes field for udts instead.
+	Views             map[string]*ViewMetadata
+	MaterializedViews map[string]*MaterializedViewMetadata
+	UserTypes         map[string]*UserTypeMetadata
 }
 
 // schema metadata for a table (a.k.a. column family)
@@ -83,6 +86,7 @@ type AggregateMetadata struct {
 }
 
 // ViewMetadata holds the metadata for views.
+// Deprecated: this is kept for backwards compatibility issues. Use MaterializedViewMetadata.
 type ViewMetadata struct {
 	Keyspace   string
 	Name       string
@@ -90,6 +94,40 @@ type ViewMetadata struct {
 	FieldTypes []TypeInfo
 }
 
+// MaterializedViewMetadata holds the metadata for materialized views.
+type MaterializedViewMetadata struct {
+	Keyspace                string
+	Name                    string
+	BaseTableId             UUID
+	BaseTable               *TableMetadata
+	BloomFilterFpChance     float64
+	Caching                 map[string]string
+	Comment                 string
+	Compaction              map[string]string
+	Compression             map[string]string
+	CrcCheckChance          float64
+	DcLocalReadRepairChance float64
+	DefaultTimeToLive       int
+	Extensions              map[string]string
+	GcGraceSeconds          int
+	Id                      UUID
+	IncludeAllColumns       bool
+	MaxIndexInterval        int
+	MemtableFlushPeriodInMs int
+	MinIndexInterval        int
+	ReadRepairChance        float64
+	SpeculativeRetry        string
+
+	baseTableName string
+}
+
+type UserTypeMetadata struct {
+	Keyspace   string
+	Name       string
+	FieldNames []string
+	FieldTypes []TypeInfo
+}
+
 // the ordering of the column with regard to its comparator
 type ColumnOrder bool
 
@@ -246,9 +284,13 @@ func (s *schemaDescriber) refreshSchema(keyspaceName string) error {
 	if err != nil {
 		return err
 	}
+	materializedViews, err := getMaterializedViewsMetadata(s.session, keyspaceName)
+	if err != nil {
+		return err
+	}
 
 	// organize the schema data
-	compileMetadata(s.session.cfg.ProtoVersion, keyspace, tables, columns, functions, aggregates, views)
+	compileMetadata(s.session.cfg.ProtoVersion, keyspace, tables, columns, functions, aggregates, views, materializedViews)
 
 	// update the cache
 	s.cache[keyspaceName] = keyspace
@@ -269,6 +311,7 @@ func compileMetadata(
 	functions []FunctionMetadata,
 	aggregates []AggregateMetadata,
 	views []ViewMetadata,
+	materializedViews []MaterializedViewMetadata,
 ) {
 	keyspace.Tables = make(map[string]*TableMetadata)
 	for i := range tables {
@@ -290,6 +333,24 @@ func compileMetadata(
 	for i := range views {
 		keyspace.Views[views[i].Name] = &views[i]
 	}
+	// Views currently holds the types and hasn't been deleted for backward compatibility issues.
+	// That's why it's ok to copy Views into Types in this case. For the real Views use MaterializedViews.
+	types := make([]UserTypeMetadata, len(views))
+	for i := range views {
+		types[i].Keyspace = views[i].Keyspace
+		types[i].Name = views[i].Name
+		types[i].FieldNames = views[i].FieldNames
+		types[i].FieldTypes = views[i].FieldTypes
+	}
+	keyspace.UserTypes = make(map[string]*UserTypeMetadata, len(views))
+	for i := range types {
+		keyspace.UserTypes[types[i].Name] = &types[i]
+	}
+	keyspace.MaterializedViews = make(map[string]*MaterializedViewMetadata, len(materializedViews))
+	for _, materializedView := range materializedViews {
+		materializedView.BaseTable = keyspace.Tables[materializedView.baseTableName]
+		keyspace.MaterializedViews[materializedView.Name] = &materializedView
+	}
 
 	// add columns from the schema data
 	for i := range columns {
@@ -912,6 +973,75 @@ func getViewsMetadata(session *Session, keyspaceName string) ([]ViewMetadata, er
 	return views, nil
 }
 
+func getMaterializedViewsMetadata(session *Session, keyspaceName string) ([]MaterializedViewMetadata, error) {
+	if !session.useSystemSchema {
+		return nil, nil
+	}
+	var tableName = "system_schema.views"
+	stmt := fmt.Sprintf(`
+		SELECT
+			view_name,
+			base_table_id,
+			base_table_name,
+			bloom_filter_fp_chance,
+			caching,
+			comment,
+			compaction,
+			compression,
+			crc_check_chance,
+			dclocal_read_repair_chance,
+			default_time_to_live,
+			extensions,
+			gc_grace_seconds,
+			id,
+			include_all_columns,
+			max_index_interval,
+			memtable_flush_period_in_ms,
+			min_index_interval,
+			read_repair_chance,
+			speculative_retry
+		FROM %s
+		WHERE keyspace_name = ?`, tableName)
+
+	var materializedViews []MaterializedViewMetadata
+
+	rows := session.control.query(stmt, keyspaceName).Scanner()
+	for rows.Next() {
+		materializedView := MaterializedViewMetadata{Keyspace: keyspaceName}
+		err := rows.Scan(&materializedView.Name,
+			&materializedView.BaseTableId,
+			&materializedView.baseTableName,
+			&materializedView.BloomFilterFpChance,
+			&materializedView.Caching,
+			&materializedView.Comment,
+			&materializedView.Compaction,
+			&materializedView.Compression,
+			&materializedView.CrcCheckChance,
+			&materializedView.DcLocalReadRepairChance,
+			&materializedView.DefaultTimeToLive,
+			&materializedView.Extensions,
+			&materializedView.GcGraceSeconds,
+			&materializedView.Id,
+			&materializedView.IncludeAllColumns,
+			&materializedView.MaxIndexInterval,
+			&materializedView.MemtableFlushPeriodInMs,
+			&materializedView.MinIndexInterval,
+			&materializedView.ReadRepairChance,
+			&materializedView.SpeculativeRetry,
+		)
+		if err != nil {
+			return nil, err
+		}
+		materializedViews = append(materializedViews, materializedView)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return materializedViews, nil
+}
+
 func getFunctionsMetadata(session *Session, keyspaceName string) ([]FunctionMetadata, error) {
 	if session.cfg.ProtoVersion == protoVersion1 || !session.hasAggregatesAndFunctions {
 		return nil, nil
diff --git a/session.go b/session.go
index 6c7edff..afeae47 100644
--- a/session.go
+++ b/session.go
@@ -27,7 +27,7 @@ import (
 // scenario is to have one global session object to interact with the
 // whole Cassandra cluster.
 //
-// This type extends the Node interface by adding a convinient query builder
+// This type extends the Node interface by adding a convenient query builder
 // and automatically sets a default consistency level on all operations
 // that do not have a consistency level set.
 type Session struct {
@@ -655,7 +655,7 @@ func (s *Session) ExecuteBatch(batch *Batch) error {
 }
 
 // ExecuteBatchCAS executes a batch operation and returns true if successful and
-// an iterator (to scan aditional rows if more than one conditional statement)
+// an iterator (to scan additional rows if more than one conditional statement)
 // was sent.
 // Further scans on the interator must also remember to include
 // the applied boolean as the first argument to *Iter.Scan
@@ -698,7 +698,7 @@ func (s *Session) MapExecuteBatchCAS(batch *Batch, dest map[string]interface{})
 type hostMetrics struct {
 	// Attempts is count of how many times this query has been attempted for this host.
 	// An attempt is either a retry or fetching next page of results.
-	Attempts     int
+	Attempts int
 
 	// TotalLatency is the sum of attempt latencies for this host in nanoseconds.
 	TotalLatency int64
@@ -872,7 +872,7 @@ func (q *Query) Latency() int64 {
 }
 
 func (q *Query) AddLatency(l int64, host *HostInfo) {
-	q.metrics.attempt(0, time.Duration(l) * time.Nanosecond, host, false)
+	q.metrics.attempt(0, time.Duration(l)*time.Nanosecond, host, false)
 }
 
 // Consistency sets the consistency level for this query. If no consistency
@@ -944,7 +944,7 @@ func (q *Query) DefaultTimestamp(enable bool) *Query {
 // WithTimestamp will enable the with default timestamp flag on the query
 // like DefaultTimestamp does. But also allows to define value for timestamp.
 // It works the same way as USING TIMESTAMP in the query itself, but
-// should not break prepared query optimization
+// should not break prepared query optimization.
 //
 // Only available on protocol >= 3
 func (q *Query) WithTimestamp(timestamp int64) *Query {
@@ -1135,7 +1135,7 @@ func (q *Query) PageState(state []byte) *Query {
 // NoSkipMetadata will override the internal result metadata cache so that the driver does not
 // send skip_metadata for queries, this means that the result will always contain
 // the metadata to parse the rows and will not reuse the metadata from the prepared
-// staement. This should only be used to work around cassandra bugs, such as when using
+// statement. This should only be used to work around cassandra bugs, such as when using
 // CAS operations which do not end in Cas.
 //
 // See https://issues.apache.org/jira/browse/CASSANDRA-11099
@@ -1608,7 +1608,7 @@ func (b *Batch) Latency() int64 {
 }
 
 func (b *Batch) AddLatency(l int64, host *HostInfo) {
-	b.metrics.attempt(0, time.Duration(l) * time.Nanosecond, host, false)
+	b.metrics.attempt(0, time.Duration(l)*time.Nanosecond, host, false)
 }
 
 // GetConsistency returns the currently configured consistency level for the batch
@@ -1722,7 +1722,7 @@ func (b *Batch) DefaultTimestamp(enable bool) *Batch {
 // WithTimestamp will enable the with default timestamp flag on the query
 // like DefaultTimestamp does. But also allows to define value for timestamp.
 // It works the same way as USING TIMESTAMP in the query itself, but
-// should not break prepared query optimization
+// should not break prepared query optimization.
 //
 // Only available on protocol >= 3
 func (b *Batch) WithTimestamp(timestamp int64) *Batch {
@@ -1733,7 +1733,7 @@ func (b *Batch) WithTimestamp(timestamp int64) *Batch {
 
 func (b *Batch) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
 	latency := end.Sub(start)
-	_, metricsForHost := b.metrics.attempt(1, latency, host, b.observer != nil)
+	attempt, metricsForHost := b.metrics.attempt(1, latency, host, b.observer != nil)
 
 	if b.observer == nil {
 		return
@@ -1753,6 +1753,7 @@ func (b *Batch) attempt(keyspace string, end, start time.Time, iter *Iter, host
 		Host:    host,
 		Metrics: metricsForHost,
 		Err:     iter.err,
+		Attempt: attempt,
 	})
 }
 
@@ -1999,6 +2000,11 @@ type ObservedBatch struct {
 
 	// The metrics per this host
 	Metrics *hostMetrics
+
+	// Attempt is the index of attempt at executing this query.
+	// An attempt might be either retry or fetching next page of a query.
+	// The first attempt is number zero and any retries have non-zero attempt number.
+	Attempt int
 }
 
 // BatchObserver is the interface implemented by batch observers / stat collectors.


Test patch:
diff --git a/cassandra_test.go b/cassandra_test.go
index 20e47b7..d06274b 100644
--- a/cassandra_test.go
+++ b/cassandra_test.go
@@ -325,12 +325,12 @@ func TestPagingWithBind(t *testing.T) {
 		t.Fatal("create table:", err)
 	}
 	for i := 0; i < 100; i++ {
-		if err := session.Query("INSERT INTO paging_bind (id,val) VALUES (?,?)", 1,i).Exec(); err != nil {
+		if err := session.Query("INSERT INTO paging_bind (id,val) VALUES (?,?)", 1, i).Exec(); err != nil {
 			t.Fatal("insert:", err)
 		}
 	}
 
-	q := session.Query("SELECT val FROM paging_bind WHERE id = ? AND val < ?",1, 50).PageSize(10)
+	q := session.Query("SELECT val FROM paging_bind WHERE id = ? AND val < ?", 1, 50).PageSize(10)
 	iter := q.Iter()
 	var id int
 	count := 0
@@ -2233,6 +2233,50 @@ func TestViewMetadata(t *testing.T) {
 	}
 }
 
+func TestMaterializedViewMetadata(t *testing.T) {
+	if flagCassVersion.Before(3, 0, 0) {
+		return
+	}
+	session := createSession(t)
+	defer session.Close()
+	createMaterializedViews(t, session)
+
+	materializedViews, err := getMaterializedViewsMetadata(session, "gocql_test")
+	if err != nil {
+		t.Fatalf("failed to query view metadata with err: %v", err)
+	}
+	if materializedViews == nil {
+		t.Fatal("failed to query view metadata, nil returned")
+	}
+	if len(materializedViews) != 1 {
+		t.Fatal("expected one view")
+	}
+	expectedView := MaterializedViewMetadata{
+		Keyspace:                "gocql_test",
+		Name:                    "view_view",
+		baseTableName:           "view_table",
+		BloomFilterFpChance:     0.01,
+		Caching:                 map[string]string{"keys": "ALL", "rows_per_partition": "NONE"},
+		Comment:                 "",
+		Compaction:              map[string]string{"class": "org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy", "max_threshold": "32", "min_threshold": "4"},
+		Compression:             map[string]string{"chunk_length_in_kb": "64", "class": "org.apache.cassandra.io.compress.LZ4Compressor"},
+		CrcCheckChance:          1,
+		DcLocalReadRepairChance: 0.1,
+		DefaultTimeToLive:       0,
+		Extensions:              map[string]string{},
+		GcGraceSeconds:          864000,
+		IncludeAllColumns:       false, MaxIndexInterval: 2048, MemtableFlushPeriodInMs: 0, MinIndexInterval: 128, ReadRepairChance: 0,
+		SpeculativeRetry: "99PERCENTILE",
+	}
+
+	expectedView.BaseTableId = materializedViews[0].BaseTableId
+	expectedView.Id = materializedViews[0].Id
+
+	if !reflect.DeepEqual(materializedViews[0], expectedView) {
+		t.Fatalf("materialized view is %+v, but expected %+v", materializedViews[0], expectedView)
+	}
+}
+
 func TestAggregateMetadata(t *testing.T) {
 	session := createSession(t)
 	defer session.Close()
@@ -2363,6 +2407,7 @@ func TestKeyspaceMetadata(t *testing.T) {
 	}
 	createAggregate(t, session)
 	createViews(t, session)
+	createMaterializedViews(t, session)
 
 	if err := session.Query("CREATE INDEX index_metadata ON test_metadata ( third_id )").Exec(); err != nil {
 		t.Fatalf("failed to create index with err: %v", err)
@@ -2433,6 +2478,37 @@ func TestKeyspaceMetadata(t *testing.T) {
 	if !found {
 		t.Fatal("failed to find the view in metadata")
 	}
+	_, found = keyspaceMetadata.UserTypes["basicview"]
+	if !found {
+		t.Fatal("failed to find the types in metadata")
+	}
+	textType := TypeText
+	if flagCassVersion.Before(3, 0, 0) {
+		textType = TypeVarchar
+	}
+	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},
+		},
+	}
+	if !reflect.DeepEqual(*keyspaceMetadata.UserTypes["basicview"], expectedType) {
+		t.Fatalf("type is %+v, but expected %+v", keyspaceMetadata.UserTypes["basicview"], expectedType)
+	}
+	if flagCassVersion.Major >= 3 {
+		materializedView, found := keyspaceMetadata.MaterializedViews["view_view"]
+		if !found {
+			t.Fatal("failed to find the materialized view in metadata")
+		}
+		if materializedView.BaseTable.Name != "view_table" {
+			t.Fatalf("expected name: %s, materialized view base table name: %s", "view_table", materializedView.BaseTable.Name)
+		}
+	}
 }
 
 // Integration test of the routing key calculation
diff --git a/common_test.go b/common_test.go
index f071453..c7b912f 100644
--- a/common_test.go
+++ b/common_test.go
@@ -184,6 +184,26 @@ func createViews(t *testing.T, session *Session) {
 	}
 }
 
+func createMaterializedViews(t *testing.T, session *Session) {
+	if flagCassVersion.Before(3, 0, 0) {
+		return
+	}
+	if err := session.Query(`CREATE TABLE IF NOT EXISTS gocql_test.view_table (
+		    userid text,
+		    year int,
+		    month int,
+    		    PRIMARY KEY (userid));`).Exec(); err != nil {
+		t.Fatalf("failed to create materialized view with err: %v", err)
+	}
+	if err := session.Query(`CREATE MATERIALIZED VIEW IF NOT EXISTS gocql_test.view_view AS
+		   SELECT year, month, userid
+		   FROM gocql_test.view_table
+		   WHERE year IS NOT NULL AND month IS NOT NULL AND userid IS NOT NULL
+		   PRIMARY KEY (userid, year);`).Exec(); err != nil {
+		t.Fatalf("failed to create materialized view with err: %v", err)
+	}
+}
+
 func createFunctions(t *testing.T, session *Session) {
 	if err := session.Query(`
 		CREATE OR REPLACE FUNCTION gocql_test.avgState ( state tuple<int,bigint>, val int )
diff --git a/conn_test.go b/conn_test.go
index 13605d1..dbb5f9f 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -11,6 +11,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -242,7 +243,7 @@ func TestStartupTimeout(t *testing.T) {
 		t.Fatal("ConnectTimeout is not respected")
 	}
 
-	if !strings.Contains(err.Error(), "no connections were made when creating the session") {
+	if !errors.Is(err, ErrNoConnectionsStarted) {
 		t.Fatalf("Expected to receive no connections error - got '%s'", err)
 	}
 
@@ -519,90 +520,6 @@ func TestSpeculativeExecution(t *testing.T) {
 	}
 }
 
-func TestStreams_Protocol1(t *testing.T) {
-	srv := NewTestServer(t, protoVersion1, context.Background())
-	defer srv.Stop()
-
-	// TODO: these are more like session tests and should instead operate
-	// on a single Conn
-	cluster := testCluster(protoVersion1, srv.Address)
-	cluster.NumConns = 1
-	cluster.ProtoVersion = 1
-
-	db, err := cluster.CreateSession()
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer db.Close()
-
-	var wg sync.WaitGroup
-	for i := 1; i < 128; i++ {
-		// here were just validating that if we send NumStream request we get
-		// a response for every stream and the lengths for the queries are set
-		// correctly.
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			if err := db.Query("void").Exec(); err != nil {
-				t.Error(err)
-			}
-		}()
-	}
-	wg.Wait()
-}
-
-func TestStreams_Protocol3(t *testing.T) {
-	srv := NewTestServer(t, protoVersion3, context.Background())
-	defer srv.Stop()
-
-	// TODO: these are more like session tests and should instead operate
-	// on a single Conn
-	cluster := testCluster(protoVersion3, srv.Address)
-	cluster.NumConns = 1
-	cluster.ProtoVersion = 3
-
-	db, err := cluster.CreateSession()
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer db.Close()
-
-	for i := 1; i < 32768; i++ {
-		// the test server processes each conn synchronously
-		// here were just validating that if we send NumStream request we get
-		// a response for every stream and the lengths for the queries are set
-		// correctly.
-		if err = db.Query("void").Exec(); err != nil {
-			t.Fatal(err)
-		}
-	}
-}
-
-func BenchmarkProtocolV3(b *testing.B) {
-	srv := NewTestServer(b, protoVersion3, context.Background())
-	defer srv.Stop()
-
-	// TODO: these are more like session tests and should instead operate
-	// on a single Conn
-	cluster := NewCluster(srv.Address)
-	cluster.NumConns = 1
-	cluster.ProtoVersion = 3
-
-	db, err := cluster.CreateSession()
-	if err != nil {
-		b.Fatal(err)
-	}
-	defer db.Close()
-
-	b.ResetTimer()
-	b.ReportAllocs()
-	for i := 0; i < b.N; i++ {
-		if err = db.Query("void").Exec(); err != nil {
-			b.Fatal(err)
-		}
-	}
-}
-
 // This tests that the policy connection pool handles SSL correctly
 func TestPolicyConnPoolSSL(t *testing.T) {
 	srv := NewSSLTestServer(t, defaultProto, context.Background())
diff --git a/marshal_test.go b/marshal_test.go
index 5160850..8410018 100644
--- a/marshal_test.go
+++ b/marshal_test.go
@@ -1708,32 +1708,215 @@ func TestMarshalTuple(t *testing.T) {
 		},
 	}
 
-	expectedData := []byte("\x00\x00\x00\x03foo\x00\x00\x00\x03bar")
-	value := []interface{}{"foo", "bar"}
-
-	data, err := Marshal(info, value)
-	if err != nil {
-		t.Errorf("marshalTest: %v", err)
-		return
+	stringToPtr := func(s string) *string { return &s }
+	checkString := func(t *testing.T, exp string, got string) {
+		if got != exp {
+			t.Errorf("expected string to be %v, got %v", exp, got)
+		}
 	}
 
-	if !bytes.Equal(data, expectedData) {
-		t.Errorf("marshalTest: expected %x (%v), got %x (%v)",
-			expectedData, decBigInt(expectedData), data, decBigInt(data))
-		return
+	type tupleStruct struct {
+		A string
+		B *string
+	}
+	var (
+		s1 *string
+		s2 *string
+	)
+
+	testCases := []struct {
+		name       string
+		expected   []byte
+		value      interface{}
+		checkValue interface{}
+		check      func(*testing.T, interface{})
+	}{
+		{
+			name:       "interface-slice:two-strings",
+			expected:   []byte("\x00\x00\x00\x03foo\x00\x00\x00\x03bar"),
+			value:      []interface{}{"foo", "bar"},
+			checkValue: []interface{}{&s1, &s2},
+			check: func(t *testing.T, v interface{}) {
+				checkString(t, "foo", *s1)
+				checkString(t, "bar", *s2)
+			},
+		},
+		{
+			name:       "interface-slice:one-string-one-nil-string",
+			expected:   []byte("\x00\x00\x00\x03foo\xff\xff\xff\xff"),
+			value:      []interface{}{"foo", nil},
+			checkValue: []interface{}{&s1, &s2},
+			check: func(t *testing.T, v interface{}) {
+				checkString(t, "foo", *s1)
+				if s2 != nil {
+					t.Errorf("expected string to be nil, got %v", *s2)
+				}
+			},
+		},
+		{
+			name:     "struct:two-strings",
+			expected: []byte("\x00\x00\x00\x03foo\x00\x00\x00\x03bar"),
+			value: tupleStruct{
+				A: "foo",
+				B: stringToPtr("bar"),
+			},
+			checkValue: &tupleStruct{},
+			check: func(t *testing.T, v interface{}) {
+				got := v.(*tupleStruct)
+				if got.A != "foo" {
+					t.Errorf("expected A string to be %v, got %v", "foo", got.A)
+				}
+				if got.B == nil {
+					t.Errorf("expected B string to be %v, got nil", "bar")
+				}
+				if *got.B != "bar" {
+					t.Errorf("expected B string to be %v, got %v", "bar", got.B)
+				}
+			},
+		},
+		{
+			name:       "struct:one-string-one-nil-string",
+			expected:   []byte("\x00\x00\x00\x03foo\xff\xff\xff\xff"),
+			value:      tupleStruct{A: "foo", B: nil},
+			checkValue: &tupleStruct{},
+			check: func(t *testing.T, v interface{}) {
+				got := v.(*tupleStruct)
+				if got.A != "foo" {
+					t.Errorf("expected A string to be %v, got %v", "foo", got.A)
+				}
+				if *got.B != "" {
+					t.Errorf("expected B string to be empty, got %v", *got.B)
+				}
+			},
+		},
+		{
+			name:     "arrayslice:two-strings",
+			expected: []byte("\x00\x00\x00\x03foo\x00\x00\x00\x03bar"),
+			value: [2]*string{
+				stringToPtr("foo"),
+				stringToPtr("bar"),
+			},
+			checkValue: &[2]*string{},
+			check: func(t *testing.T, v interface{}) {
+				got := v.(*[2]*string)
+				checkString(t, "foo", *(got[0]))
+				checkString(t, "bar", *(got[1]))
+			},
+		},
+		{
+			name:     "arrayslice:one-string-one-nil-string",
+			expected: []byte("\x00\x00\x00\x03foo\xff\xff\xff\xff"),
+			value: [2]*string{
+				stringToPtr("foo"),
+				nil,
+			},
+			checkValue: &[2]*string{},
+			check: func(t *testing.T, v interface{}) {
+				got := v.(*[2]*string)
+				checkString(t, "foo", *(got[0]))
+				checkString(t, "", *(got[1]))
+			},
+		},
 	}
 
-	var s1, s2 string
-	val := []interface{}{&s1, &s2}
-	err = Unmarshal(info, expectedData, val)
-	if err != nil {
-		t.Errorf("unmarshalTest: %v", err)
-		return
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			data, err := Marshal(info, tc.value)
+			if err != nil {
+				t.Errorf("marshalTest: %v", err)
+				return
+			}
+
+			if !bytes.Equal(data, tc.expected) {
+				t.Errorf("marshalTest: expected %x (%v), got %x (%v)",
+					tc.expected, decBigInt(tc.expected), data, decBigInt(data))
+				return
+			}
+
+			err = Unmarshal(info, data, tc.checkValue)
+			if err != nil {
+				t.Errorf("unmarshalTest: %v", err)
+				return
+			}
+
+			tc.check(t, tc.checkValue)
+		})
 	}
+}
 
-	if s1 != "foo" || s2 != "bar" {
-		t.Errorf("unmarshalTest: expected [foo, bar], got [%s, %s]", s1, s2)
+func TestUnmarshalTuple(t *testing.T) {
+	info := TupleTypeInfo{
+		NativeType: NativeType{proto: 3, typ: TypeTuple},
+		Elems: []TypeInfo{
+			NativeType{proto: 3, typ: TypeVarchar},
+			NativeType{proto: 3, typ: TypeVarchar},
+		},
 	}
+
+	// As per the CQL spec, a tuple is a sequence of "bytes" values.
+	// Here we encode a null value (length -1) and the "foo" string (length 3)
+
+	data := []byte("\xff\xff\xff\xff\x00\x00\x00\x03foo")
+
+	t.Run("struct-ptr", func(t *testing.T) {
+		var tmp struct {
+			A *string
+			B *string
+		}
+
+		err := Unmarshal(info, data, &tmp)
+		if err != nil {
+			t.Errorf("unmarshalTest: %v", err)
+			return
+		}
+
+		if *tmp.A != "" || *tmp.B != "foo" {
+			t.Errorf("unmarshalTest: expected [nil, foo], got [%v, %v]", *tmp.A, *tmp.B)
+		}
+	})
+	t.Run("struct-nonptr", func(t *testing.T) {
+		var tmp struct {
+			A string
+			B string
+		}
+
+		err := Unmarshal(info, data, &tmp)
+		if err != nil {
+			t.Errorf("unmarshalTest: %v", err)
+			return
+		}
+
+		if tmp.A != "" || tmp.B != "foo" {
+			t.Errorf("unmarshalTest: expected [nil, foo], got [%v, %v]", tmp.A, tmp.B)
+		}
+	})
+
+	t.Run("array", func(t *testing.T) {
+		var tmp [2]*string
+
+		err := Unmarshal(info, data, &tmp)
+		if err != nil {
+			t.Errorf("unmarshalTest: %v", err)
+			return
+		}
+
+		if *tmp[0] != "" || *tmp[1] != "foo" {
+			t.Errorf("unmarshalTest: expected [nil, foo], got [%v, %v]", *tmp[0], *tmp[1])
+		}
+	})
+	t.Run("array-nonptr", func(t *testing.T) {
+		var tmp [2]string
+
+		err := Unmarshal(info, data, &tmp)
+		if err != nil {
+			t.Errorf("unmarshalTest: %v", err)
+			return
+		}
+
+		if tmp[0] != "" || tmp[1] != "foo" {
+			t.Errorf("unmarshalTest: expected [nil, foo], got [%v, %v]", tmp[0], tmp[1])
+		}
+	})
 }
 
 func TestMarshalNil(t *testing.T) {
diff --git a/metadata_test.go b/metadata_test.go
index 61b2ecf..c874864 100644
--- a/metadata_test.go
+++ b/metadata_test.go
@@ -94,7 +94,7 @@ func TestCompileMetadata(t *testing.T) {
 		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "schema_version", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UUIDType"},
 		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "tokens", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.SetType(org.apache.cassandra.db.marshal.UTF8Type)"},
 	}
-	compileMetadata(1, keyspace, tables, columns, nil, nil, nil)
+	compileMetadata(1, keyspace, tables, columns, nil, nil, nil, nil)
 	assertKeyspaceMetadata(
 		t,
 		keyspace,
@@ -375,7 +375,7 @@ func TestCompileMetadata(t *testing.T) {
 			Validator: "org.apache.cassandra.db.marshal.UTF8Type",
 		},
 	}
-	compileMetadata(2, keyspace, tables, columns, nil, nil, nil)
+	compileMetadata(2, keyspace, tables, columns, nil, nil, nil, nil)
 	assertKeyspaceMetadata(
 		t,
 		keyspace,