row: harden inconsistent scan machinery

Table statistics collection uses inconsistent scan machinery which means
that we paginate over the table via separate transactions with
constantly advancing read timestamps. In particular, once we determine
that the current read timestamp is at least 5 minutes old (controlled
via `sql.stats.max_timestamp_age` cluster setting), we commit the
current txn and open a new one in which we advance the timestamp by the
amount of time that has passed since the start of the previous one (i.e.
by the duration of the previous txn).

This process requires an "initial timestamp" for the very first txn that
we use. That value comes from evaluating AS OF SYSTEM TIME clause of
CREATE STATISTICS stmt which happens when the job record is created.
Previously, if there was a long delay between the job record creation
and the job being actually executed, we would return "cannot specify
timestamp older than ..." error before creating the first txn. It's
unclear to me why this check was put in place since it's not really
necessary - we could simply remove it, which would make the very
first txn of the inconsistent scan to be committed right away, and
things would proceed easily.

This commit fixes this problem (by removing the check) but also improves
things a bit further by explicitly advancing the "initial timestamp"
before the first txn is open. If the initial timestamp is too old, it is
advanced to make its age to be 1/10 of the max timestamp age. An example
of the txn cycle:
```
  current time:      100
  initial timestamp: 50
  max timestamp age: 10

  time
  100:     advance initial timestamp to 99
  100:     start scan, timestamp=99
  100-109: continue scanning at timestamp=99
  109:     bump timestamp to 108
  109-118: continue scanning at timestamp=108
  118:     bump timestamp to 117
  118-127: continue scanning at timestamp=117
```

Release note (bug fix): CockroachDB could previously encounter "cannot
specify timestamp older than ..." error during the table statistics
collection in some cases (like when the cluster is overloaded), and this
is now fixed. The bug has been present since 19.1 version.

Author: yuzefovich

pkg/sql/create_stats_test.go

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -37,11 +38,19 @@ func TestStatsWithLowTTL(t *testing.T) {
 	// The test depends on reasonable timings, so don't run under race.
 	skip.UnderRace(t)
 
+	var blockTableReader atomic.Bool
+	blockCh := make(chan struct{})
+
 	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{
 		Knobs: base.TestingKnobs{
 			DistSQL: &execinfra.TestingKnobs{
 				// Set the batch size small to avoid having to use a large number of rows.
 				TableReaderBatchBytesLimit: 100,
+				TableReaderStartScanCb: func() {
+					if blockTableReader.Load() {
+						<-blockCh
+					}
+				},
 			},
 		},
 	})
@@ -131,6 +140,15 @@ SET CLUSTER SETTING sql.stats.automatic_collection.enabled = false;
 	// Set up timestamp advance to keep timestamps no older than 1s.
 	r.Exec(t, `SET CLUSTER SETTING sql.stats.max_timestamp_age = '1s'`)
 
+	// Block start of the inconsistent scan for 2s so that the initial timestamp
+	// becomes way too old.
+	blockTableReader.Store(true)
+	go func() {
+		defer blockTableReader.Store(false)
+		time.Sleep(2 * time.Second)
+		close(blockCh)
+	}()
+
 	_, err := db.Exec(`CREATE STATISTICS foo FROM t AS OF SYSTEM TIME '-0.1s'`)
 	if err != nil {
 		t.Error(err)

pkg/sql/distsql_plan_stats.go

@@ -71,6 +71,7 @@ var maxTimestampAge = settings.RegisterDurationSetting(
 	"sql.stats.max_timestamp_age",
 	"maximum age of timestamp during table statistics collection",
 	5*time.Minute,
+	// TODO(yuzefovich): we should add non-negative duration validation.
 )
 
 // minAutoHistogramSamples and maxAutoHistogramSamples are the bounds used by

pkg/sql/execinfra/server_config.go

@@ -337,6 +337,10 @@ type TestingKnobs struct {
 	// will be assigned to the gateway before we start assigning partitions to
 	// other nodes.
 	MinimumNumberOfGatewayPartitions int
+
+	// TableReaderStartScanCb, when non-nil, will be called whenever the
+	// TableReader processor starts its scan.
+	TableReaderStartScanCb func()
 }
 
 // ModuleTestingKnobs is part of the base.ModuleTestingKnobs interface.

pkg/sql/execinfrapb/processors_sql.proto

@@ -102,9 +102,8 @@ message TableReaderSpec {
   //     50:    bump timestamp to 40
   //     ...
   //
-  // Note: it is an error to perform a historical read at an initial timestamp
-  // older than this value.
-  //
+  // Note: if initial timestamp is older than this value when the scan begins,
+  // the timestamp will be advanced forward.
   optional uint64 max_timestamp_age_nanos = 9 [(gogoproto.nullable) = false];
 
   // Indicates the row-level locking strength to be used by the scan. If set to

pkg/sql/row/fetcher.go

@@ -637,14 +637,28 @@ func (rf *Fetcher) StartInconsistentScan(
 		return errors.AssertionFailedf("no spans")
 	}
 
-	txnTimestamp := initialTimestamp
 	txnStartTime := timeutil.Now()
-	if txnStartTime.Sub(txnTimestamp.GoTime()) >= maxTimestampAge {
-		return errors.Errorf(
-			"AS OF SYSTEM TIME: cannot specify timestamp older than %s for this operation",
-			maxTimestampAge,
-		)
+	if txnStartTime.Sub(initialTimestamp.GoTime()) >= maxTimestampAge {
+		// The initial timestamp is too far into the past already (which can
+		// happen when the cluster is overloaded, or there was a delay in the
+		// stats job being picked up for execution, or some other reason). In
+		// such case we'll advance the timestamp so that its age is about 1/10
+		// of the maximum age.
+		targetTimestampAge := maxTimestampAge.Nanoseconds() / 10
+		if targetTimestampAge < int64(100*time.Millisecond) {
+			// Ensure at least 100ms timestamp age. We shouldn't reach this line
+			// unless the max timestamp age setting is set way too low (or
+			// negative, by mistake), so we're just being conservative.
+			targetTimestampAge = int64(100 * time.Millisecond)
+		}
+		advanceBy := txnStartTime.Sub(initialTimestamp.GoTime()).Nanoseconds() - targetTimestampAge
+		if log.V(1) {
+			log.Infof(ctx, "initial timestamp %v too far into the past, advancing it by %v", initialTimestamp, advanceBy)
+		}
+		initialTimestamp = initialTimestamp.Add(advanceBy, 0 /* logical */)
 	}
+
+	txnTimestamp := initialTimestamp
 	txn := kv.NewTxnWithSteppingEnabled(ctx, db, 0 /* gatewayNodeID */, qualityOfService)
 	if err := txn.SetFixedTimestamp(ctx, txnTimestamp); err != nil {
 		return err
@@ -691,7 +705,7 @@ func (rf *Fetcher) StartInconsistentScan(
 	}
 
 	if err := rf.kvFetcher.SetupNextFetch(
-		ctx, spans, nil, batchBytesLimit,
+		ctx, spans, nil /* spanIDs */, batchBytesLimit,
 		rf.rowLimitToKeyLimit(rowLimitHint), false, /* spansCanOverlap */
 	); err != nil {
 		return err

pkg/sql/rowexec/tablereader.go

@@ -205,6 +205,9 @@ func (tr *tableReader) Start(ctx context.Context) {
 }
 
 func (tr *tableReader) startScan(ctx context.Context) error {
+	if cb := tr.FlowCtx.Cfg.TestingKnobs.TableReaderStartScanCb; cb != nil {
+		cb()
+	}
 	limitBatches := !tr.parallelize
 	var bytesLimit rowinfra.BytesLimit
 	if !limitBatches {