Merge branch 'main' into CBG-4584

Author: bbrks

base/bucket_n1ql.go

@@ -20,7 +20,8 @@ const PrimaryIndexName = "#primary"
 
 // IndexOptions used to build the 'with' clause
 type N1qlIndexOptions struct {
-	NumReplica      uint `json:"num_replica,omitempty"`          // Number of replicas
-	IndexTombstones bool `json:"retain_deleted_xattr,omitempty"` // Whether system xattrs on tombstones should be indexed
-	DeferBuild      bool `json:"defer_build,omitempty"`          // Whether to defer initial build of index (requires a subsequent BUILD INDEX invocation)
+	NumReplica      uint    `json:"num_replica,omitempty"`          // Number of replicas
+	IndexTombstones bool    `json:"retain_deleted_xattr,omitempty"` // Whether system xattrs on tombstones should be indexed
+	DeferBuild      bool    `json:"defer_build,omitempty"`          // Whether to defer initial build of index (requires a subsequent BUILD INDEX invocation)
+	NumPartitions   *uint32 `json:"num_partition,omitempty"`        // The number of partitions to use for the index. 1 will be a non-partitioned index.
 }

base/collection.go

@@ -239,7 +239,7 @@ func (b *GocbV2Bucket) IsSupported(feature sgbucket.BucketStoreFeature) bool {
 		// Available on all supported server versions
 		return true
 	case sgbucket.BucketStoreFeatureN1ql:
-		agent, err := b.getGoCBAgent()
+		agent, err := b.GetGoCBAgent()
 		if err != nil {
 			return false
 		}
@@ -295,7 +295,7 @@ func (b *GocbV2Bucket) StartDCPFeed(ctx context.Context, args sgbucket.FeedArgum
 
 func (b *GocbV2Bucket) GetStatsVbSeqno(maxVbno uint16, useAbsHighSeqNo bool) (uuids map[uint16]uint64, highSeqnos map[uint16]uint64, seqErr error) {
 
-	agent, agentErr := b.getGoCBAgent()
+	agent, agentErr := b.GetGoCBAgent()
 	if agentErr != nil {
 		return nil, nil, agentErr
 	}
@@ -353,7 +353,7 @@ func (b *GocbV2Bucket) GetMaxVbno() (uint16, error) {
 }
 
 func (b *GocbV2Bucket) getConfigSnapshot() (*gocbcore.ConfigSnapshot, error) {
-	agent, err := b.getGoCBAgent()
+	agent, err := b.GetGoCBAgent()
 	if err != nil {
 		return nil, fmt.Errorf("no gocbcore.Agent: %w", err)
 	}
@@ -448,7 +448,7 @@ func (b *GocbV2Bucket) BucketItemCount(ctx context.Context) (itemCount int, err
 }
 
 func (b *GocbV2Bucket) MgmtEps() (url []string, err error) {
-	agent, err := b.getGoCBAgent()
+	agent, err := b.GetGoCBAgent()
 	if err != nil {
 		return url, err
 	}
@@ -460,7 +460,7 @@ func (b *GocbV2Bucket) MgmtEps() (url []string, err error) {
 }
 
 func (b *GocbV2Bucket) QueryEpsCount() (int, error) {
-	agent, err := b.getGoCBAgent()
+	agent, err := b.GetGoCBAgent()
 	if err != nil {
 		return 0, err
 	}
@@ -479,7 +479,7 @@ func (b *GocbV2Bucket) MaxTTL(ctx context.Context) (int, error) {
 }
 
 func (b *GocbV2Bucket) HttpClient(ctx context.Context) *http.Client {
-	agent, err := b.getGoCBAgent()
+	agent, err := b.GetGoCBAgent()
 	if err != nil {
 		WarnfCtx(ctx, "Unable to obtain gocbcore.Agent while retrieving httpClient:%v", err)
 		return nil
@@ -527,7 +527,7 @@ func (b *GocbV2Bucket) releaseKvOp() {
 }
 
 // GetGoCBAgent returns the underlying agent from gocbcore
-func (b *GocbV2Bucket) getGoCBAgent() (*gocbcore.Agent, error) {
+func (b *GocbV2Bucket) GetGoCBAgent() (*gocbcore.Agent, error) {
 	return b.bucket.Internal().IORouter()
 }
 

base/collection_gocb.go

@@ -415,7 +415,7 @@ func (c *Collection) isRecoverableWriteError(err error) bool {
 // current use cases (on-demand import).  If there's a need for expiry as part of normal get, this shouldn't be
 // used - an enhanced version of Get() should be implemented to avoid two ops
 func (c *Collection) GetExpiry(ctx context.Context, k string) (expiry uint32, getMetaError error) {
-	agent, err := c.Bucket.getGoCBAgent()
+	agent, err := c.Bucket.GetGoCBAgent()
 	if err != nil {
 		WarnfCtx(ctx, "Unable to obtain gocbcore.Agent while retrieving expiry:%v", err)
 		return 0, err
@@ -569,7 +569,7 @@ func (c *Collection) setCollectionID() error {
 		c.kvCollectionID = DefaultCollectionID
 		return nil
 	}
-	agent, err := c.Bucket.getGoCBAgent()
+	agent, err := c.Bucket.GetGoCBAgent()
 	if err != nil {
 		return err
 	}

base/collection_n1ql_common.go

@@ -161,11 +161,16 @@ func createIndex(ctx context.Context, store N1QLStore, indexName string, express
 	if filterExpression != "" {
 		filterExpressionStr = " WHERE " + filterExpression
 	}
+	var partitionExpresionStr string
+	if options.NumPartitions != nil && *options.NumPartitions > 1 {
+		partitionExpresionStr = " PARTITION BY HASH(META().id)"
+	}
 
-	createStatement := fmt.Sprintf("CREATE INDEX `%s`%s ON %s(%s)%s", indexName, ifNotExistsStr, store.EscapedKeyspace(), expression, filterExpressionStr)
+	createStatement := fmt.Sprintf("CREATE INDEX `%s`%s ON %s(%s)%s %s", indexName, ifNotExistsStr, store.EscapedKeyspace(), expression, partitionExpresionStr, filterExpressionStr)
 
 	// Replace any KeyspaceQueryToken references in the index expression
 	createStatement = strings.ReplaceAll(createStatement, KeyspaceQueryToken, store.EscapedKeyspace())
+
 	createErr := createIndexFromStatement(ctx, store, indexName, createStatement, options)
 	if IsIndexAlreadyExistsError(createErr) || IsCreateDuplicateIndexError(createErr) {
 		// Pre-7.1 compatibility: Swallow this error like Server does when specifying `IF NOT EXISTS`

base/collection_view.go

@@ -161,7 +161,7 @@ type NoNameDesignDocument struct {
 
 func (b *GocbV2Bucket) putDDocForTombstones(ctx context.Context, ddoc *gocb.DesignDocument) error {
 	username, password, _ := b.Spec.Auth.GetCredentials()
-	agent, err := b.getGoCBAgent()
+	agent, err := b.GetGoCBAgent()
 	if err != nil {
 		return fmt.Errorf("Unable to get handle for bucket router: %v", err)
 	}

base/dcp_client_stream_observer.go

@@ -48,8 +48,10 @@ func (dc *DCPClient) Mutation(mutation gocbcore.DcpMutation) {
 		cas:        mutation.Cas,
 		datatype:   mutation.Datatype,
 		collection: mutation.CollectionID,
-		key:        mutation.Key,
-		value:      mutation.Value,
+
+		// The byte slices must be copied to ensure that memory associated with the underlying memd mutationEvent and Packet are independent and can be released or reused by gocbcore as needed.
+		key:   EfficientBytesClone(mutation.Key),
+		value: EfficientBytesClone(mutation.Value),
 	}
 	dc.workerForVbno(mutation.VbID).Send(dc.ctx, e)
 }
@@ -69,8 +71,10 @@ func (dc *DCPClient) Deletion(deletion gocbcore.DcpDeletion) {
 		revNo:      deletion.RevNo,
 		datatype:   deletion.Datatype,
 		collection: deletion.CollectionID,
-		key:        deletion.Key,
-		value:      deletion.Value,
+
+		// The byte slices must be copied to ensure that memory associated with the underlying memd mutationEvent and Packet are independent and can be released or reused by gocbcore as needed.
+		key:   EfficientBytesClone(deletion.Key),
+		value: EfficientBytesClone(deletion.Value),
 	}
 	dc.workerForVbno(deletion.VbID).Send(dc.ctx, e)
 

base/dcp_common.go

@@ -522,15 +522,13 @@ func dcpKeyFilter(key []byte, metaKeys *MetadataKeys) bool {
 }
 
 // Makes a feedEvent that can be passed to a FeedEventCallbackFunc implementation
+// The byte slices must be copied to ensure that memory associated with the memd mutationEvent and Packet are independent and can be released or reused by gocbcore as needed.
 func makeFeedEvent(key []byte, value []byte, dataType uint8, cas uint64, expiry uint32, vbNo uint16, collectionID uint32, opcode sgbucket.FeedOpcode) sgbucket.FeedEvent {
 
-	// not currently doing rq.Extras handling (as in gocouchbase/upr_feed, makeUprEvent) as SG doesn't use
-	// expiry/flags information, and snapshot handling is done by cbdatasource and sent as
-	// SnapshotStart, SnapshotEnd
 	event := sgbucket.FeedEvent{
 		Opcode:       opcode,
-		Key:          key,
-		Value:        value,
+		Key:          EfficientBytesClone(key),
+		Value:        EfficientBytesClone(value),
 		CollectionID: collectionID,
 		DataType:     dataType,
 		Cas:          cas,

base/dcp_common_test.go

@@ -17,6 +17,7 @@ import (
 	"testing"
 
 	"github.com/couchbase/cbgt"
+	sgbucket "github.com/couchbase/sg-bucket"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -83,3 +84,26 @@ func TestDCPNameLength(t *testing.T) {
 		})
 	}
 }
+
+// TestFeedEventByteSliceCopy( ensures that the byte slices in the FeedEvent are copies and not the original ones - CBG-4540
+func TestFeedEventByteSliceCopy(t *testing.T) {
+	const (
+		keyData   = "key"
+		valueData = "value"
+	)
+	keySlice := []byte(keyData)
+	valueSlice := []byte(valueData)
+	e := makeFeedEvent(keySlice, valueSlice, 0, 0, 0, 0, 0, sgbucket.FeedOpMutation)
+	require.Equal(t, keyData, string(e.Key))
+	require.Equal(t, valueData, string(e.Value))
+	require.Equal(t, keyData, string(keySlice))
+	require.Equal(t, valueData, string(valueSlice))
+
+	// mutate the originals and ensure the FeedEvent byte slices didn't change with it
+	keySlice[0] = 'x'
+	valueSlice[0] = 'x'
+	assert.Equal(t, keyData, string(e.Key))
+	assert.Equal(t, valueData, string(e.Value))
+	assert.NotEqual(t, keyData, string(keySlice))
+	assert.NotEqual(t, valueData, string(valueSlice))
+}

base/error.go

@@ -78,6 +78,9 @@ var (
 	// ErrSkippedSequencesMissing is returned when attempting to remove a sequence range form the skipped sequence list and at least one sequence in that range is not present
 	ErrSkippedSequencesMissing = &sgError{"Sequence range has sequences that aren't present in skipped list"}
 
+	// ErrMaxSequenceReleasedExceeded is returned when the maximum number of sequences to be released as part of nextSequenceGreaterThan is exceeded
+	ErrMaxSequenceReleasedExceeded = &sgError{"Maximum number of sequences to release to catch up with document sequence exceeded"}
+
 	// ErrInvalidJSON is returned when the JSON being unmarshalled cannot be parsed.
 	ErrInvalidJSON = HTTPErrorf(http.StatusBadRequest, "Invalid JSON")
 )

base/stats.go

@@ -645,6 +645,8 @@ type DatabaseStats struct {
 	SequenceReleasedCount *SgwIntStat `json:"sequence_released_count"`
 	// The total number of sequences reserved by Sync Gateway.
 	SequenceReservedCount *SgwIntStat `json:"sequence_reserved_count"`
+	// The total number of corrupt sequences above the MaxSequencesToRelease threshold seen at the sequence allocator
+	CorruptSequenceCount *SgwIntStat `json:"corrupt_sequence_count"`
 	// The total number of warnings relating to the channel name size.
 	WarnChannelNameSizeCount *SgwIntStat `json:"warn_channel_name_size_count"`
 	// The total number of warnings relating to the channel count exceeding the channel count threshold.
@@ -1754,6 +1756,10 @@ func (d *DbStats) initDatabaseStats() error {
 	if err != nil {
 		return err
 	}
+	resUtil.CorruptSequenceCount, err = NewIntStat(SubsystemDatabaseKey, "corrupt_sequence_count", StatUnitNoUnits, CorruptSequenceCountDesc, StatAddedVersion3dot2dot4, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	if err != nil {
+		return err
+	}
 	resUtil.WarnChannelNameSizeCount, err = NewIntStat(SubsystemDatabaseKey, "warn_channel_name_size_count", StatUnitNoUnits, WarnChannelNameSizeCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
@@ -1839,6 +1845,7 @@ func (d *DbStats) unregisterDatabaseStats() {
 	prometheus.Unregister(d.DatabaseStats.SequenceIncrCount)
 	prometheus.Unregister(d.DatabaseStats.SequenceReleasedCount)
 	prometheus.Unregister(d.DatabaseStats.SequenceReservedCount)
+	prometheus.Unregister(d.DatabaseStats.CorruptSequenceCount)
 	prometheus.Unregister(d.DatabaseStats.WarnChannelNameSizeCount)
 	prometheus.Unregister(d.DatabaseStats.WarnChannelsPerDocCount)
 	prometheus.Unregister(d.DatabaseStats.WarnGrantsPerDocCount)