Merge #143270 #143451

143270: kvserver: better obs in TestTxnReadWithinUncertaintyIntervalAfterRangeMerge r=tbg a=tbg

Closes #143260.

This is a complex test with a rare failure mode. Trace all of the relevant operations so that we can meaningfully engage with it.

Epic: none
Release note: none

143451: sql,backfill: avoid holding a protected timestamp during the backfill r=rafiss a=rafiss

### sql,rowexec: use WriteTS as timestamp for AddSSTable in backfill

PR #64023 made it so we use the backfill's read TS as the batch TS for
the AddSSTable operation. Simultaneously with that change, #63945 was
being worked on, which made it so that we use a fixed _write_ timestamp
for the keys that are backfilled. This write timestamp is the actual
minimum lower bound timestamp for the backfill operation, and the read
timstamp is allowed to change, so it makes more sense to use the write
timestamp for the AddSSTable operation. Looking at the diffs in the PRs
makes it seem pretty likely that this was just missed as part of a
trivial branch skew.

This commit does not cause any behavior change, since at the time of
writing, the read timestamp and write timestamp are always identical for
the backfill.

Release note: None

### sql,backfill: choose new read timestamp when backfill job is resumed

When #73861 got merged, we seem to have accidentally made it so the index
backfiller will never select a new read timestamp. That is contrary to
the goals of #63945, where we initially added the ability to advance the
read timestamp forward.

This mistake is why we were seeing behavior where a GC threshold error
would cause the backfill to retry infinitely (which in turn led us to treat
as a permanent error in #139203, which was something we never should have
done). We never noticed the bug because when CREATE INDEX was added to the
declarative schema changer in #92128, we turned off the declarative schema
changer in the test that would have caught this (i.e. TestIndexBackfillAfterGC).

This commit makes it so we choose a current timestamp as the read
timestamp when planning the backfill, and fixes up that test to show
that this works for index backfills in the declarative schema changer.

Release note (bug fix): Fixed a bug where a GC threshold error (which
appears as "batch timestamp must be after replica GC threshold ...")
could cause a schema change that backfills data to fail. Now, the error
will cause the backfill to be retried at a higher timestamp to avoid the
error.

### sql, backfill: use a current timestamp to scan each chunk of the source index

There's no need for the backfill to use a fixed read timestamp for the
entire job. Instead, each chunk of the original index that needs to be
scanned can use a current timestamp. This allows us to remove all the
logic for adding protected timestamps for the backfill.

Release note (performance improvement): Schema changes that require
data to be backfilled no longer hold a protected timestamp for the
entire duration of the backfill, which means there is less overhead
caused by MVCC garbage collection after the backfill completes.

### sql: stop setting readAsOf in index backfiller

As of the parent commit, this is unused by the index backfiller.

Release note: None

---

fixes #140629
informs #142339
informs #142117
informs #141773

Co-authored-by: Tobias Grieger <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>

Author: 3 people

pkg/kv/kvpb/BUILD.bazel

@@ -78,6 +78,7 @@ go_test(
         "//pkg/roachpb",
         "//pkg/storage/enginepb",
         "//pkg/testutils/echotest",
+        "//pkg/testutils/skip",
         "//pkg/util/buildutil",
         "//pkg/util/encoding",
         "//pkg/util/hlc",

pkg/kv/kvpb/batch_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/kr/pretty"
 	"github.com/stretchr/testify/require"
@@ -554,6 +555,7 @@ func TestRefreshSpanIterateSkipLocked(t *testing.T) {
 }
 
 func TestResponseKeyIterate(t *testing.T) {
+	skip.UnderNonTestBuild(t) // some assertions that are checked are only returned in test builds
 	keyA, keyB := roachpb.Key("a"), roachpb.Key("b")
 	keyC, keyD := roachpb.Key("c"), roachpb.Key("d")
 

pkg/kv/kvserver/client_replica_test.go

@@ -1011,21 +1011,52 @@ func TestTxnReadWithinUncertaintyIntervalAfterRangeMerge(t *testing.T) {
 		// writing.
 		manuals[2].Increment(2000)
 
+		defer func() {
+			if !t.Failed() {
+				return
+			}
+			t.Logf("maxNanos=%d", maxNanos)
+			t.Logf("manuals[2]=%d", manuals[2].UnixNano())
+		}()
+
 		// Write the data from a different transaction to establish the time for the
 		// key as 10 ns in the future.
-		_, pErr := kv.SendWrapped(ctx, tc.Servers[2].DistSenderI().(kv.Sender), putArgs(keyC, []byte("value")))
-		require.Nil(t, pErr)
+		{
+			ctx, fagrs := tracing.ContextWithRecordingSpan(ctx, tc.Servers[0].Tracer(), "keyC-write")
+			resp, pErr := kv.SendWrapped(ctx, tc.Servers[2].DistSenderI().(kv.Sender), putArgs(keyC, []byte("value")))
+			rec := fagrs()
+			require.Nil(t, pErr, "%v", rec)
+			defer func() {
+				if !t.Failed() {
+					return
+				}
+				t.Logf("keyC-write: %+v", resp)
+				t.Logf("keyC-write: %s", rec)
+			}()
+		}
 
 		// Create two identical transactions. The first one will perform a read to a
 		// store before the merge, the second will only read after the merge.
 		txn := roachpb.MakeTransaction("txn1", keyA, isolation.Serializable, 1, now, maxOffset, instanceId, 0, false /* omitInRangefeeds */)
 		txn2 := roachpb.MakeTransaction("txn2", keyA, isolation.Serializable, 1, now, maxOffset, instanceId, 0, false /* omitInRangefeeds */)
 
-		// Simulate a read which will cause the observed time to be set to now
-		resp, pErr := kv.SendWrappedWith(ctx, tc.Servers[1].DistSenderI().(kv.Sender), kvpb.Header{Txn: &txn}, getArgs(keyA))
-		require.Nil(t, pErr)
-		// The client needs to update its transaction to the returned transaction which has observed timestamps in it
-		txn = *resp.Header().Txn
+		// Simulate a read which will cause the observed time to be set to now.
+		{
+			ctx, fagrs := tracing.ContextWithRecordingSpan(ctx, tc.Servers[0].Tracer(), "txn1-keyA-get")
+			resp, pErr := kv.SendWrappedWith(ctx, tc.Servers[1].DistSenderI().(kv.Sender), kvpb.Header{Txn: &txn}, getArgs(keyA))
+			rec := fagrs()
+			require.Nil(t, pErr, "%v", rec)
+			// The client needs to update its transaction to the returned transaction which has observed timestamps in it
+			txn = *resp.Header().Txn
+			defer func() {
+				if !t.Failed() {
+					return
+				}
+				t.Logf("txn1-keyA-get: %+v", resp)
+				t.Logf("txn1-keyA-get: %s", rec)
+				t.Logf("txn1-keyA-get resp txn: %s", txn)
+			}()
+		}
 
 		// Now move the ranges, being careful not to move either leaseholder
 		// C: 3 (RHS - replica) => 0 (LHS leaseholder)
@@ -1048,8 +1079,13 @@ func TestTxnReadWithinUncertaintyIntervalAfterRangeMerge(t *testing.T) {
 
 		// Try and read the transaction from the context of a new transaction. This
 		// will fail as expected as the observed timestamp will not be set.
-		_, pErr = kv.SendWrappedWith(ctx, tc.Servers[0].DistSenderI().(kv.Sender), kvpb.Header{Txn: &txn2}, getArgs(keyC))
-		require.IsType(t, &kvpb.ReadWithinUncertaintyIntervalError{}, pErr.GetDetail())
+		{
+			ctx, fagrs := tracing.ContextWithRecordingSpan(ctx, tc.Servers[0].Tracer(), "txn2get-should-rwue")
+			_, pErr := kv.SendWrappedWith(ctx, tc.Servers[0].DistSenderI().(kv.Sender), kvpb.Header{Txn: &txn2},
+				getArgs(keyC))
+			rec := fagrs()
+			require.IsType(t, &kvpb.ReadWithinUncertaintyIntervalError{}, pErr.GetDetail(), "%s", rec)
+		}
 
 		// Try and read the key from the existing transaction. This should fail the
 		// same way.
@@ -1058,8 +1094,12 @@ func TestTxnReadWithinUncertaintyIntervalAfterRangeMerge(t *testing.T) {
 		// - Other error (Bad) - We expect an uncertainty error so the client can choose a new timestamp and retry.
 		// - Not found (Bad) - Error because the data was written before us.
 		// - Found (Bad) - The write HLC timestamp is after our timestamp.
-		_, pErr = kv.SendWrappedWith(ctx, tc.Servers[0].DistSenderI().(kv.Sender), kvpb.Header{Txn: &txn}, getArgs(keyC))
-		require.IsType(t, &kvpb.ReadWithinUncertaintyIntervalError{}, pErr.GetDetail())
+		{
+			ctx, fagrs := tracing.ContextWithRecordingSpan(ctx, tc.Servers[0].Tracer(), "txn1get-should-rwue")
+			_, pErr := kv.SendWrappedWith(ctx, tc.Servers[0].DistSenderI().(kv.Sender), kvpb.Header{Txn: &txn}, getArgs(keyC))
+			rec := fagrs()
+			require.IsType(t, &kvpb.ReadWithinUncertaintyIntervalError{}, pErr.GetDetail(), "%s", rec)
+		}
 	}
 
 	testutils.RunTrueAndFalse(t, "alignLeaseholders", func(t *testing.T, alignLeaseholders bool) {

pkg/sql/backfill.go

@@ -1012,8 +1012,6 @@ func (sc *SchemaChanger) distIndexBackfill(
 		log.Infof(ctx, "writing at persisted safe write time %v...", writeAsOf)
 	}
 
-	readAsOf := sc.clock.Now()
-
 	var p *PhysicalPlan
 	var extEvalCtx extendedEvalContext
 	var planCtx *PlanningCtx
@@ -1045,7 +1043,7 @@ func (sc *SchemaChanger) distIndexBackfill(
 		)
 		indexBatchSize := indexBackfillBatchSize.Get(&sc.execCfg.Settings.SV)
 		chunkSize := sc.getChunkSize(indexBatchSize)
-		spec, err := initIndexBackfillerSpec(*tableDesc.TableDesc(), writeAsOf, readAsOf, writeAtRequestTimestamp, chunkSize, addedIndexes, 0 /* sourceIndexID*/)
+		spec, err := initIndexBackfillerSpec(*tableDesc.TableDesc(), writeAsOf, writeAtRequestTimestamp, chunkSize, addedIndexes, 0)
 		if err != nil {
 			return err
 		}

pkg/sql/backfill_protected_timestamp_test.go

@@ -24,8 +24,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigptsreader"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
-	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
@@ -118,7 +116,7 @@ func TestValidationWithProtectedTS(t *testing.T) {
 	}
 	for _, sql := range []string{
 		"SET CLUSTER SETTING sql.stats.automatic_collection.enabled = false",
-		"ALTER DATABASE defaultdb CONFIGURE ZONE USING gc.ttlseconds = 1",
+		"ALTER DATABASE defaultdb CONFIGURE ZONE USING gc.ttlseconds = 3",
 		"CREATE TABLE t(n int)",
 		"ALTER TABLE t CONFIGURE ZONE USING range_min_bytes = 0, range_max_bytes = 67108864, gc.ttlseconds = 1",
 		"INSERT INTO t(n) SELECT * FROM generate_series(1, 250000)",
@@ -223,10 +221,10 @@ func TestValidationWithProtectedTS(t *testing.T) {
 	}
 }
 
-// TestBackfillWithProtectedTS runs operations that backfill into a table and
+// TestBackfillWithProtectedTS runs a query backfill into a table and
 // confirms that a protected timestamp is setup. It also confirms that if the
 // protected timestamp is not ready in time we do not infinitely retry.
-func TestBackfillWithProtectedTS(t *testing.T) {
+func TestBackfillQueryWithProtectedTS(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
@@ -239,7 +237,6 @@ func TestBackfillWithProtectedTS(t *testing.T) {
 	backfillQueryResume := make(chan struct{})
 	blockBackFillsForPTSFailure := atomic.Bool{}
 	blockBackFillsForPTSCheck := atomic.Bool{}
-	usingDeclarativeSchemaChanger := atomic.Bool{}
 	var s serverutils.TestServerInterface
 	var db *gosql.DB
 	var tableID uint32
@@ -248,46 +245,11 @@ func TestBackfillWithProtectedTS(t *testing.T) {
 			SQLEvalContext: &eval.TestingKnobs{
 				ForceProductionValues: true,
 			},
-			SQLDeclarativeSchemaChanger: &scexec.TestingKnobs{
-				RunBeforeBackfill: func() error {
-					// Cause the backfill to pause before adding the protected
-					// timestamp. This knob is for testing schema changes that
-					// are on the declarative schema changer.
-					if blockBackFillsForPTSFailure.Load() && usingDeclarativeSchemaChanger.Load() {
-						if !blockBackFillsForPTSFailure.Swap(false) {
-							return nil
-						}
-						backfillQueryWait <- struct{}{}
-						<-backfillQueryResume
-					}
-					return nil
-				},
-			},
-			DistSQL: &execinfra.TestingKnobs{
-				RunBeforeBackfillChunk: func(sp roachpb.Span) error {
-					// Cause the backfill to pause after it already began running
-					// and has installed a protected timestamp. This knob is for
-					// testing schema changes that use the index backfiller.
-					if blockBackFillsForPTSCheck.Load() && usingDeclarativeSchemaChanger.Load() {
-						_, prefix, err := s.Codec().DecodeTablePrefix(sp.Key)
-						if err != nil || prefix != tableID {
-							//nolint:returnerrcheck
-							return nil
-						}
-						if !blockBackFillsForPTSCheck.Swap(false) {
-							return nil
-						}
-						backfillQueryWait <- struct{}{}
-						<-backfillQueryResume
-					}
-					return nil
-				},
-			},
 			SQLSchemaChanger: &sql.SchemaChangerTestingKnobs{
 				RunBeforeQueryBackfill: func() error {
 					// Cause the backfill to pause before adding the protected
 					// timestamp. This knob is for testing CREATE MATERIALIZED VIEW.
-					if blockBackFillsForPTSFailure.Load() && !usingDeclarativeSchemaChanger.Load() {
+					if blockBackFillsForPTSFailure.Load() {
 						if !blockBackFillsForPTSFailure.Swap(false) {
 							return nil
 						}
@@ -302,7 +264,7 @@ func TestBackfillWithProtectedTS(t *testing.T) {
 					// Detect the first scan on table from the backfill, which is
 					// after the PTS has been set up. This knob is for testing CREATE
 					// MATERIALIZED VIEW.
-					if blockBackFillsForPTSCheck.Load() && !usingDeclarativeSchemaChanger.Load() &&
+					if blockBackFillsForPTSCheck.Load() &&
 						request.Txn != nil &&
 						request.Txn.Name == "schemaChangerBackfill" &&
 						request.Requests[0].GetInner().Method() == kvpb.Scan {
@@ -380,35 +342,23 @@ func TestBackfillWithProtectedTS(t *testing.T) {
 	const rowsAddedPerIteration = 1
 
 	for _, tc := range []struct {
-		name                   string
-		tableName              string
-		backfillSchemaChange   string
-		jobDescriptionPrefix   string
-		postTestQuery          string
-		expectedCount          int
-		usingDeclSchemaChanger bool
+		name                 string
+		tableName            string
+		backfillSchemaChange string
+		jobDescriptionPrefix string
+		postTestQuery        string
+		expectedCount        int
 	}{
 		{
-			name:                   "create materialized view",
-			tableName:              "t_mat_view",
-			backfillSchemaChange:   "CREATE MATERIALIZED VIEW test AS (SELECT n from t_mat_view)",
-			jobDescriptionPrefix:   "CREATE MATERIALIZED VIEW",
-			postTestQuery:          "SELECT count(*) FROM test",
-			expectedCount:          initialRowCount - rowsDeletedPerIteration + rowsAddedPerIteration,
-			usingDeclSchemaChanger: false,
-		},
-		{
-			name:                   "create index",
-			tableName:              "t_idx",
-			backfillSchemaChange:   "CREATE INDEX idx ON t_idx(n)",
-			jobDescriptionPrefix:   "CREATE INDEX idx",
-			postTestQuery:          "SELECT count(*) FROM t_idx@idx",
-			expectedCount:          initialRowCount - 2*rowsDeletedPerIteration + 2*rowsAddedPerIteration,
-			usingDeclSchemaChanger: true,
+			name:                 "create materialized view",
+			tableName:            "t_mat_view",
+			backfillSchemaChange: "CREATE MATERIALIZED VIEW test AS (SELECT n from t_mat_view)",
+			jobDescriptionPrefix: "CREATE MATERIALIZED VIEW",
+			postTestQuery:        "SELECT count(*) FROM test",
+			expectedCount:        initialRowCount - rowsDeletedPerIteration + rowsAddedPerIteration,
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			usingDeclarativeSchemaChanger.Store(tc.usingDeclSchemaChanger)
 			for _, sql := range []string{
 				fmt.Sprintf("CREATE TABLE %s(n int primary key)", tc.tableName),
 				fmt.Sprintf("ALTER TABLE %s CONFIGURE ZONE USING range_min_bytes = 0, range_max_bytes = 67108864, gc.ttlseconds = 5", tc.tableName),

pkg/sql/distsql_plan_backfill.go

@@ -40,7 +40,7 @@ func initColumnBackfillerSpec(
 
 func initIndexBackfillerSpec(
 	desc descpb.TableDescriptor,
-	writeAsOf, readAsOf hlc.Timestamp,
+	writeAsOf hlc.Timestamp,
 	writeAtBatchTimestamp bool,
 	chunkSize int64,
 	indexesToBackfill []descpb.IndexID,
@@ -50,7 +50,6 @@ func initIndexBackfillerSpec(
 		Table:                 desc,
 		WriteAsOf:             writeAsOf,
 		WriteAtBatchTimestamp: writeAtBatchTimestamp,
-		ReadAsOf:              readAsOf,
 		Type:                  execinfrapb.BackfillerSpec_Index,
 		ChunkSize:             chunkSize,
 		IndexesToBackfill:     indexesToBackfill,

pkg/sql/execinfrapb/processors_bulk_io.proto

@@ -61,9 +61,9 @@ message BackfillerSpec {
   optional uint64 update_chunk_size_threshold_bytes = 14 [(gogoproto.nullable) = false];
 
   // WriteAsOf is the time that the backfill entries should be written.
-  // Note: Older nodes may also use this as the read time instead of readAsOf.
   optional util.hlc.Timestamp writeAsOf = 7 [(gogoproto.nullable) = false];
   // The timestamp to perform index backfill historical scans at.
+  // This is only used by the column backfiller.
   optional util.hlc.Timestamp readAsOf = 9 [(gogoproto.nullable) = false];
 
   // IndexesToBackfill is the set of indexes to backfill. This is populated only

pkg/sql/index_backfiller.go

@@ -23,7 +23,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
-	"github.com/cockroachdb/errors"
 )
 
 // IndexBackfillPlanner holds dependencies for an index backfiller
@@ -44,23 +43,19 @@ func (ib *IndexBackfillPlanner) MaybePrepareDestIndexesForBackfill(
 	if !current.MinimumWriteTimestamp.IsEmpty() {
 		return current, nil
 	}
-	// Pick an arbitrary read timestamp for the reads of the backfill.
-	// It's safe to use any timestamp to read even if we've partially backfilled
-	// at an earlier timestamp because other writing transactions have been
-	// writing at the appropriate timestamps in-between.
-	backfillReadTimestamp := ib.execCfg.Clock.Now()
+	minWriteTimestamp := ib.execCfg.Clock.Now()
 	targetSpans := make([]roachpb.Span, len(current.DestIndexIDs))
 	for i, idxID := range current.DestIndexIDs {
 		targetSpans[i] = td.IndexSpan(ib.execCfg.Codec, idxID)
 	}
 	if err := scanTargetSpansToPushTimestampCache(
-		ctx, ib.execCfg.DB, backfillReadTimestamp, targetSpans,
+		ctx, ib.execCfg.DB, minWriteTimestamp, targetSpans,
 	); err != nil {
 		return scexec.BackfillProgress{}, err
 	}
 	return scexec.BackfillProgress{
 		Backfill:              current.Backfill,
-		MinimumWriteTimestamp: backfillReadTimestamp,
+		MinimumWriteTimestamp: minWriteTimestamp,
 	}, nil
 }
 
@@ -72,19 +67,6 @@ func (ib *IndexBackfillPlanner) BackfillIndexes(
 	job *jobs.Job,
 	descriptor catalog.TableDescriptor,
 ) (retErr error) {
-	// Potentially install a protected timestamp before the GC interval is hit,
-	// which can help avoid transaction retry errors, with shorter GC intervals.
-	protectedTimestampCleaner := ib.execCfg.ProtectedTimestampManager.TryToProtectBeforeGC(ctx,
-		job,
-		descriptor,
-		progress.MinimumWriteTimestamp)
-	defer func() {
-		cleanupError := protectedTimestampCleaner(ctx)
-		if cleanupError != nil {
-			retErr = errors.CombineErrors(retErr, cleanupError)
-		}
-	}()
-
 	var completed = struct {
 		syncutil.Mutex
 		g roachpb.SpanGroup
@@ -117,12 +99,17 @@ func (ib *IndexBackfillPlanner) BackfillIndexes(
 		return nil
 	}
 	now := ib.execCfg.DB.Clock().Now()
+	// Pick now as the read timestamp for the backfill. It's safe to use this
+	// timestamp to read even if we've partially backfilled at an earlier
+	// timestamp because other writing transactions have been writing at the
+	// appropriate timestamps in-between.
+	readAsOf := now
 	run, retErr := ib.plan(
 		ctx,
 		descriptor,
 		now,
 		progress.MinimumWriteTimestamp,
-		progress.MinimumWriteTimestamp,
+		readAsOf,
 		spansToDo,
 		progress.DestIndexIDs,
 		progress.SourceIndexID,
@@ -197,9 +184,8 @@ func (ib *IndexBackfillPlanner) plan(
 		chunkSize := indexBackfillBatchSize.Get(&ib.execCfg.Settings.SV)
 		const writeAtRequestTimestamp = true
 		spec, err := initIndexBackfillerSpec(
-			*td.TableDesc(), writeAsOf, readAsOf, writeAtRequestTimestamp, chunkSize,
-			indexesToBackfill,
-			sourceIndexID,
+			*td.TableDesc(), writeAsOf, writeAtRequestTimestamp, chunkSize,
+			indexesToBackfill, sourceIndexID,
 		)
 		if err != nil {
 			return err