sql/inspect: detect dangling secondary index entries in empty spans

INSPECT previously skipped index consistency checks when the primary
index span had no rows, preventing detection of dangling secondary index
entries in empty tables.

This change ensures INSPECT always scans secondary indexes, even if the
primary is empty, by using an empty predicate to trigger a full scan.
Any entries found are considered dangling.

Adds a test (TestDanglingIndexEntryInEmptyTable) migrated from
pkg/sql/scrub_test.go.

Informs #155485

Epic: CRDB-55075

Release note (bug fix): INSPECT now detects dangling secondary index
entries even when the primary index spans contain no data.

Author: spilchen

pkg/sql/inspect/index_consistency_check.go

@@ -165,41 +165,46 @@ func (c *indexConsistencyCheck) Start(
 		return errors.Wrap(err, "converting span to query bounds")
 	}
 
-	// Nothing to do if no rows exist in the span.
+	// If no rows exist in the primary index span, we still need to check for dangling
+	// secondary index entries. We run the check with an empty predicate, which will
+	// scan the entire secondary index within the span. Any secondary index entries found
+	// will be dangling since there are no corresponding primary index rows.
 	if !hasRows {
-		return nil
-	}
-
-	if len(bounds.Start) == 0 || len(bounds.End) == 0 {
-		return errors.AssertionFailedf("query bounds from span didn't produce start or end: %+v", bounds)
-	}
+		// Use empty predicate and no query arguments
+		predicate = ""
+		queryArgs = []interface{}{}
+	} else {
+		if len(bounds.Start) == 0 || len(bounds.End) == 0 {
+			return errors.AssertionFailedf("query bounds from span didn't produce start or end: %+v", bounds)
+		}
 
-	// Generate SQL predicate from the bounds
-	pkColNames := colNames(pkColumns)
-	// Encode column names for SQL usage
-	encodedPkColNames := make([]string, len(pkColNames))
-	for i, colName := range pkColNames {
-		encodedPkColNames[i] = encodeColumnName(colName)
-	}
-	predicate, err = spanutils.RenderQueryBounds(
-		encodedPkColNames, pkColDirs, pkColTypes,
-		len(bounds.Start), len(bounds.End), true, 1,
-	)
-	if err != nil {
-		return errors.Wrap(err, "rendering query bounds")
-	}
+		// Generate SQL predicate from the bounds
+		pkColNames := colNames(pkColumns)
+		// Encode column names for SQL usage
+		encodedPkColNames := make([]string, len(pkColNames))
+		for i, colName := range pkColNames {
+			encodedPkColNames[i] = encodeColumnName(colName)
+		}
+		predicate, err = spanutils.RenderQueryBounds(
+			encodedPkColNames, pkColDirs, pkColTypes,
+			len(bounds.Start), len(bounds.End), true, 1,
+		)
+		if err != nil {
+			return errors.Wrap(err, "rendering query bounds")
+		}
 
-	if strings.TrimSpace(predicate) == "" {
-		return errors.AssertionFailedf("query bounds from span didn't produce predicate: %+v", bounds)
-	}
+		if strings.TrimSpace(predicate) == "" {
+			return errors.AssertionFailedf("query bounds from span didn't produce predicate: %+v", bounds)
+		}
 
-	// Prepare query arguments: end bounds first, then start bounds
-	queryArgs = make([]interface{}, 0, len(bounds.End)+len(bounds.Start))
-	for _, datum := range bounds.End {
-		queryArgs = append(queryArgs, datum)
-	}
-	for _, datum := range bounds.Start {
-		queryArgs = append(queryArgs, datum)
+		// Prepare query arguments: end bounds first, then start bounds
+		queryArgs = make([]interface{}, 0, len(bounds.End)+len(bounds.Start))
+		for _, datum := range bounds.End {
+			queryArgs = append(queryArgs, datum)
+		}
+		for _, datum := range bounds.Start {
+			queryArgs = append(queryArgs, datum)
+		}
 	}
 
 	checkQuery := c.createIndexCheckQuery(

pkg/sql/inspect/index_consistency_check_test.go

@@ -552,6 +552,69 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 	}
 }
 
+// TestDanglingIndexEntryInEmptyTable verifies that INSPECT can detect dangling
+// secondary index entries even when the primary index is completely empty.
+func TestDanglingIndexEntryInEmptyTable(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	s, db, kvDB := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(ctx)
+	codec := s.ApplicationLayer().Codec()
+
+	// Create an empty table with a secondary index.
+	_, err := db.Exec(`
+CREATE DATABASE t;
+CREATE TABLE t.test (k INT PRIMARY KEY, v INT);
+CREATE INDEX secondary ON t.test (v);
+`)
+	require.NoError(t, err)
+
+	tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, codec, "t", "test")
+	secondaryIndex := tableDesc.PublicNonPrimaryIndexes()[0]
+
+	// Manually insert a dangling secondary index entry for a non-existent primary key.
+	// This creates corruption: a secondary index entry pointing to primary key (10)
+	// which doesn't exist in the primary index.
+	values := []tree.Datum{tree.NewDInt(10), tree.NewDInt(314)}
+	err = insertSecondaryIndexEntry(ctx, codec, values, kvDB, tableDesc, secondaryIndex)
+	require.NoError(t, err)
+
+	// Enable INSPECT command.
+	_, err = db.Exec(`SET enable_inspect_command = true`)
+	require.NoError(t, err)
+
+	// Run INSPECT and expect it to find the dangling index entry.
+	// INSPECT should return an error when inconsistencies are found.
+	_, err = db.Query(`INSPECT TABLE t.test AS OF SYSTEM TIME '-1us' WITH OPTIONS INDEX ALL`)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "INSPECT found inconsistencies")
+
+	// Verify the error details using SHOW INSPECT ERRORS.
+	rows, err := db.Query(`SHOW INSPECT ERRORS FOR TABLE t.test WITH DETAILS`)
+	require.NoError(t, err)
+	defer rows.Close()
+
+	var errorType, dbName, schemaName, tableName, primaryKey, aost, details string
+	var jobID int64
+	found := false
+	for rows.Next() {
+		err = rows.Scan(&errorType, &dbName, &schemaName, &tableName, &primaryKey, &jobID, &aost, &details)
+		require.NoError(t, err)
+		found = true
+
+		// Verify the error is a dangling secondary index entry.
+		require.Equal(t, "dangling_secondary_index_entry", errorType)
+		require.Equal(t, "t", dbName)
+		require.Equal(t, "test", tableName)
+		require.Equal(t, "'(10)'", primaryKey)
+		// Verify details contain the expected column value.
+		require.Contains(t, details, `"v": "314"`)
+	}
+
+	require.True(t, found, "expected to find at least one inspect error, but found none")
+}
+
 func TestIndexConsistencyWithReservedWordColumns(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/sql/inspect_job.go

@@ -132,6 +132,11 @@ func InspectChecksForTable(
 ) ([]*jobspb.InspectDetails_Check, error) {
 	checks := []*jobspb.InspectDetails_Check{}
 
+	// Skip virtual tables since they don't have physical storage to inspect.
+	if table.IsVirtualTable() {
+		return checks, nil
+	}
+
 	for _, index := range table.PublicNonPrimaryIndexes() {
 		if isUnsupportedIndexForIndexConsistencyCheck(index, table) {
 			if p != nil {

pkg/sql/logictest/testdata/logic_test/inspect

@@ -61,8 +61,9 @@ SELECT * FROM last_inspect_job;
 ----
 succeeded  1
 
+# TODO(155483): use a really old timestamp from before 'foo' was created.
 statement ok
-INSPECT TABLE foo AS OF SYSTEM TIME 1 WITH OPTIONS INDEX (idx_c1,idx_c2);
+INSPECT TABLE foo AS OF SYSTEM TIME '-50ms' WITH OPTIONS INDEX (idx_c1,idx_c2);
 
 query TI
 SELECT * FROM last_inspect_job;
@@ -86,7 +87,7 @@ INSPECT DATABASE test;
 query TI
 SELECT * FROM last_inspect_job;
 ----
-succeeded  28
+succeeded  2
 
 statement error index "idx_c3" does not exist
 INSPECT DATABASE test WITH OPTIONS INDEX (idx_c1, idx_c3);

pkg/sql/scrub_test.go

@@ -261,86 +261,6 @@ func addIndexEntryForDatums(
 	return kvDB.Put(context.Background(), entry.Key, &entry.Value)
 }
 
-// TestScrubIndexDanglingIndexReference tests that
-// `SCRUB TABLE ... INDEX“ will find dangling index references, which
-// are index entries that have no corresponding primary k/v. To test
-// this an index entry is generated and inserted. This creates a
-// dangling index error as the corresponding primary k/v is not equal.
-func TestScrubIndexDanglingIndexReference(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	defer log.Scope(t).Close(t)
-	s, db, kvDB := serverutils.StartServer(t, base.TestServerArgs{})
-	defer s.Stopper().Stop(context.Background())
-
-	// Create the table and the row entry.
-	if _, err := db.Exec(`
-CREATE DATABASE t;
-CREATE TABLE t.test (k INT PRIMARY KEY, v INT);
-CREATE INDEX secondary ON t.test (v);
-`); err != nil {
-		t.Fatalf("unexpected error: %s", err)
-	}
-
-	tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, keys.SystemSQLCodec, "t", "test")
-	secondaryIndex := tableDesc.PublicNonPrimaryIndexes()[0]
-	// Construct datums and secondary k/v for our row values (k, v).
-	values := []tree.Datum{tree.NewDInt(10), tree.NewDInt(314)}
-
-	// Put the new secondary k/v into the database.
-	if err := addIndexEntryForDatums(values, kvDB, tableDesc, secondaryIndex); err != nil {
-		t.Fatalf("unexpected error: %s", err)
-	}
-
-	// Run SCRUB and find the index errors we created.
-	rows, err := db.Query(`EXPERIMENTAL SCRUB TABLE t.test WITH OPTIONS INDEX ALL`)
-	if err != nil {
-		t.Fatalf("unexpected error: %s", err)
-	}
-	defer rows.Close()
-
-	results, err := sqlutils.GetInspectResultRows(rows)
-	if err != nil {
-		t.Fatalf("unexpected error: %s", err)
-	}
-
-	if len(results) != 1 {
-		t.Fatalf("expected 1 result, got %d. got %#v", len(results), results)
-	}
-	if result := results[0]; result.ErrorType != scrub.DanglingIndexReferenceError {
-		t.Fatalf("expected %q error, instead got: %s",
-			scrub.DanglingIndexReferenceError, result.ErrorType)
-	} else if result.Database != "t" {
-		t.Fatalf("expected database %q, got %q", "t", result.Database)
-	} else if result.Table != "test" {
-		t.Fatalf("expected table %q, got %q", "test", result.Table)
-	} else if result.PrimaryKey != "(10)" {
-		t.Fatalf("expected primaryKey %q, got %q", "(10)", result.PrimaryKey)
-	} else if result.Repaired {
-		t.Fatalf("expected repaired %v, got %v", false, result.Repaired)
-	} else if !strings.Contains(result.Details, `"v": "314"`) {
-		t.Fatalf("expected error details to contain `%s`, got %s", `"v": "314"`, result.Details)
-	}
-
-	// Run SCRUB DATABASE to make sure it also catches the problem.
-	rows, err = db.Query(`EXPERIMENTAL SCRUB DATABASE t`)
-	if err != nil {
-		t.Fatalf("unexpected error: %+v", err)
-	}
-	defer rows.Close()
-	scrubDatabaseResults, err := sqlutils.GetInspectResultRows(rows)
-	if err != nil {
-		t.Fatalf("unexpected error: %s", err)
-	} else if len(scrubDatabaseResults) != 1 {
-		t.Fatalf("expected 1 result, got %d. got %#v", len(scrubDatabaseResults), scrubDatabaseResults)
-	} else if !(scrubDatabaseResults[0].ErrorType == results[0].ErrorType &&
-		scrubDatabaseResults[0].Database == results[0].Database &&
-		scrubDatabaseResults[0].Table == results[0].Table &&
-		scrubDatabaseResults[0].Details == results[0].Details) {
-		t.Fatalf("expected results to be equal, SCRUB TABLE got %v. SCRUB DATABASE got %v",
-			results, scrubDatabaseResults)
-	}
-}
-
 // TestScrubIndexCatchesStoringMismatch tests that
 // `SCRUB TABLE ... INDEX ALL` will fail if an index entry only differs
 // by its STORING values. To test this, a row's underlying secondary