sql: Deflake TestTxnObeysTableModificationTime

This commit rewrites this test and a local execution with
`--race --stress` succeeded for at least 10 minutes.

Release note: None

Author: Xiang-Gu

pkg/sql/catalog/lease/BUILD.bazel

@@ -77,6 +77,7 @@ go_test(
     embed = [":lease"],
     deps = [
         "//pkg/base",
+        "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv",
@@ -105,7 +106,6 @@ go_test(
         "//pkg/sql/rowenc/keyside",
         "//pkg/sql/sem/tree",
         "//pkg/sql/sessiondata",
-        "//pkg/sql/sqltestutils",
         "//pkg/sql/tests",
         "//pkg/sql/types",
         "//pkg/storage",
@@ -114,7 +114,6 @@ go_test(
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
-        "//pkg/util",
         "//pkg/util/admission",
         "//pkg/util/encoding",
         "//pkg/util/hlc",

pkg/sql/catalog/lease/lease_test.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/cockroachdb/cockroach-go/v2/crdb"
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
@@ -49,15 +50,13 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc/keyside"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
-	"github.com/cockroachdb/cockroach/pkg/sql/sqltestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
-	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -981,219 +980,105 @@ func TestTxnObeysTableModificationTime(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
 	params := createTestServerParams()
-	params.ScanMaxIdleTime = time.Millisecond
-	s, sqlDB, kvDB := serverutils.StartServer(t, params)
+	params.Knobs.JobsTestingKnobs = jobs.NewTestingKnobsWithShortIntervals()
+	s, sqlDB, _ := serverutils.StartServer(t, params)
 	defer s.Stopper().Stop(context.Background())
 
-	// Disable strict GC TTL enforcement because we're going to shove a zero-value
-	// TTL into the system with AddImmediateGCZoneConfig.
-	defer sqltestutils.DisableGCTTLStrictEnforcement(t, sqlDB)()
-
-	_, err := sqlDB.Exec(`SET CLUSTER SETTING sql.gc_job.wait_for_gc.interval = '1s';`)
-	require.NoError(t, err)
-
-	// Refresh protected timestamp cache immediately to make MVCC GC queue to
-	// process GC immediately.
-	_, err = sqlDB.Exec(`SET CLUSTER SETTING kv.protectedts.poll_interval = '1s';`)
-	require.NoError(t, err)
-
-	// This test intentionally relies on uncontended transactions not being pushed
-	// in order to verify what it claims to verify. The default closed timestamp
-	// interval in 20.1+ is 3s. When run under the race detector, the process can
-	// stall for upwards of 3s leading to the write transaction getting pushed.
-	//
-	// In order to mitigate that push, we increase the target_duration when the
-	// test is run under race.
-	if util.RaceEnabled || skip.Stress() {
-		_, err := sqlDB.Exec(
-			"SET CLUSTER SETTING kv.closed_timestamp.target_duration = '120s'")
-		require.NoError(t, err)
-	}
-
-	if _, err := sqlDB.Exec(`
+	_, err := sqlDB.Exec(`
 CREATE DATABASE t;
 CREATE TABLE t.kv (k CHAR PRIMARY KEY, v CHAR);
 INSERT INTO t.kv VALUES ('a', 'b');
-`); err != nil {
-		t.Fatal(err)
-	}
-	tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, keys.SystemSQLCodec, "t", "kv")
-
-	// A read-write transaction that uses the old version of the descriptor.
-	txReadWrite, err := sqlDB.Begin()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// A read-only transaction that uses the old version of the descriptor.
-	txRead, err := sqlDB.Begin()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// A write-only transaction that uses the old version of the descriptor.
-	txWrite, err := sqlDB.Begin()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// Modify the table descriptor.
-	if _, err := sqlDB.Exec(`ALTER TABLE t.kv ADD m CHAR DEFAULT 'z';`); err != nil {
-		t.Fatal(err)
-	}
-
-	rows, err := txReadWrite.Query(`SELECT * FROM t.kv`)
-	if err != nil {
-		t.Fatal(err)
-	}
+`)
+	require.NoError(t, err)
 
-	checkSelectResults := func(rows *gosql.Rows) {
-		defer func() {
-			if err := rows.Close(); err != nil {
-				t.Fatal(err)
-			}
-		}()
-		for rows.Next() {
-			// The transaction is unable to see column m.
+	// requireOneRow ensures `res` contains only one row with two
+	// columns ("a", "b").
+	requireOneRow := func(res *gosql.Rows) {
+		for res.Next() {
 			var k, v, m string
-			if err := rows.Scan(&k, &v, &m); !testutils.IsError(
-				err, "expected 2 destination arguments in Scan, not 3",
-			) {
-				t.Fatalf("err = %v", err)
-			}
-			err = rows.Scan(&k, &v)
-			if err != nil {
-				t.Fatal(err)
-			}
-			if k != "a" || v != "b" {
-				t.Fatalf("didn't find expected row: %s %s", k, v)
-			}
+			require.Error(t, res.Scan(&k, &v, &m))
+			require.NoError(t, res.Scan(&k, &v))
+			require.Equal(t, "a", k)
+			require.Equal(t, "b", v)
 		}
+		require.NoError(t, res.Close())
 	}
 
-	checkSelectResults(rows)
-
-	rows, err = txRead.Query(`SELECT * FROM t.kv`)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	checkSelectResults(rows)
-
-	// Read-only transaction commits just fine.
-	if err := txRead.Commit(); err != nil {
-		t.Fatal(err)
-	}
-
-	// This INSERT will cause the transaction to be pushed,
-	// which will be detected when we attempt to Commit() below.
-	if _, err := txReadWrite.Exec(`INSERT INTO t.kv VALUES ('c', 'd');`); err != nil {
-		t.Fatal(err)
-	}
-
-	checkDeadlineErr := func(err error, t *testing.T) {
+	// A helper to assert err is a transaction restart error with an error string
+	// that matches the supplied regex.
+	requireRestartTransactionErrWithMsg := func(t *testing.T, err error, re string) {
 		var pqe (*pq.Error)
 		if !errors.As(err, &pqe) || pgcode.MakeCode(string(pqe.Code)) != pgcode.SerializationFailure ||
-			!testutils.IsError(err, "RETRY_COMMIT_DEADLINE_EXCEEDED") {
-			t.Fatalf("expected deadline exceeded, got: %v", err)
+			!testutils.IsError(err, re) {
+			t.Fatalf("expected a %v error, got: %v", re, err)
 		}
 	}
 
-	// The transaction read at one timestamp and wrote at another so it
-	// has to be restarted because the spans read were modified by the backfill.
-	checkDeadlineErr(txReadWrite.Commit(), t)
-
-	// This INSERT will cause the transaction to be pushed transparently,
-	// which will be detected when we attempt to Commit() below only because
-	// a deadline has been set.
-	if _, err := txWrite.Exec(`INSERT INTO t.kv VALUES ('c', 'd');`); err != nil {
-		t.Fatal(err)
-	}
-
-	checkDeadlineErr(txWrite.Commit(), t)
-
-	// Test the deadline exceeded error with a CREATE/DROP INDEX.
-	txWrite, err = sqlDB.Begin()
-	if err != nil {
-		t.Fatal(err)
-	}
-	txUpdate, err := sqlDB.Begin()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// Modify the table descriptor.
-	if _, err := sqlDB.Exec(`CREATE INDEX foo ON t.kv (v)`); err != nil {
-		t.Fatal(err)
-	}
-
-	// This INSERT will cause the transaction to be pushed transparently,
-	// which will be detected when we attempt to Commit() below only because
-	// a deadline has been set.
-	if _, err := txWrite.Exec(`INSERT INTO t.kv VALUES ('c', 'd');`); err != nil {
-		t.Fatal(err)
+	// requireWriteTooOldErr ensures `err` is a WriteTooOldError.
+	requireWriteTooOldErr := func(t *testing.T, err error) {
+		requireRestartTransactionErrWithMsg(t, err, "WriteTooOldError")
 	}
 
-	checkDeadlineErr(txWrite.Commit(), t)
-
-	if _, err := txUpdate.Exec(`UPDATE t.kv SET v = 'c' WHERE k = 'a';`); err != nil {
-		t.Fatal(err)
+	// requireSessionExpiredErr ensures `err` is a liveness session expired error.
+	requireSessionExpiredErr := func(t *testing.T, err error) {
+		requireRestartTransactionErrWithMsg(t, err, "liveness session expired")
 	}
 
-	checkDeadlineErr(txUpdate.Commit(), t)
-
-	txWrite, err = sqlDB.Begin()
-	if err != nil {
-		t.Fatal(err)
-	}
-	txRead, err = sqlDB.Begin()
-	if err != nil {
-		t.Fatal(err)
-	}
+	// A read-write transaction that uses the old version of the descriptor.
+	txReadWrite, err := sqlDB.Begin()
+	require.NoError(t, err)
+	// A read-only transaction that uses the old version of the descriptor.
+	txRead, err := sqlDB.Begin()
+	require.NoError(t, err)
+	// A write-only transaction that uses the old version of the descriptor.
+	txWrite, err := sqlDB.Begin()
+	require.NoError(t, err)
 
 	// Modify the table descriptor.
-	if _, err := sqlDB.Exec(`DROP INDEX t.kv@foo`); err != nil {
-		t.Fatal(err)
-	}
-
-	rows, err = txRead.Query(`SELECT k, v FROM t.kv@foo`)
-	if err != nil {
-		t.Fatal(err)
-	}
-	checkSelectResults(rows)
-
-	// Uses old descriptor and inserts values into index span which
-	// will be cleaned up.
-	if _, err := txWrite.Exec(`INSERT INTO t.kv VALUES ('c', 'd');`); err != nil {
-		t.Fatal(err)
-	}
-
-	if err := txRead.Commit(); err != nil {
-		t.Fatal(err)
-	}
-	if err := txWrite.Commit(); err != nil {
-		t.Fatal(err)
-	}
-
-	tableSpan := tableDesc.TableSpan(keys.SystemSQLCodec)
-
-	// Allow async schema change waiting for GC to complete (when dropping an
-	// index) and clear the index keys.
-	if _, err := sqltestutils.AddImmediateGCZoneConfig(sqlDB, tableDesc.GetID()); err != nil {
-		t.Fatal(err)
-	}
-
-	// Reset closed timestamp so that deleted keys can be GC'ed within the
-	// SucceedSoon window.
-	if util.RaceEnabled || skip.Stress() {
-		_, err := sqlDB.Exec(
-			"SET CLUSTER SETTING kv.closed_timestamp.target_duration = '3s'")
-		require.NoError(t, err)
-	}
-
-	testutils.SucceedsSoon(t, func() error {
-		return tests.CheckKeyCountIncludingTombstonedE(t, s, tableSpan, 2)
-	})
+	_, err = sqlDB.Exec(`ALTER TABLE t.kv ADD m CHAR DEFAULT 'z';`)
+	require.NoError(t, err)
+	// Wait a short bit so that the SCHEMA CHANGE GC job created by the above ADD COLUMN
+	// gets to run, which will mark the old primary index of `t.kv` as tombstoned.
+	time.Sleep(1 * time.Second)
+
+	// Read-only transaction:
+	// 1). reads uses a historical version of `t.kv` and sees only two columns `k`
+	// and `v`;
+	// 2). it commits just fine;
+	rows, err := txRead.Query(`SELECT * FROM t.kv`)
+	require.NoError(t, err)
+	requireOneRow(rows)
+	require.NoError(t, txRead.Commit())
+
+	// Read-write transaction:
+	// 1). reads uses a historical version of `t.kv` and sees only two columns `k`
+	// and `v`;
+	// 2). writes to that historical version of `t.kv` attempt to write to the old
+	// primary index, which has previously been written with a higher timestamp
+	// (when the SCHEMA CHANGE GC job lays a range tombstone on it). This is a
+	// write-write conflict, and it will bump up transaction's write_ts, perform a
+	// read refresh which would fail (bc reading from that new and higher
+	// timestamp would return zero row, instead of one), and thus a
+	// TransactionRetryWriteTooOld error will be returned.
+	rows, err = txReadWrite.Query(`SELECT * FROM t.kv`)
+	require.NoError(t, err)
+	requireOneRow(rows)
+	_, err = txReadWrite.Exec(`INSERT INTO t.kv VALUES ('c', 'd');`)
+	requireWriteTooOldErr(t, err)
+	require.NoError(t, txReadWrite.Rollback())
+
+	// Write-only transaction:
+	// 1). writes, similarly to the read-write transaction, will trigger a read
+	// refresh and this time it will succeed (bc there is no reads).
+	// 2). commits, however, will update the transaction's deadline, and the
+	// leased descriptor such an "old" transaction is using is the very original
+	// version (v1) of `t.kv` that expires at the modification time of v2 of
+	// `t.kv`. But the bumped commit timestamp is larger than the leased
+	// descriptor's expiration time, leading to a "liveness session expired"
+	// error.
+	_, err = txWrite.Exec(`INSERT INTO t.kv VALUES ('c', 'd');`)
+	require.NoError(t, err)
+	requireSessionExpiredErr(t, txWrite.Commit())
 }
 
 // Test that a lease on a table descriptor is always acquired on the latest