Merge #142917

142917: sql: improve observability of index merge timestamp r=spilchen a=spilchen

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 duplicate key error, which was very confusing because duplicates are allowed for non-unique indexes.

Epic: none
Release note: none
Closes: #142861 

Co-authored-by: Matt Spilchen <[email protected]>

Author: craig[bot]

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)
 			}