CBG-2838: Rely on index 'IF NOT EXISTS' and idempotent deferred index builds fo… (#7161)

Author: bbrks

base/cluster_n1ql.go

@@ -30,11 +30,12 @@ var _ N1QLStore = &ClusterOnlyN1QLStore{}
 // ClusterOnlyN1QLStore instances.  Anticipates future refactoring of N1QLStore to differentiate between
 // collection-scoped and non-collection-scoped operations.
 type ClusterOnlyN1QLStore struct {
-	cluster             *gocb.Cluster
-	bucketName          string // User to build keyspace for query when not otherwise set
-	scopeName           string // Used to build keyspace for query when not otherwise set
-	collectionName      string // Used to build keyspace for query when not otherwise set
-	supportsCollections bool
+	cluster                  *gocb.Cluster
+	bucketName               string // User to build keyspace for query when not otherwise set
+	scopeName                string // Used to build keyspace for query when not otherwise set
+	collectionName           string // Used to build keyspace for query when not otherwise set
+	supportsCollections      bool
+	supportsIfNotExistsInDDL bool // 7.1.0+ MB-38737
 }
 
 func NewClusterOnlyN1QLStore(cluster *gocb.Cluster, bucketName, scopeName, collectionName string) (*ClusterOnlyN1QLStore, error) {
@@ -51,11 +52,25 @@ func NewClusterOnlyN1QLStore(cluster *gocb.Cluster, bucketName, scopeName, colle
 		return nil, err
 	}
 	clusterOnlyn1qlStore.supportsCollections = isMinimumVersion(uint64(major), uint64(minor), 7, 0)
+	clusterOnlyn1qlStore.supportsIfNotExistsInDDL = isMinimumVersion(uint64(major), uint64(minor), 7, 1)
 
 	return clusterOnlyn1qlStore, nil
 
 }
 
+func (cl *ClusterOnlyN1QLStore) IsSupported(feature sgbucket.BucketStoreFeature) bool {
+	switch feature {
+	case sgbucket.BucketStoreFeatureN1ql:
+		return true
+	case sgbucket.BucketStoreFeatureCollections:
+		return cl.supportsCollections
+	case sgbucket.BucketStoreFeatureN1qlIfNotExistsDDL:
+		return cl.supportsIfNotExistsInDDL
+	default:
+		return false
+	}
+}
+
 func (cl *ClusterOnlyN1QLStore) GetName() string {
 	return cl.bucketName
 }
@@ -72,6 +87,10 @@ func (cl *ClusterOnlyN1QLStore) CreateIndex(ctx context.Context, indexName strin
 	return CreateIndex(ctx, cl, indexName, expression, filterExpression, options)
 }
 
+func (cl *ClusterOnlyN1QLStore) CreateIndexIfNotExists(ctx context.Context, indexName string, expression string, filterExpression string, options *N1qlIndexOptions) error {
+	return CreateIndexIfNotExists(ctx, cl, indexName, expression, filterExpression, options)
+}
+
 func (cl *ClusterOnlyN1QLStore) CreatePrimaryIndex(ctx context.Context, indexName string, options *N1qlIndexOptions) error {
 	return CreatePrimaryIndex(ctx, cl, indexName, options)
 }

base/collection.go

@@ -244,6 +244,8 @@ func (b *GocbV2Bucket) IsSupported(feature sgbucket.BucketStoreFeature) bool {
 			return false
 		}
 		return len(agent.N1qlEps()) > 0
+	case sgbucket.BucketStoreFeatureN1qlIfNotExistsDDL:
+		return isMinimumVersion(b.clusterCompatMajorVersion, b.clusterCompatMinorVersion, 7, 1)
 	// added in Couchbase Server 6.6
 	case sgbucket.BucketStoreFeatureCreateDeletedWithXattr:
 		status, err := b.bucket.Internal().CapabilityStatus(gocb.CapabilityCreateAsDeleted)

base/collection_n1ql.go

@@ -133,6 +133,10 @@ func (c *Collection) CreateIndex(ctx context.Context, indexName string, expressi
 	return CreateIndex(ctx, c, indexName, expression, filterExpression, options)
 }
 
+func (c *Collection) CreateIndexIfNotExists(ctx context.Context, indexName string, expression string, filterExpression string, options *N1qlIndexOptions) error {
+	return CreateIndexIfNotExists(ctx, c, indexName, expression, filterExpression, options)
+}
+
 func (c *Collection) CreatePrimaryIndex(ctx context.Context, indexName string, options *N1qlIndexOptions) error {
 	return CreatePrimaryIndex(ctx, c, indexName, options)
 }

base/collection_n1ql_common.go

@@ -12,6 +12,7 @@ package base
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"slices"
 	"strings"
@@ -51,6 +52,7 @@ type N1QLStore interface {
 	GetName() string
 	BuildDeferredIndexes(ctx context.Context, indexSet []string) error
 	CreateIndex(ctx context.Context, indexName string, expression string, filterExpression string, options *N1qlIndexOptions) error
+	CreateIndexIfNotExists(ctx context.Context, indexName string, expression string, filterExpression string, options *N1qlIndexOptions) error
 	CreatePrimaryIndex(ctx context.Context, indexName string, options *N1qlIndexOptions) error
 	DropIndex(ctx context.Context, indexName string) error
 	ExplainQuery(ctx context.Context, statement string, params map[string]interface{}) (plan map[string]interface{}, err error)
@@ -75,6 +77,8 @@ type N1QLStore interface {
 
 	// waitUntilQueryServiceReady waits until the query service is ready to accept requests
 	waitUntilQueryServiceReady(timeout time.Duration) error
+
+	sgbucket.BucketStoreFeatureIsSupported
 }
 
 func ExplainQuery(ctx context.Context, store N1QLStore, statement string, params map[string]interface{}) (plan map[string]interface{}, err error) {
@@ -119,7 +123,7 @@ func (im *indexManager) GetAllIndexes() ([]gocb.QueryIndex, error) {
 	return im.cluster.GetAllIndexes(im.bucketName, opts)
 }
 
-// CreateIndex issues a CREATE INDEX query in the current bucket, using the form:
+// CreateIndex issues a CREATE INDEX query in the N1QLStore keyspace, using the form:
 //
 //	CREATE INDEX indexName ON bucket.Name(expression) WHERE filterExpression WITH options
 //
@@ -128,31 +132,60 @@ func (im *indexManager) GetAllIndexes() ([]gocb.QueryIndex, error) {
 //	  CreateIndex("myIndex", "field1, field2, nested.field", "field1 > 0", N1qlIndexOptions{numReplica:1})
 //	CREATE INDEX myIndex on myBucket(field1, field2, nested.field) WHERE field1 > 0 WITH {"numReplica":1}
 func CreateIndex(ctx context.Context, store N1QLStore, indexName string, expression string, filterExpression string, options *N1qlIndexOptions) error {
-	createStatement := fmt.Sprintf("CREATE INDEX `%s` ON %s(%s)", indexName, store.EscapedKeyspace(), expression)
+	return createIndex(ctx, store, indexName, expression, filterExpression, false, options)
+}
+
+// CreateIndexIfNotExists issues a CREATE INDEX query in the N1QLStore keyspace, using the form:
+//
+//	CREATE INDEX indexName ON bucket.Name(expression) IF NOT EXISTS WHERE filterExpression WITH options
+//
+// Sample usage with resulting statement:
+//
+//	  CreateIndex("myIndex", "field1, field2, nested.field", "field1 > 0", N1qlIndexOptions{numReplica:1})
+//	CREATE INDEX myIndex on myBucket(field1, field2, nested.field) WHERE field1 > 0 WITH {"numReplica":1}
+func CreateIndexIfNotExists(ctx context.Context, store N1QLStore, indexName string, expression string, filterExpression string, options *N1qlIndexOptions) error {
+	return createIndex(ctx, store, indexName, expression, filterExpression, true, options)
+}
+
+// createIndex is a common function for CreateIndex and CreateIndexIfNotExists
+func createIndex(ctx context.Context, store N1QLStore, indexName string, expression string, filterExpression string, ifNotExists bool, options *N1qlIndexOptions) error {
+	var ifNotExistsStr string
+	// Server 7.1+ - we can still safely _not_ use this when it's not available, because we have equivalent error handling inside this function to swallow `ErrAlreadyExists`.
+	// Would still prefer to use it when we can, to guard us against future error string changes, which is why we do both conditionally.
+	if ifNotExists && store.IsSupported(sgbucket.BucketStoreFeatureN1qlIfNotExistsDDL) {
+		ifNotExistsStr = " IF NOT EXISTS"
+	}
 
 	// Add filter expression, when present
+	var filterExpressionStr string
 	if filterExpression != "" {
-		createStatement = fmt.Sprintf("%s WHERE %s", createStatement, filterExpression)
+		filterExpressionStr = " WHERE " + filterExpression
 	}
 
-	// Replace any KeyspaceQueryToken references in the index expression
-	createStatement = strings.Replace(createStatement, KeyspaceQueryToken, store.EscapedKeyspace(), -1)
+	createStatement := fmt.Sprintf("CREATE INDEX `%s`%s ON %s(%s)%s", indexName, ifNotExistsStr, store.EscapedKeyspace(), expression, filterExpressionStr)
 
-	createErr := createIndex(ctx, store, indexName, createStatement, options)
-	if createErr != nil {
-		if strings.Contains(createErr.Error(), "already exists") || strings.Contains(createErr.Error(), "duplicate index name") {
-			return ErrAlreadyExists
+	// 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`
+		if ifNotExists {
+			return nil
 		}
+		return ErrAlreadyExists
 	}
 	return createErr
 }
 
 func CreatePrimaryIndex(ctx context.Context, store N1QLStore, indexName string, options *N1qlIndexOptions) error {
 	createStatement := fmt.Sprintf("CREATE PRIMARY INDEX `%s` ON %s", indexName, store.EscapedKeyspace())
-	return createIndex(ctx, store, indexName, createStatement, options)
+	return createIndexFromStatement(ctx, store, indexName, createStatement, options)
 }
 
-func createIndex(ctx context.Context, store N1QLStore, indexName string, createStatement string, options *N1qlIndexOptions) error {
+// ErrIndexBackgroundRetry is returned when an index creation operation returned an error but just needs to wait for a server-side readiness or retry.
+var ErrIndexBackgroundRetry = errors.New("Indexer error - waiting for server background retry")
+
+func createIndexFromStatement(ctx context.Context, store N1QLStore, indexName string, createStatement string, options *N1qlIndexOptions) error {
 
 	if options != nil {
 		withClause, marshalErr := JSONMarshal(options)
@@ -162,23 +195,16 @@ func createIndex(ctx context.Context, store N1QLStore, indexName string, createS
 		createStatement = fmt.Sprintf(`%s with %s`, createStatement, withClause)
 	}
 
-	DebugfCtx(ctx, KeyQuery, "Attempting to create index using statement: [%s]", UD(createStatement))
+	TracefCtx(ctx, KeyQuery, "Attempting to create index %q using statement: [%s]", indexName, UD(createStatement))
 
 	err := store.executeStatement(createStatement)
 	if err == nil {
 		return nil
 	}
 
-	if IsIndexerRetryIndexError(err) {
-		InfofCtx(ctx, KeyQuery, "Indexer error creating index - waiting for server background retry.  Error:%v", err)
-		// Wait for bucket to be created in background before returning
-		return waitForIndexExistence(ctx, store, indexName, true)
-	}
-
-	if IsCreateDuplicateIndexError(err) {
-		InfofCtx(ctx, KeyQuery, "Duplicate index creation in progress - waiting for index readiness.  Error:%v", err)
-		// Wait for bucket to be created in background before returning
-		return waitForIndexExistence(ctx, store, indexName, true)
+	if IsIndexerRetryIndexError(err) || IsCreateDuplicateIndexError(err) {
+		DebugfCtx(ctx, KeyQuery, "Index %q is already being created on server: %v", indexName, err)
+		return fmt.Errorf("%w: %s", ErrIndexBackgroundRetry, err.Error())
 	}
 
 	return pkgerrors.WithStack(RedactErrorf("Error creating index with statement: %s.  Error: %v", UD(createStatement), err))
@@ -211,56 +237,35 @@ func waitForIndexExistence(ctx context.Context, store N1QLStore, indexName strin
 	return nil
 }
 
-// BuildDeferredIndexes issues a build command for any deferred sync gateway indexes associated with the bucket.
+// BuildDeferredIndexes issues a build command for any deferred sync gateway indexes associated with the N1QLStore keyspace.
 func BuildDeferredIndexes(ctx context.Context, s N1QLStore, indexSet []string) error {
-
 	if len(indexSet) == 0 {
 		return nil
 	}
 
-	// Only build indexes that are in deferred state.  Query system:indexes to validate the provided set of indexes
-	statement := fmt.Sprintf("SELECT indexes.name, indexes.state FROM system:indexes WHERE indexes.keyspace_id = '%s'", s.IndexMetaKeyspaceID())
+	InfofCtx(ctx, KeyQuery, "Building deferred indexes: %v", indexSet)
 
-	if s.IndexMetaBucketID() != "" {
-		statement += fmt.Sprintf("AND indexes.bucket_id = '%s' ", s.IndexMetaBucketID())
-	}
-	if s.IndexMetaScopeID() != "" {
-		statement += fmt.Sprintf("AND indexes.scope_id = '%s' ", s.IndexMetaScopeID())
+	// the provided indexes can be in a state that is not yet ready to take a build command
+	// there's a delay between the time of index creation and when it's actually found in the system:indexes table
+	// this results in buildIndexes returning a not found error for an index that was very recently created
+	worker := func() (shouldRetry bool, err error, value interface{}) {
+		err = buildIndexes(ctx, s, indexSet)
+		if IsIndexNotFoundError(err) {
+			DebugfCtx(ctx, KeyQuery, "Index not found error when building indexes - will retry: %v", err)
+			return true, err, nil
+		}
+		return err != nil, err, nil
 	}
-
-	statement += fmt.Sprintf("AND indexes.name IN [%s]", StringSliceToN1QLArray(indexSet, "'"))
-	// mod: bucket name
-
-	results, err := s.executeQuery(statement)
+	sleeper := CreateDoublingSleeperDurationFunc(500, time.Second*30)
+	err, _ := RetryLoop(ctx, "BuildDeferredIndexes", worker, sleeper)
 	if err != nil {
 		return err
 	}
-	deferredIndexes := make([]string, 0)
-	var indexInfo struct {
-		Name  string `json:"name"`
-		State string `json:"state"`
-	}
-	for results.Next(ctx, &indexInfo) {
-		// If index is deferred (not built), add to set of deferred indexes
-		if indexInfo.State == IndexStateDeferred {
-			deferredIndexes = append(deferredIndexes, indexInfo.Name)
-		}
-	}
-	closeErr := results.Close()
-	if closeErr != nil {
-		return closeErr
-	}
-
-	if len(deferredIndexes) == 0 {
-		return nil
-	}
 
-	InfofCtx(ctx, KeyQuery, "Building deferred indexes: %v", deferredIndexes)
-	buildErr := buildIndexes(ctx, s, deferredIndexes)
-	return buildErr
+	return nil
 }
 
-// BuildIndexes executes a BUILD INDEX statement in the current bucket, using the form:
+// BuildIndexes executes a BUILD INDEX statement in the N1QLStore keyspace, using the form:
 //
 //	BUILD INDEX ON `bucket.Name`(`index1`, `index2`, ...)
 func buildIndexes(ctx context.Context, s N1QLStore, indexNames []string) error {
@@ -351,7 +356,7 @@ func getIndexMetaWithoutRetry(ctx context.Context, store N1QLStore, indexName st
 	return true, indexInfo, nil
 }
 
-// DropIndex drops the specified index from the current bucket.
+// DropIndex drops the specified index from the N1QLStore keyspace.
 func DropIndex(ctx context.Context, store N1QLStore, indexName string) error {
 	statement := fmt.Sprintf("DROP INDEX default:%s.`%s`", store.EscapedKeyspace(), indexName)
 
@@ -369,7 +374,7 @@ func DropIndex(ctx context.Context, store N1QLStore, indexName string) error {
 	return err
 }
 
-// AsN1QLStore tries to return the given DataStore as a N1QLStore, based on underlying buckets.
+// AsN1QLStore tries to return the given DataStore as a N1QLStore.
 func AsN1QLStore(dataStore DataStore) (N1QLStore, bool) {
 
 	switch typedDataStore := dataStore.(type) {
@@ -378,7 +383,7 @@ func AsN1QLStore(dataStore DataStore) (N1QLStore, bool) {
 	case *LeakyDataStore:
 		return typedDataStore, true
 	default:
-		// bail out for unrecognised/unsupported buckets
+		// bail out for unrecognised/unsupported data store types
 		return nil, false
 	}
 }
@@ -389,6 +394,9 @@ func AsN1QLStore(dataStore DataStore) (N1QLStore, bool) {
 //
 // Stuck with doing a string compare to differentiate between 'not found' and other errors
 func IsIndexNotFoundError(err error) bool {
+	if err == nil {
+		return false
+	}
 	return strings.Contains(err.Error(), "not found")
 }
 
@@ -425,6 +433,13 @@ func IsIndexerRetryBuildError(err error) bool {
 	return false
 }
 
+func IsIndexAlreadyExistsError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(err.Error(), "already exists")
+}
+
 // Check for transient indexer errors (can be retried)
 func isTransientIndexerError(err error) bool {
 	if err == nil {
@@ -594,7 +609,7 @@ func WaitForIndexesOnline(ctx context.Context, keyspace string, mgr *indexManage
 		if watchedOnlineIndexCount == len(indexNames) {
 			return false, nil, nil
 		}
-		InfofCtx(ctx, KeyAll, "Indexes %s not ready - retrying...", strings.Join(offlineIndexes, ", "))
+		DebugfCtx(ctx, KeyAll, "Indexes %s not ready - retrying...", strings.Join(offlineIndexes, ", "))
 		return true, nil, nil
 	}, retrySleeper)
 	return err

base/leaky_datastore.go

@@ -394,6 +394,14 @@ func (lds *LeakyDataStore) CreateIndex(ctx context.Context, indexName string, ex
 	return n1qlStore.CreateIndex(ctx, indexName, expression, filterExpression, options)
 }
 
+func (lds *LeakyDataStore) CreateIndexIfNotExists(ctx context.Context, indexName string, expression string, filterExpression string, options *N1qlIndexOptions) error {
+	n1qlStore, err := lds.getN1QLStore()
+	if err != nil {
+		return err
+	}
+	return n1qlStore.CreateIndexIfNotExists(ctx, indexName, expression, filterExpression, options)
+}
+
 func (lds *LeakyDataStore) CreatePrimaryIndex(ctx context.Context, indexName string, options *N1qlIndexOptions) error {
 	n1qlStore, err := lds.getN1QLStore()
 	if err != nil {