builtins: add process_vector_index_fixups builtin

Add a new builtin function that waits until all pending fixups have
been processed for a given index. This is used for testing.

Previously, we were processing fixups in deterministic tests by calling
ProcessFixups just before inserting vectors or searching for them.
However, that was causing deadlocks, because fixups were being processed
in the same transaction that was updating the index.

The vector_index test is now updated to call the new builtin when it
needs to force fixup processing.

Epic: CRDB-42943
Release note: None

Author: andy-kimball

pkg/sql/faketreeeval/evalctx.go

@@ -577,6 +577,13 @@ func (ep *DummyEvalPlanner) RetryCounter() int {
 	return 0
 }
 
+// ProcessVectorIndexFixups is part of the eval.Planner interface.
+func (ep *DummyEvalPlanner) ProcessVectorIndexFixups(
+	ctx context.Context, tableID descpb.ID, indexID descpb.IndexID,
+) error {
+	return nil
+}
+
 // DummyPrivilegedAccessor implements the tree.PrivilegedAccessor interface by returning errors.
 type DummyPrivilegedAccessor struct{}
 

pkg/sql/logictest/testdata/logic_test/vector_index

@@ -539,12 +539,37 @@ INSERT INTO exec_test (a, b, vec1) VALUES
   (2, 1, '[4, 5, 6]'),
   (3, 2, '[7, 8, 9]'),
   (4, 2, '[10, 11, 12]'),
-  (5, 2, '[13, 14, 15]'),
+  (5, 2, '[13, 14, 15]');
+
+statement ok
+INSERT INTO exec_test (a, b, vec1) VALUES
   (6, NULL, '[16, 17, 18]'),
   (7, NULL, '[1, 1, 1]'),
   (8, NULL, NULL),
   (9, 3, NULL);
 
+# Define 
+statement ok
+CREATE PROCEDURE process_vector_index_fixups_for(desc_name STRING, idx_name STRING)
+LANGUAGE SQL
+AS $$
+  SELECT crdb_internal.process_vector_index_fixups(
+    (SELECT id FROM system.namespace WHERE name = desc_name),
+    (
+      SELECT index_id
+      FROM crdb_internal.table_indexes
+      WHERE descriptor_name = desc_name AND index_name = idx_name
+    )
+  )
+$$;
+
+# Process split fixups for the two indexes.
+statement ok
+CALL process_vector_index_fixups_for('exec_test', 'idx1');
+
+statement ok
+CALL process_vector_index_fixups_for('exec_test', 'idx2');
+
 statement error pgcode 22000 pq: expected 3 dimensions, not 1
 INSERT INTO exec_test (a, b, vec1) VALUES (10, 1, '[1]');
 
@@ -765,9 +790,22 @@ INSERT INTO distance_metrics (a, v) VALUES
   (1, '[0, 0]'),
   (2, '[-2, -2]'),
   (3, '[2, 2]'),
-  (4, '[4, 4]'),
+  (4, '[4, 4]');
+
+statement ok
+INSERT INTO distance_metrics (a, v) VALUES
   (5, '[-2, 2]');
 
+# Process split fixups for the three indexes.
+statement ok
+CALL process_vector_index_fixups_for('distance_metrics', 'idx1');
+
+statement ok
+CALL process_vector_index_fixups_for('distance_metrics', 'idx2');
+
+statement ok
+CALL process_vector_index_fixups_for('distance_metrics', 'idx3');
+
 # Results using L2 distance.
 query ITF rowsort
 SELECT a, v, round(v <-> '[2, 2]', 2) dist FROM distance_metrics ORDER BY v <-> '[2, 2]' LIMIT 5;

pkg/sql/planner.go

@@ -1077,3 +1077,15 @@ func (p *planner) ExtendHistoryRetention(ctx context.Context, jobID jobspb.JobID
 func (p *planner) RetryCounter() int {
 	return p.autoRetryCounter + p.autoRetryStmtCounter
 }
+
+// ProcessVectorIndexFixups is part of the eval.Planner interface.
+func (p *planner) ProcessVectorIndexFixups(
+	ctx context.Context, tableID descpb.ID, indexID descpb.IndexID,
+) error {
+	vi, err := p.execCfg.VecIndexManager.Get(ctx, tableID, indexID)
+	if err != nil {
+		return err
+	}
+	vi.ProcessFixups()
+	return nil
+}

pkg/sql/sem/builtins/builtins.go

@@ -9357,6 +9357,30 @@ WHERE object_id = table_descriptor_id
 			},
 		},
 	),
+	"crdb_internal.process_vector_index_fixups": makeBuiltin(
+		tree.FunctionProperties{
+			Category:     builtinconstants.CategoryTesting,
+			Undocumented: true,
+		},
+		tree.Overload{
+			Types: tree.ParamTypes{
+				{Name: "table_id", Typ: types.Int},
+				{Name: "index_id", Typ: types.Int},
+			},
+			ReturnType: tree.FixedReturnType(types.Void),
+			Fn: func(ctx context.Context, evalCtx *eval.Context, args tree.Datums) (tree.Datum, error) {
+				tableID := descpb.ID(tree.MustBeDInt(args[0]))
+				indexID := descpb.IndexID(tree.MustBeDInt(args[1]))
+				err := evalCtx.Planner.ProcessVectorIndexFixups(ctx, tableID, indexID)
+				if err != nil {
+					return nil, err
+				}
+				return tree.DVoidDatum, nil
+			},
+			Info:       "Waits until all outstanding fixups for the vector index with the given ID have been processed.",
+			Volatility: volatility.Volatile,
+		},
+	),
 }
 
 var lengthImpls = func(incBitOverload bool) builtinDefinition {

pkg/sql/sem/builtins/fixed_oids.go

@@ -2795,6 +2795,7 @@ var builtinOidsArray = []string{
 	2840: `lead(val: ltree, n: int) -> ltree`,
 	2841: `lead(val: ltree, n: int, default: ltree) -> ltree`,
 	2842: `last_value(val: ltree) -> ltree`,
+	2843: `crdb_internal.process_vector_index_fixups(table_id: int, index_id: int) -> void`,
 }
 
 var builtinOidsBySignature map[string]oid.Oid

pkg/sql/sem/eval/deps.go

@@ -455,6 +455,10 @@ type Planner interface {
 
 	// RetryCounter is the number of times this statement has been retried.
 	RetryCounter() int
+
+	// ProcessVectorIndexFixups waits until all outstanding fixups for the vector
+	// index with the given ID have been processed.
+	ProcessVectorIndexFixups(ctx context.Context, tableID descpb.ID, indexID descpb.IndexID) error
 }
 
 // InternalRows is an iterator interface that's exposed by the internal

pkg/sql/vecindex/mutation_searcher.go

@@ -52,12 +52,6 @@ func (s *MutationSearcher) Init(
 	s.txn.Init(evalCtx, idx.Store().(*vecstore.Store), txn, getFullVectorsFetchSpec)
 	s.idxCtx.Init(&s.txn)
 	s.evalCtx = evalCtx
-
-	// If the index is deterministic, then synchronously run the background worker
-	// to process any pending fixups.
-	if s.idx.Options().IsDeterministic {
-		s.idx.ProcessFixups()
-	}
 }
 
 // SearchForInsert triggers a search for the partition in which to insert the

pkg/sql/vecindex/searcher.go

@@ -62,12 +62,6 @@ func (s *Searcher) Init(
 	}
 	s.searchSet.MaxResults, s.searchSet.MaxExtraResults =
 		cspann.IncreaseRerankResults(baseBeamSize, maxResults, rerankMultiplier)
-
-	// If the index is deterministic, then synchronously run the background worker
-	// to process any pending fixups.
-	if idx.Options().IsDeterministic {
-		s.idx.ProcessFixups()
-	}
 }
 
 // Search triggers a search over the index for the given vector, within the