stats: fix "non-nullable col with no value" for multiple column families

This commit fixes an edge case in the timestamp-advancing mechanism of
the inconsistent scans that are used by the table statistics collection.
In particular, previously on a table with multiple column families it
was possible for the scan to stop in the middle of a SQL row, and if
that row happens to exist at the old timestamp but to no longer exist at
the new timestamp, we'd hit an internal error. The bug is now fixed by
setting WholeRowsOfSize option of the BatchRequest if we're scanning
a table with multiple column families which guarantees that the scan at
each timestamp will stop only at the ends of SQL rows.

To reproduce this behavior I extended an existing test that stresses the
timestamp-advancing mechanism.

Release note (bug fix): Previously, on a table with multiple column
families CockroachDB could encounter "Non-nullable column "‹×›:‹×›"
with no value" error during table statistics collection in rare cases.
The bug has been present since v19.2 and is now fixed.

Author: yuzefovich

pkg/sql/create_stats_test.go

@@ -38,14 +38,20 @@ func TestStatsWithLowTTL(t *testing.T) {
 	// The test depends on reasonable timings, so don't run under race.
 	skip.UnderRace(t)
 
+	rng, _ := randutil.NewTestRand()
 	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,
+				// Set the batch size small to avoid having to use a large
+				// number of rows.
+				//
+				// We use a random bytes limit so that the scans have a chance
+				// to stop at different points within the SQL row (in case of
+				// multiple column families).
+				TableReaderBatchBytesLimit: 50 + int64(rng.Intn(100)),
 				TableReaderStartScanCb: func() {
 					if blockTableReader.Load() {
 						<-blockCh
@@ -57,51 +63,80 @@ func TestStatsWithLowTTL(t *testing.T) {
 	defer s.Stopper().Stop(context.Background())
 
 	r := sqlutils.MakeSQLRunner(db)
-	r.Exec(t, `
-SET CLUSTER SETTING sql.stats.automatic_collection.enabled = false;
-`)
-	r.Exec(t, `
-		CREATE DATABASE test;
-		USE test;
-		CREATE TABLE t (k INT PRIMARY KEY, a INT, b INT);
-	`)
-	const numRows = 20
-	r.Exec(t, `INSERT INTO t SELECT k, 2*k, 3*k FROM generate_series(0, $1) AS g(k)`, numRows-1)
+	r.Exec(t, `SET CLUSTER SETTING sql.stats.automatic_collection.enabled = false;`)
+	// Sometimes use single column family, sometimes use multiple.
+	if rng.Float64() < 0.5 {
+		r.Exec(t, `CREATE TABLE t (k INT PRIMARY KEY, a INT NOT NULL, b INT NOT NULL, FAMILY (k, a, b));`)
+	} else {
+		r.Exec(t, `CREATE TABLE t (k INT PRIMARY KEY, a INT NOT NULL, b INT NOT NULL, FAMILY (k), FAMILY (a), FAMILY (b));`)
+	}
+	const initialNumRows, maxNumRows = 20, 100
+	r.Exec(t, `INSERT INTO t SELECT k, 2*k, 3*k FROM generate_series(0, $1) AS g(k)`, initialNumRows-1)
 
-	// Start a goroutine that keeps updating rows in the table and issues
-	// GCRequests simulating a 2 second TTL. While this is running, reading at a
-	// timestamp older than 2 seconds will likely error out.
+	// Start a goroutine that keeps modifying rows (updating, deleting,
+	// inserting new ones) in the table and issues GCRequests simulating a 2
+	// second TTL. While this is running, reading at a timestamp older than 2
+	// seconds will likely error out.
 	var goroutineErr error
 	var wg sync.WaitGroup
 	wg.Add(1)
 	stopCh := make(chan struct{})
+	// onlyUpdates determines whether only UPDATE stmts are issued in the
+	// goroutine.
+	var onlyUpdates atomic.Bool
+	onlyUpdates.Store(true)
 
 	go func() {
 		defer wg.Done()
 
 		// Open a separate connection to the database.
 		db2 := s.SQLConn(t)
 
-		_, err := db2.Exec("USE test")
-		if err != nil {
-			goroutineErr = err
-			return
-		}
-		rng, _ := randutil.NewTestRand()
+		nextPK := initialNumRows
 		for {
 			select {
 			case <-stopCh:
 				return
 			default:
 			}
-			k := rng.Intn(numRows)
-			if _, err := db2.Exec(`UPDATE t SET a=a+1, b=b+2 WHERE k=$1`, k); err != nil {
-				goroutineErr = err
-				return
+			switch rng.Intn(4) {
+			case 0, 1:
+				// In 50% cases try to update an existing row (it's ok if the
+				// row doesn't exist).
+				k := rng.Intn(nextPK)
+				if _, err := db2.Exec(`UPDATE t SET a=a+1, b=b+2 WHERE k=$1`, k); err != nil {
+					goroutineErr = err
+					return
+				}
+			case 2:
+				if onlyUpdates.Load() {
+					continue
+				}
+				// In 25% cases try to delete a row (it's ok if the row doesn't
+				// exist).
+				k := rng.Intn(nextPK)
+				if _, err := db2.Exec(`DELETE FROM t WHERE k=$1`, k); err != nil {
+					goroutineErr = err
+					return
+				}
+			case 3:
+				if onlyUpdates.Load() {
+					continue
+				}
+				// In 25% cases insert a new row, but don't insert too many rows
+				// to allow for the stats collection to complete.
+				if nextPK == maxNumRows {
+					continue
+				}
+				if _, err := db2.Exec(`INSERT INTO t SELECT $1, 2*$1, 3*$1`, nextPK); err != nil {
+					goroutineErr = err
+					return
+				}
+				nextPK++
 			}
 			// Force a table GC of values older than 2 seconds.
 			if err := s.ForceTableGC(
-				context.Background(), "test", "t", s.Clock().Now().Add(-int64(2*time.Second), 0),
+				context.Background(), "defaultdb", "t", s.Clock().Now().Add(-int64(2*time.Second), 0),
 			); err != nil {
 				goroutineErr = err
 				return
@@ -139,6 +174,7 @@ 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'`)
+	onlyUpdates.Store(false)
 
 	// Block start of the inconsistent scan for 2s so that the initial timestamp
 	// becomes way too old.

pkg/sql/row/fetcher.go

@@ -702,6 +702,14 @@ func (rf *Fetcher) StartInconsistentScan(
 			}
 		}
 
+		if rf.args.Spec.MaxKeysPerRow > 1 {
+			// If the table has multiple column families, we need to ensure that
+			// the scan stops at the end of the full SQL row - otherwise, the
+			// row might be deleted between two timestamps leading to incorrect
+			// results (or internal errors).
+			ba.Header.WholeRowsOfSize = int32(rf.args.Spec.MaxKeysPerRow)
+		}
+
 		log.VEventf(ctx, 2, "inconsistent scan: sending a batch with %d requests", len(ba.Requests))
 		res, err := txn.Send(ctx, ba)
 		if err != nil {