sql: support vector indexes with external row data

This commit fixes a few cases where vector-indexing logic was not
correctly using the table's `ExternalRowData` if it was set. It also
initializes `cspann.VectorIndex` instances in read-only mode if the
table with the index has external row data.

Fixes #144679

Release note (bug fix): Fixed a bug in alpha versions of 25.2 as well
as `v25.2.0-beta.1` that would cause vector indexes to return incorrect
or no results from a standby reader in a PCR setup.

Author: DrewKimball

pkg/ccl/testccl/sqlccl/BUILD.bazel

@@ -41,8 +41,6 @@ go_test(
         "//pkg/settings/cluster",
         "//pkg/spanconfig",
         "//pkg/sql",
-        "//pkg/sql/catalog/lease",
-        "//pkg/sql/catalog/replication",
         "//pkg/sql/gcjob",
         "//pkg/sql/isql",
         "//pkg/sql/lexbase",

pkg/ccl/testccl/sqlccl/standby_read_test.go

@@ -10,16 +10,12 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
-	"github.com/cockroachdb/cockroach/pkg/sql"
-	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
-	"github.com/cockroachdb/cockroach/pkg/sql/catalog/replication"
-	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
-	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -73,33 +69,14 @@ func TestStandbyRead(t *testing.T) {
 
 	srcRunner := sqlutils.MakeSQLRunner(srcDB)
 	dstRunner := sqlutils.MakeSQLRunner(dstDB)
-	dstInternal := dstTenant.InternalDB().(*sql.InternalDB)
 
 	dstRunner.Exec(t, `SET CLUSTER SETTING sql.defaults.distsql = always`)
 	dstRunner.Exec(t, `SET distsql = always`)
 
-	waitForReplication := func() {
-		now := ts.Clock().Now()
-		err := replication.SetupOrAdvanceStandbyReaderCatalog(
-			ctx, serverutils.TestTenantID(), now, dstInternal, dstTenant.ClusterSettings(),
-		)
-		if err != nil {
-			t.Fatal(err)
-		}
-		now = ts.Clock().Now()
-		lm := dstTenant.LeaseManager().(*lease.Manager)
-		testutils.SucceedsSoon(t, func() error {
-			if lm.GetSafeReplicationTS().Less(now) {
-				return errors.AssertionFailedf("waiting for descriptor close timestamp to catch up")
-			}
-			return nil
-		})
-	}
-
 	for _, tc := range testcases {
 		var runner *sqlutils.SQLRunner
 		if tc.standby {
-			waitForReplication()
+			testcluster.WaitForStandbyTenantReplication(t, ctx, ts.Clock(), dstTenant)
 			runner = dstRunner
 		} else {
 			runner = srcRunner

pkg/sql/opt_exec_factory.go

@@ -1865,7 +1865,9 @@ func (ef *execFactory) ConstructVectorSearch(
 
 	// Encode the prefix constraint as a list of roachpb.Keys.
 	var sb span.Builder
-	sb.Init(ef.planner.EvalContext(), ef.planner.ExecCfg().Codec, tabDesc, indexDesc)
+	sb.InitAllowingExternalRowData(
+		ef.planner.EvalContext(), ef.planner.ExecCfg().Codec, tabDesc, indexDesc,
+	)
 	prefixKeys, err := sb.KeysFromVectorPrefixConstraint(ef.ctx, prefixConstraint)
 	if err != nil {
 		return nil, err

pkg/sql/vecindex/BUILD.bazel

@@ -43,6 +43,7 @@ go_library(
 go_test(
     name = "vecindex_test",
     srcs = [
+        "main_test.go",
         "manager_test.go",
         "searcher_test.go",
         "vecindex_test.go",
@@ -51,6 +52,8 @@ go_test(
     embed = [":vecindex"],
     deps = [
         "//pkg/base",
+        "//pkg/ccl",
+        "//pkg/ccl/storageccl",
         "//pkg/keys",
         "//pkg/roachpb",
         "//pkg/security/securityassets",
@@ -78,7 +81,9 @@ go_test(
         "//pkg/sql/vecindex/vecpb",
         "//pkg/sql/vecindex/vecstore",
         "//pkg/testutils/serverutils",
+        "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
         "//pkg/util/encoding",
         "//pkg/util/leaktest",
         "//pkg/util/log",

pkg/sql/vecindex/main_test.go

@@ -0,0 +1,31 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package vecindex_test
+
+import (
+	"os"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/ccl"
+	_ "github.com/cockroachdb/cockroach/pkg/ccl/storageccl"
+	"github.com/cockroachdb/cockroach/pkg/security/securityassets"
+	"github.com/cockroachdb/cockroach/pkg/security/securitytest"
+	"github.com/cockroachdb/cockroach/pkg/server"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
+)
+
+func TestMain(m *testing.M) {
+	defer ccl.TestingEnableEnterprise()()
+	securityassets.SetLoader(securitytest.EmbeddedAssets)
+	randutil.SeedForTests()
+	serverutils.InitTestServerFactory(server.TestServerFactory)
+	serverutils.InitTestClusterFactory(testcluster.TestClusterFactory)
+	os.Exit(m.Run())
+}
+
+//go:generate ../../util/leaktest/add-leaktest.sh *_test.go

pkg/sql/vecindex/manager.go

@@ -171,13 +171,17 @@ func (m *Manager) GetWithDesc(
 			// TODO(drewk): use the config to populate the index options as well.
 			config := index.GetVecConfig()
 			quantizer := quantize.NewRaBitQuantizer(int(config.Dims), config.Seed)
-			store, err := vecstore.NewWithColumnID(m.db, quantizer, m.codec, desc, index.GetID(), index.VectorColumnID())
+			store, err := vecstore.NewWithColumnID(
+				ctx, m.db, quantizer, m.codec, desc, index.GetID(), index.VectorColumnID(),
+			)
 			if err != nil {
 				return nil, err
 			}
 
 			return cspann.NewIndex(
-				m.ctx, store, quantizer, config.Seed, m.getIndexOptions(config), m.stopper)
+				m.ctx, store, quantizer, config.Seed,
+				m.getIndexOptions(config, store.ReadOnly()), m.stopper,
+			)
 		},
 	)
 }
@@ -207,12 +211,14 @@ func (m *Manager) Get(
 			// passed to cspann.NewIndex, and we don't want that to be the context of
 			// the Get call.
 			return cspann.NewIndex(
-				m.ctx, store, quantizer, config.Seed, m.getIndexOptions(config), m.stopper)
+				m.ctx, store, quantizer, config.Seed,
+				m.getIndexOptions(config, store.ReadOnly()), m.stopper,
+			)
 		},
 	)
 }
 
-func (m *Manager) getIndexOptions(config vecpb.Config) *cspann.IndexOptions {
+func (m *Manager) getIndexOptions(config vecpb.Config, readOnly bool) *cspann.IndexOptions {
 	return &cspann.IndexOptions{
 		MinPartitionSize: int(config.MinPartitionSize),
 		MaxPartitionSize: int(config.MaxPartitionSize),
@@ -222,6 +228,7 @@ func (m *Manager) getIndexOptions(config vecpb.Config) *cspann.IndexOptions {
 			return StalledOpTimeoutSetting.Get(m.sv)
 		},
 		IsDeterministic: config.IsDeterministic,
+		ReadOnly:        readOnly,
 	}
 }
 

pkg/sql/vecindex/manager_test.go

@@ -7,17 +7,13 @@ package vecindex_test
 
 import (
 	"context"
-	"os"
 	"strconv"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
-	"github.com/cockroachdb/cockroach/pkg/security/securityassets"
-	"github.com/cockroachdb/cockroach/pkg/security/securitytest"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
-	"github.com/cockroachdb/cockroach/pkg/server"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catenumpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
@@ -34,18 +30,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
-	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/stretchr/testify/require"
 )
 
-func TestMain(m *testing.M) {
-	securityassets.SetLoader(securitytest.EmbeddedAssets)
-	randutil.SeedForTests()
-	serverutils.InitTestServerFactory(server.TestServerFactory)
-
-	os.Exit(m.Run())
-}
-
 func buildTestTable(tableID catid.DescID, tableName string) catalog.MutableTableDescriptor {
 	return tabledesc.NewBuilder(&descpb.TableDescriptor{
 		ID:       tableID,

pkg/sql/vecindex/searcher_test.go

@@ -65,6 +65,7 @@ func TestSearcher(t *testing.T) {
 
 	quantizer := quantize.NewUnQuantizer(2)
 	store, err := vecstore.NewWithColumnID(
+		ctx,
 		internalDB,
 		quantizer,
 		codec,

pkg/sql/vecindex/vecindex_test.go

@@ -25,15 +25,19 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/vecindex/cspann/testutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/vecindex/vecencoding"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/cockroachdb/cockroach/pkg/util/vector"
+	"github.com/stretchr/testify/require"
 )
 
-// TestVecindexConcurrency builds an index on multiple goroutines, with
+// TestVecIndexConcurrency builds an index on multiple goroutines, with
 // background splits and merges enabled.
-func TestVecindexConcurrency(t *testing.T) {
+func TestVecIndexConcurrency(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
@@ -143,6 +147,77 @@ func TestVecindexConcurrency(t *testing.T) {
 	logProgress()
 }
 
+// TestVecIndexStandbyReader builds an index on a source tenant and verifies
+// that a PCR standby reader can read the index, but doesn't attempt to initiate
+// fixups.
+func TestVecIndexStandbyReader(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	skip.UnderDuress(t, "slow test")
+
+	rnd, seed := randutil.NewTestRand()
+	t.Logf("random seed: %v", seed)
+
+	ctx := context.Background()
+	tc := serverutils.StartCluster(t, 3, /* numNodes */
+		base.TestClusterArgs{
+			ReplicationMode: base.ReplicationManual,
+			ServerArgs: base.TestServerArgs{
+				DefaultTestTenant: base.TestControlsTenantsExplicitly,
+			},
+		})
+	defer tc.Stopper().Stop(ctx)
+	ts := tc.Server(0)
+
+	_, srcDB, err := ts.TenantController().StartSharedProcessTenant(ctx,
+		base.TestSharedProcessTenantArgs{
+			TenantID:    serverutils.TestTenantID(),
+			TenantName:  "src",
+			UseDatabase: "defaultdb",
+		},
+	)
+	require.NoError(t, err)
+	dstTenant, dstDB, err := ts.TenantController().StartSharedProcessTenant(ctx,
+		base.TestSharedProcessTenantArgs{
+			TenantID:    serverutils.TestTenantID2(),
+			TenantName:  "dst",
+			UseDatabase: "defaultdb",
+		},
+	)
+	require.NoError(t, err)
+
+	srcRunner := sqlutils.MakeSQLRunner(srcDB)
+	dstRunner := sqlutils.MakeSQLRunner(dstDB)
+
+	// Enable vector indexes.
+	srcRunner.Exec(t, `SET CLUSTER SETTING feature.vector_index.enabled = true`)
+
+	// Construct the table.
+	srcRunner.Exec(t, "CREATE TABLE t (id INT PRIMARY KEY, v VECTOR(512), VECTOR INDEX foo (v))")
+
+	// Load features and build the index.
+	const batchSize = 10
+	const numBatches = 100
+	vectors := testutils.LoadFeatures(t, batchSize*numBatches)
+	for i := 0; i < numBatches; i++ {
+		insertVectors(t, srcRunner, i*batchSize, vectors.Slice(i*batchSize, batchSize))
+	}
+
+	// Wait for the standby reader to catch up.
+	asOf := testcluster.WaitForStandbyTenantReplication(t, ctx, ts.Clock(), dstTenant)
+
+	const queryTemplate = `SELECT * FROM t@foo %s ORDER BY v <-> '%s' LIMIT 3`
+	asOfClause := fmt.Sprintf("AS OF SYSTEM TIME %s", asOf.AsOfSystemTime())
+	for range 10 {
+		// Select a random vector from the set and run an ANN query against both
+		// tenants. The query results should be identical.
+		vec := vectors.At(rnd.Intn(vectors.Count)).String()
+		expected := srcRunner.QueryStr(t, fmt.Sprintf(queryTemplate, asOfClause, vec))
+		dstRunner.CheckQueryResults(t, fmt.Sprintf(queryTemplate, "", vec), expected)
+	}
+}
+
 // Insert block of vectors within the scope of a transaction.
 func insertVectors(t *testing.T, runner *sqlutils.SQLRunner, startId int, vectors vector.Set) {
 	var valuesClause strings.Builder
@@ -161,8 +236,8 @@ func insertVectors(t *testing.T, runner *sqlutils.SQLRunner, startId int, vector
 	runner.Exec(t, query, args...)
 }
 
-// TestVecindexDeletion tests that rows can be properly deleted from a vector index.
-func TestVecindexDeletion(t *testing.T) {
+// TestVecIndexDeletion tests that rows can be properly deleted from a vector index.
+func TestVecIndexDeletion(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 

pkg/sql/vecindex/vecstore/BUILD.bazel

@@ -28,6 +28,7 @@ go_library(
         "//pkg/sql/vecindex/cspann/quantize",
         "//pkg/sql/vecindex/cspann/workspace",
         "//pkg/sql/vecindex/vecencoding",
+        "//pkg/util/log",
         "//pkg/util/unique",
         "//pkg/util/vector",
         "@com_github_cockroachdb_errors//:errors",