CBG-4548 allow creation of partitioned indexes (#7428)

Co-authored-by: Ben Brooks <[email protected]>

Author: torcolvin

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)
 	}

db/background_mgr_resync_dcp.go

@@ -389,6 +389,7 @@ func initializePrincipalDocsIndex(ctx context.Context, db *Database) error {
 		NumReplicas:                db.Options.NumIndexReplicas,
 		MetadataIndexes:            IndexesPrincipalOnly,
 		UseXattrs:                  db.UseXattrs(),
+		NumPartitions:              db.numIndexPartitions(),
 	}
 
 	return InitializeIndexes(ctx, n1qlStore, options)

db/database.go

@@ -185,6 +185,7 @@ type DatabaseContextOptions struct {
 	MaxConcurrentChangesBatches   *int              // Maximum number of changes batches to process concurrently per replication
 	MaxConcurrentRevs             *int              // Maximum number of revs to process concurrently per replication
 	NumIndexReplicas              uint              // Number of replicas for GSI indexes
+	NumIndexPartitions            *uint32           // Number of partitions for GSI indexes, if not set will default to 1
 	ImportVersion                 uint64            // Version included in import DCP checkpoints, incremented when collections added to db
 }
 
@@ -1915,6 +1916,14 @@ func (context *DatabaseContext) UseXattrs() bool {
 	return context.Options.EnableXattr
 }
 
+// numIndexPartitions returns the number of index partitions to use for the database's indexes.
+func (context *DatabaseContext) numIndexPartitions() uint32 {
+	if context.Options.NumIndexPartitions != nil {
+		return *context.Options.NumIndexPartitions
+	}
+	return DefaultNumIndexPartitions
+}
+
 func (context *DatabaseContext) UseViews() bool {
 	return context.Options.UseViews
 }

db/database_collection.go

@@ -265,6 +265,11 @@ func (c *DatabaseCollection) UseXattrs() bool {
 	return c.dbCtx.Options.EnableXattr
 }
 
+// numIndexPartitions returns the number of partitions for the collection's indexes. This is controlled at a database level.
+func (c *DatabaseCollection) numIndexPartitions() uint32 {
+	return c.dbCtx.numIndexPartitions()
+}
+
 // User will return the user object.
 func (c *DatabaseCollectionWithUser) User() auth.User {
 	return c.user

db/indexes.go

@@ -21,16 +21,19 @@ import (
 )
 
 const (
-	indexNameFormat   = "sg_%s_%s%d"    // Name, xattrs, version.  e.g. "sg_channels_x1"
-	syncRelativeToken = "$relativesync" // Relative sync token (no keyspace), used to swap between xattr/non-xattr handling in n1ql statements
-	syncToken         = "$sync"         // Sync token, used to swap between xattr/non-xattr handling in n1ql statements
-	indexToken        = "$idx"          // Index token, used to hint which index should be used for the query
+	indexNameFormat              = "sg_%s_%s%d"    // Name, xattrs, version.  e.g. "sg_channels_x1"
+	partitionableIndexNameFormat = "%s_p%d"        // indexName, numPartitions
+	syncRelativeToken            = "$relativesync" // Relative sync token (no keyspace), used to swap between xattr/non-xattr handling in n1ql statements
+	syncToken                    = "$sync"         // Sync token, used to swap between xattr/non-xattr handling in n1ql statements
+	indexToken                   = "$idx"          // Index token, used to hint which index should be used for the query
 
 	// N1ql-encoded wildcard expression matching the '_sync:' prefix used for all sync gateway's system documents.
 	// Need to escape the underscore in '_sync' to prevent it being treated as a N1QL wildcard
 	SyncDocWildcard  = `\\_sync:%`
 	SyncUserWildcard = `\\_sync:user:%`
 	SyncRoleWildcard = `\\_sync:role:%`
+
+	DefaultNumIndexPartitions uint32 = 1
 )
 
 // Index and query definitions use syncToken ($sync) to represent the location of sync gateway's metadata.
@@ -175,6 +178,10 @@ var (
 			"ORDER BY [op.name, LEAST($sync.`sequence`, op.val.seq),IFMISSING(op.val.rev,null),IFMISSING(op.val.del,null)] " +
 			"LIMIT 1",
 	}
+	partitionableIndexes = map[SGIndexType]bool{
+		IndexAllDocs:  true,
+		IndexChannels: true,
+	}
 )
 
 var sgIndexes map[SGIndexType]SGIndex
@@ -191,6 +198,7 @@ func init() {
 			filterExpression: indexFilterExpressions[i],
 			flags:            indexFlags[i],
 			creationMode:     indexCreationModes[i],
+			partitionable:    partitionableIndexes[i],
 		}
 		// If a readiness query is specified for this index, mark the index as required and add to SGIndex
 		readinessQuery, ok := readinessQueries[i]
@@ -214,18 +222,23 @@ type SGIndex struct {
 	readinessQuery   string            // Query used to determine view readiness
 	flags            SGIndexFlags      // Additional index options
 	creationMode     indexCreationMode // Signal when to create indexes
+	partitionable    bool              // Whether the index is partitionable
 }
 
-func (i *SGIndex) fullIndexName(useXattrs bool) string {
-	return i.indexNameForVersion(i.version, useXattrs)
+func (i *SGIndex) fullIndexName(useXattrs bool, numPartitions uint32) string {
+	return i.indexNameForVersion(i.version, useXattrs, numPartitions)
 }
 
-func (i *SGIndex) indexNameForVersion(version int, useXattrs bool) string {
+func (i *SGIndex) indexNameForVersion(version int, useXattrs bool, numPartitions uint32) string {
 	xattrsToken := ""
 	if useXattrs {
 		xattrsToken = "x"
 	}
-	return fmt.Sprintf(indexNameFormat, i.simpleName, xattrsToken, version)
+	indexName := fmt.Sprintf(indexNameFormat, i.simpleName, xattrsToken, version)
+	if i.partitionable && numPartitions > 1 {
+		indexName = fmt.Sprintf(partitionableIndexNameFormat, indexName, numPartitions)
+	}
+	return indexName
 }
 
 // Tombstone indexing is required for indexes that need to index the _sync xattrs even when the document
@@ -281,8 +294,10 @@ func (i *SGIndex) shouldCreate(options InitializeIndexOptions) bool {
 // Creates index associated with specified SGIndex if not already present.  Always defers build - a subsequent BUILD INDEX
 // will need to be invoked for any created indexes.
 func (i *SGIndex) createIfNeeded(ctx context.Context, bucket base.N1QLStore, options InitializeIndexOptions) error {
-
-	indexName := i.fullIndexName(options.UseXattrs)
+	if options.NumPartitions < 1 {
+		return fmt.Errorf("Invalid number of partitions specified for index %s: %d, needs to be greater than 0", i.simpleName, options.NumPartitions)
+	}
+	indexName := i.fullIndexName(options.UseXattrs, options.NumPartitions)
 
 	// Create index
 	base.InfofCtx(ctx, base.KeyQuery, "Creating index %s if it doesn't already exist...", indexName)
@@ -294,6 +309,9 @@ func (i *SGIndex) createIfNeeded(ctx context.Context, bucket base.N1QLStore, opt
 		NumReplica:      options.NumReplicas,
 		IndexTombstones: i.shouldIndexTombstones(options.UseXattrs),
 	}
+	if i.partitionable && options.NumPartitions > 1 {
+		n1qlOptions.NumPartitions = &options.NumPartitions
+	}
 
 	// Initial retry 1 seconds, max wait 30s, waits up to 10m
 	sleeper := base.CreateMaxDoublingSleeperFunc(20, 1000, 30000)
@@ -344,11 +362,14 @@ type InitializeIndexOptions struct {
 	MetadataIndexes            CollectionIndexesType           // indicate which indexes to create
 	Serverless                 bool                            // if true, create indexes for serverless
 	UseXattrs                  bool                            // if true, create indexes on xattrs, otherwise, use inline sync data
+	NumPartitions              uint32                          // number of partitions to use for the index
 }
 
 // Initializes Sync Gateway indexes for datastore.  Creates required indexes if not found, then waits for index readiness.
 func InitializeIndexes(ctx context.Context, n1QLStore base.N1QLStore, options InitializeIndexOptions) error {
-
+	if options.NumPartitions < 1 {
+		return fmt.Errorf("Invalid number of partitions specified: %d, needs to be greater than 0", options.NumPartitions)
+	}
 	base.InfofCtx(ctx, base.KeyAll, "Initializing indexes with numReplicas: %d...", options.NumReplicas)
 
 	// Create any indexes that aren't present
@@ -360,7 +381,7 @@ func InitializeIndexes(ctx context.Context, n1QLStore base.N1QLStore, options In
 			continue
 		}
 
-		fullIndexName := sgIndex.fullIndexName(options.UseXattrs)
+		fullIndexName := sgIndex.fullIndexName(options.UseXattrs, options.NumPartitions)
 
 		err := sgIndex.createIfNeeded(ctx, n1QLStore, options)
 		if err != nil {
@@ -371,7 +392,6 @@ func InitializeIndexes(ctx context.Context, n1QLStore base.N1QLStore, options In
 
 		fullIndexNames = append(fullIndexNames, fullIndexName)
 	}
-
 	// Issue BUILD INDEX for any deferred indexes.
 	if len(fullIndexNames) > 0 {
 		buildErr := base.BuildDeferredIndexes(ctx, n1QLStore, fullIndexNames)
@@ -391,7 +411,7 @@ func waitForIndexes(ctx context.Context, bucket base.N1QLStore, options Initiali
 	var indexes []string
 
 	for _, sgIndex := range sgIndexes {
-		fullIndexName := sgIndex.fullIndexName(options.UseXattrs)
+		fullIndexName := sgIndex.fullIndexName(options.UseXattrs, options.NumPartitions)
 		if !sgIndex.shouldCreate(options) {
 			continue
 		}
@@ -421,17 +441,18 @@ func removeObsoleteIndexes(ctx context.Context, bucket base.N1QLStore, previewOn
 
 	// Build set of candidates for cleanup
 	removalCandidates := make([]string, 0)
+	numPartitions := DefaultNumIndexPartitions // obsolete indexes is always partitions 1, greater than 1 needs to be removed manually
 	for _, sgIndex := range indexMap {
 		// Current version, opposite xattr setting
-		removalCandidates = append(removalCandidates, sgIndex.fullIndexName(!useXattrs))
+		removalCandidates = append(removalCandidates, sgIndex.fullIndexName(!useXattrs, numPartitions))
 		// If using views we can remove current version for xattr setting too
 		if useViews {
-			removalCandidates = append(removalCandidates, sgIndex.fullIndexName(useXattrs))
+			removalCandidates = append(removalCandidates, sgIndex.fullIndexName(useXattrs, numPartitions))
 		}
 		// Older versions, both xattr and non-xattr
 		for _, prevVersion := range sgIndex.previousVersions {
-			removalCandidates = append(removalCandidates, sgIndex.indexNameForVersion(prevVersion, true))
-			removalCandidates = append(removalCandidates, sgIndex.indexNameForVersion(prevVersion, false))
+			removalCandidates = append(removalCandidates, sgIndex.indexNameForVersion(prevVersion, true, numPartitions))
+			removalCandidates = append(removalCandidates, sgIndex.indexNameForVersion(prevVersion, false, numPartitions))
 		}
 	}
 
@@ -497,8 +518,8 @@ func replaceSyncTokensQuery(statement string, useXattrs bool) string {
 }
 
 // Replace index tokens ($idx) in the provided createIndex statement with the appropriate token, depending on whether xattrs should be used.
-func replaceIndexTokensQuery(statement string, idx SGIndex, useXattrs bool) string {
-	return strings.Replace(statement, indexToken, idx.fullIndexName(useXattrs), -1)
+func replaceIndexTokensQuery(statement string, idx SGIndex, useXattrs bool, numPartitions uint32) string {
+	return strings.Replace(statement, indexToken, idx.fullIndexName(useXattrs, numPartitions), -1)
 }
 
 func copySGIndexes(inputMap map[SGIndexType]SGIndex) map[SGIndexType]SGIndex {
@@ -521,7 +542,7 @@ func GetIndexesName(options InitializeIndexOptions) []string {
 			continue
 		}
 		if sgIndex.shouldCreate(options) {
-			indexesName = append(indexesName, sgIndex.fullIndexName(options.UseXattrs))
+			indexesName = append(indexesName, sgIndex.fullIndexName(options.UseXattrs, options.NumPartitions))
 		}
 	}
 	return indexesName