Merge branch 'scylladb:master' into invalid-number-of-shards

Author: nikagra

cassandra_test.go

@@ -664,21 +664,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 {

cluster.go

@@ -206,6 +206,11 @@ type ClusterConfig struct {
 	// Also, it describes a maximum number of connections at the same time.
 	// Default: 2
 	NumConns int
+	// The gocql driver may hold excess shard connections to reuse them when existing connections are dropped.
+	// This configuration variable defines the limit for such excess connections. Once the limit is reached,
+	// gocql starts dropping any additional excess connections.
+	// The limit is computed as `MaxExcessShardConnectionsRate` * <number_of_shards>.
+	MaxExcessShardConnectionsRate float32
 	// Maximum cache size for prepared statements globally for gocql.
 	// Default: 1000
 	MaxPreparedStmts int
@@ -359,33 +364,34 @@ type Dialer interface {
 // the same host, and will not mark the node being down or up from events.
 func NewCluster(hosts ...string) *ClusterConfig {
 	cfg := &ClusterConfig{
-		Hosts:                        hosts,
-		CQLVersion:                   "3.0.0",
-		Timeout:                      11 * time.Second,
-		ConnectTimeout:               60 * time.Second,
-		ReadTimeout:                  11 * time.Second,
-		WriteTimeout:                 11 * time.Second,
-		Port:                         9042,
-		NumConns:                     2,
-		Consistency:                  Quorum,
-		MaxPreparedStmts:             defaultMaxPreparedStmts,
-		MaxRoutingKeyInfo:            1000,
-		PageSize:                     5000,
-		DefaultTimestamp:             true,
-		DriverName:                   defaultDriverName,
-		DriverVersion:                defaultDriverVersion,
-		MaxWaitSchemaAgreement:       60 * time.Second,
-		ReconnectInterval:            60 * time.Second,
-		ConvictionPolicy:             &SimpleConvictionPolicy{},
-		ReconnectionPolicy:           &ConstantReconnectionPolicy{MaxRetries: 3, Interval: 1 * time.Second},
-		InitialReconnectionPolicy:    &NoReconnectionPolicy{},
-		SocketKeepalive:              15 * time.Second,
-		WriteCoalesceWaitTime:        200 * time.Microsecond,
-		MetadataSchemaRequestTimeout: 60 * time.Second,
-		DisableSkipMetadata:          true,
-		WarningsHandlerBuilder:       DefaultWarningHandlerBuilder,
-		Logger:                       &defaultLogger{},
-		DNSResolver:                  defaultDnsResolver,
+		Hosts:                         hosts,
+		CQLVersion:                    "3.0.0",
+		Timeout:                       11 * time.Second,
+		ConnectTimeout:                60 * time.Second,
+		ReadTimeout:                   11 * time.Second,
+		WriteTimeout:                  11 * time.Second,
+		Port:                          9042,
+		MaxExcessShardConnectionsRate: 2,
+		NumConns:                      2,
+		Consistency:                   Quorum,
+		MaxPreparedStmts:              defaultMaxPreparedStmts,
+		MaxRoutingKeyInfo:             1000,
+		PageSize:                      5000,
+		DefaultTimestamp:              true,
+		DriverName:                    defaultDriverName,
+		DriverVersion:                 defaultDriverVersion,
+		MaxWaitSchemaAgreement:        60 * time.Second,
+		ReconnectInterval:             60 * time.Second,
+		ConvictionPolicy:              &SimpleConvictionPolicy{},
+		ReconnectionPolicy:            &ConstantReconnectionPolicy{MaxRetries: 3, Interval: 1 * time.Second},
+		InitialReconnectionPolicy:     &NoReconnectionPolicy{},
+		SocketKeepalive:               15 * time.Second,
+		WriteCoalesceWaitTime:         200 * time.Microsecond,
+		MetadataSchemaRequestTimeout:  60 * time.Second,
+		DisableSkipMetadata:           true,
+		WarningsHandlerBuilder:        DefaultWarningHandlerBuilder,
+		Logger:                        &defaultLogger{},
+		DNSResolver:                   defaultDnsResolver,
 	}
 
 	return cfg
@@ -545,6 +551,10 @@ func (cfg *ClusterConfig) Validate() error {
 		return fmt.Errorf("DNSResolver is empty")
 	}
 
+	if cfg.MaxExcessShardConnectionsRate < 0 {
+		return fmt.Errorf("MaxExcessShardConnectionsRate should be positive number or zero")
+	}
+
 	return cfg.ValidateAndInitSSL()
 }
 

frame.go

@@ -32,6 +32,7 @@ import (
 	"io/ioutil"
 	"net"
 	"runtime"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -913,6 +914,22 @@ func (f *framer) readTypeInfo() TypeInfo {
 		collection.Elem = f.readTypeInfo()
 
 		return collection
+	case TypeCustom:
+		if strings.HasPrefix(simple.custom, "org.apache.cassandra.db.marshal.VectorType") {
+			spec := strings.TrimPrefix(simple.custom, "org.apache.cassandra.db.marshal.VectorType")
+			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

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/google/go-cmp v0.7.0
 	github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed
 	github.com/klauspost/compress v1.18.1
-	golang.org/x/net v0.46.0
+	golang.org/x/net v0.47.0
 	gopkg.in/inf.v0 v0.9.1
 	sigs.k8s.io/yaml v1.6.0
 )

go.sum

@@ -30,6 +30,8 @@ go.yaml.in/yaml/v3 v3.0.3 h1:bXOww4E/J3f66rav3pX3m8w6jDE4knZjGOw8b5Y6iNE=
 go.yaml.in/yaml/v3 v3.0.3/go.mod h1:tBHosrYAkRZjRAOREWbDnBXUf08JOwYq++0QNwQiWzI=
 golang.org/x/net v0.46.0 h1:giFlY12I07fugqwPuWJi68oOnpfqFnJIJzaIIm2JVV4=
 golang.org/x/net v0.46.0/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210=
+golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY=
+golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=