sql: improve observability of index merge timestamp

When building an index, the merge phase copies any new rows added since
the index build started. A timestamp is used to set an upper bound for
this scan. We suspect that the chosen timestamp may be omitting some
rows.

Our theory is that this happens if new rows were inserted before the
merge phase but were committed with a timestamp after the merge. This
discrepancy could be due to clock skew between nodes. However, attempts
to reproduce this issue via unit tests were unsuccessful, so this
remains a hypothesis.

To address this, this change includes:
- An adjustment to the merge timestamp to account for potential clock
skew.
- Additional logging of the merge timestamp chosen and the timestamps
observed on each node when draining leased descriptors. These logs will
help track the timestamp across nodes during the merge phase.
- Treat index validation errors for non-unique indexes as assertion
failures. We previously treated as a duplicat key error, which was
very confusing because duplicates are allowed for non-unique indexes.

Epic: none
Release note: none
Closes: #142751

Author: spilchen

pkg/sql/backfill.go

@@ -1929,6 +1929,15 @@ func ValidateForwardIndexes(
 							idx.GetID())
 						indexName = idx.GetName()
 					}
+					if !idx.IsUnique() {
+						// For non-unique indexes, the table row count must match the index
+						// key count. Unlike unique indexes, we don't filter out any rows,
+						// so every row must have a corresponding entry in the index. A
+						// mismatch indicates an assertion failure.
+						return errors.AssertionFailedf(
+							"validation of non-unique index %s failed: expected %d rows, found %d",
+							idx.GetName(), errors.Safe(expectedCount), errors.Safe(idxLen))
+					}
 					// TODO(vivek): find the offending row and include it in the error.
 					return pgerror.WithConstraintName(pgerror.Newf(pgcode.UniqueViolation,
 						"duplicate key value violates unique constraint %q",
@@ -2989,8 +2998,15 @@ func indexTruncateInTxn(
 // part of a restore, then timestamp will be too old and the job will fail. On
 // the next resume, a mergeTimestamp newer than the GC time will be picked and
 // the job can continue.
-func getMergeTimestamp(clock *hlc.Clock) hlc.Timestamp {
-	return clock.Now()
+func getMergeTimestamp(ctx context.Context, clock *hlc.Clock) hlc.Timestamp {
+	// Use the current timestamp plus the maximum allowed offset to account for
+	// potential clock skew across nodes. The chosen timestamp must be greater
+	// than all commit timestamps used so far. This may result in seeing rows
+	// that are already present in the index being merged, but that’s fine as
+	// they will be treated as no-ops.
+	ts := clock.Now().AddDuration(clock.MaxOffset())
+	log.Infof(ctx, "merging all keys in temporary index before time %v", ts)
+	return ts
 }
 
 func (sc *SchemaChanger) distIndexMerge(
@@ -3001,8 +3017,7 @@ func (sc *SchemaChanger) distIndexMerge(
 	fractionScaler *multiStageFractionScaler,
 ) error {
 
-	mergeTimestamp := getMergeTimestamp(sc.clock)
-	log.Infof(ctx, "merging all keys in temporary index before time %v", mergeTimestamp)
+	mergeTimestamp := getMergeTimestamp(ctx, sc.clock)
 
 	// Gather the initial resume spans for the merge process.
 	progress, err := extractMergeProgress(sc.job, tableDesc, addedIndexes, temporaryIndexes)

pkg/sql/catalog/lease/lease.go

@@ -475,6 +475,7 @@ func (m *Manager) WaitForOneVersion(
 			return nil, err
 		}
 		if detail.count == 0 {
+			log.Infof(ctx, "all leases have expired at %v: desc=%v", now, descs)
 			break
 		}
 		if detail.count != lastCount {

pkg/sql/mvcc_backfiller.go

@@ -92,7 +92,7 @@ func (im *IndexBackfillerMergePlanner) MergeIndexes(
 		)
 		return tracker.SetMergeProgress(ctx, progress)
 	}
-	mergeTimeStamp := getMergeTimestamp(im.execCfg.Clock)
+	mergeTimeStamp := getMergeTimestamp(ctx, im.execCfg.Clock)
 	protectedTimestampCleaner := im.execCfg.ProtectedTimestampManager.TryToProtectBeforeGC(ctx, job, descriptor, mergeTimeStamp)
 	defer func() {
 		cleanupError := protectedTimestampCleaner(ctx)

pkg/sql/schemachanger/scexec/BUILD.bazel

@@ -109,6 +109,9 @@ go_test(
         "@com_github_stretchr_testify//require",
         "//pkg/sql/sem/idxtype",
         "//pkg/sql/catalog/lease",
+        "//pkg/sql/schemachanger/scplan",
+        "//pkg/testutils/testcluster",
+        "//pkg/util/ctxgroup",
     ],
 )
 

pkg/sql/schemachanger/scexec/exec_backfill_test.go

@@ -7,10 +7,13 @@ package scexec_test
 
 import (
 	"context"
+	"fmt"
 	"math/rand"
+	"sync/atomic"
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"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/descpb"
@@ -19,9 +22,12 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scdeps/sctestdeps"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scop"
+	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/idxtype"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/iterutil"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -333,3 +339,84 @@ func TestExecBackfiller(t *testing.T) {
 	}
 
 }
+
+// TestMergeTimestampSkew will ensure we do not miss rows during the merge phase
+// if the clocks are skewed.
+func TestMergeTimestampSkew(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	var hookEnabled atomic.Bool
+	var ingestFn func() error
+	tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			Knobs: base.TestingKnobs{
+				SQLDeclarativeSchemaChanger: &scexec.TestingKnobs{
+					BeforeStage: func(p scplan.Plan, stageIdx int) error {
+						if !hookEnabled.Load() {
+							return nil
+						}
+						return ingestFn()
+					},
+				},
+			},
+		},
+	})
+	defer tc.Stopper().Stop(ctx)
+	db1 := tc.ApplicationLayer(0).SQLConn(t, serverutils.DBName("defaultdb"))
+	r1 := sqlutils.MakeSQLRunner(db1)
+
+	// Add a second node so that we can use a separate clock for it.
+	manualClock2 := hlc.NewHybridManualClock()
+	tc.AddAndStartServer(t, base.TestServerArgs{
+		Knobs: base.TestingKnobs{
+			Server: &server.TestingKnobs{
+				WallClock: manualClock2,
+			},
+		},
+	})
+	db2 := tc.ApplicationLayer(1).SQLConn(t, serverutils.DBName("defaultdb"))
+	r2 := sqlutils.MakeSQLRunner(db2)
+
+	t.Run("create_non_unique_index", func(t *testing.T) {
+		r1.ExecMultiple(t,
+			`ALTER DATABASE defaultdb CONFIGURE ZONE USING num_replicas = 1, constraints = '[-node2]'`,
+			"CREATE TABLE t_idx(n int)",
+			"INSERT INTO t_idx(n) SELECT * FROM generate_series(1, 100)",
+		)
+		additionalInserts := 0
+
+		// Each stage we will insert a new row from a different node. That node will
+		// use a skewed clock.
+		ingestFn = func() error {
+			manualClock2.Increment(10000000)
+			keyVal := 1000 + additionalInserts
+			additionalInserts++
+			r2.Exec(t, fmt.Sprintf("INSERT INTO t_idx(n) VALUES (%d)", keyVal))
+			return nil
+		}
+		hookEnabled.Store(true)
+		defer hookEnabled.Store(false)
+
+		grp := ctxgroup.WithContext(ctx)
+		grp.GoCtx(func(ctx context.Context) error {
+			r1.Exec(t, "CREATE INDEX i1 ON t_idx (n)")
+			hookEnabled.Store(false)
+			return nil
+		})
+		require.NoError(t, grp.Wait())
+
+		// Compare row count with between new index and the primary key
+		for _, q := range []string{
+			`SELECT count(1) FROM t_idx@i1`,
+			`SELECT count(1) FROM t_idx@t_idx_pkey`,
+		} {
+			var rowCount int
+			res := r1.QueryRow(t, q)
+			res.Scan(&rowCount)
+			expectedRowCount := 100 + additionalInserts
+			require.Equal(t, expectedRowCount, rowCount, "post create index row count mismatch in query %q", q)
+		}
+	})
+}

pkg/sql/schemachanger/scexec/exec_validation.go

@@ -16,9 +16,7 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
-func executeValidateUniqueIndex(
-	ctx context.Context, deps Dependencies, op *scop.ValidateIndex,
-) error {
+func executeValidateIndex(ctx context.Context, deps Dependencies, op *scop.ValidateIndex) error {
 	descs, err := deps.Catalog().MustReadImmutableDescriptors(ctx, op.TableID)
 	if err != nil {
 		return err
@@ -118,7 +116,7 @@ func executeValidationOps(ctx context.Context, deps Dependencies, ops []scop.Op)
 func executeValidationOp(ctx context.Context, deps Dependencies, op scop.Op) (err error) {
 	switch op := op.(type) {
 	case *scop.ValidateIndex:
-		if err = executeValidateUniqueIndex(ctx, deps, op); err != nil {
+		if err = executeValidateIndex(ctx, deps, op); err != nil {
 			if !scerrors.HasSchemaChangerUserError(err) {
 				return errors.Wrapf(err, "%T: %v", op, op)
 			}