sql: enforce schema_locked for TRUNCATE

Previously, we were completely ignoring schema_locked for the TRUNCATE
command, which could changefeeds to break. To address this, this patch
blocks TRUNCATE on schema_locked tables.

Fixes: #151941

Release note (bug fix): schema_locked was not enforced on the TRUNCATE
command, which could cause changefeed jobs to fail.

Author: fqazi

pkg/backup/backup_test.go

@@ -3603,17 +3603,21 @@ func TestRestoreAsOfSystemTime(t *testing.T) {
 	sqlDB.Exec(t, `ALTER TABLE data.bank ADD COLUMN points_balance INT DEFAULT 50`)
 	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&ts[5])
 
+	sqlDB.Exec(t, "ALTER TABLE data.bank SET (schema_locked=false)")
 	sqlDB.Exec(t, `TRUNCATE TABLE data.bank`)
 	sqlDB.Exec(t, `TRUNCATE TABLE data.bank`)
 	sqlDB.Exec(t, `TRUNCATE TABLE data.bank`)
+	sqlDB.Exec(t, "ALTER TABLE data.bank SET (schema_locked=true)")
 	sqlDB.Exec(t, `CREATE TABLE other.sometable AS SELECT * FROM data.sometable`)
 	sqlDB.Exec(t, `DROP TABLE data.sometable`)
 	sqlDB.Exec(t, `CREATE INDEX ON data.teller (name)`)
 	sqlDB.Exec(t, `INSERT INTO data.bank VALUES (2, 2), (4, 4)`)
 	sqlDB.Exec(t, `INSERT INTO data.teller VALUES (2, 'craig')`)
 	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&ts[6])
 
+	sqlDB.Exec(t, "ALTER TABLE data.bank SET (schema_locked=false)")
 	sqlDB.Exec(t, `TRUNCATE TABLE data.bank`)
+	sqlDB.Exec(t, "ALTER TABLE data.bank SET (schema_locked=true)")
 	sqlDB.Exec(t, `INSERT INTO data.bank VALUES (2, 2), (4, 4)`)
 	sqlDB.Exec(t, `DROP TABLE other.sometable`)
 	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&ts[7])
@@ -5037,7 +5041,9 @@ func TestBackupRestoreIncrementalTruncateTable(t *testing.T) {
 	sqlDB.Exec(t, `INSERT INTO data.t VALUES ('before')`)
 	sqlDB.Exec(t, `BACKUP DATABASE data INTO $1`, full)
 	sqlDB.Exec(t, `UPDATE data.t SET s = 'after'`)
+	sqlDB.Exec(t, "ALTER TABLE data.t SET (schema_locked=false)")
 	sqlDB.Exec(t, `TRUNCATE data.t`)
+	sqlDB.Exec(t, "ALTER TABLE data.t SET (schema_locked=true)")
 
 	sqlDB.Exec(t, "BACKUP DATABASE data INTO $1", full)
 }

pkg/bench/rttanalysis/truncate_bench_test.go

@@ -12,36 +12,36 @@ func init() {
 	reg.Register("Truncate", []RoundTripBenchTestCase{
 		{
 			Name:  "truncate 1 column 0 rows",
-			Setup: "CREATE TABLE t(x INT);",
+			Setup: "CREATE TABLE t(x INT) WITH (schema_locked=false);",
 			Stmt:  "TRUNCATE t",
 		},
 		{
 			Name: "truncate 1 column 1 row",
-			Setup: `CREATE TABLE t(x INT); 
+			Setup: `CREATE TABLE t(x INT) WITH (schema_locked=false); 
 INSERT INTO t (x) VALUES (1);`,
 			Stmt: "TRUNCATE t",
 		},
 		{
 			Name: "truncate 1 column 2 rows",
-			Setup: `CREATE TABLE t(x INT); 
+			Setup: `CREATE TABLE t(x INT) WITH (schema_locked=false);
 INSERT INTO t (x) VALUES (1);
 INSERT INTO t (x) VALUES (2);`,
 			Stmt: "TRUNCATE t",
 		},
 		{
 			Name:  "truncate 2 column 0 rows",
-			Setup: `CREATE TABLE t(x INT, y INT);`,
+			Setup: `CREATE TABLE t(x INT, y INT) WITH (schema_locked=false);`,
 			Stmt:  "TRUNCATE t",
 		},
 		{
 			Name: "truncate 2 column 1 rows",
-			Setup: `CREATE TABLE t(x INT, y INT); 
+			Setup: `CREATE TABLE t(x INT, y INT) WITH (schema_locked=false); 
 INSERT INTO t (x, y) VALUES (1, 1);`,
 			Stmt: "TRUNCATE t",
 		},
 		{
 			Name: "truncate 2 column 2 rows",
-			Setup: `CREATE TABLE t(x INT, y INT); 
+			Setup: `CREATE TABLE t(x INT, y INT) WITH (schema_locked=false);
 INSERT INTO t (x, y) VALUES (1, 1);
 INSERT INTO t (x,y) VALUES (2, 2);`,
 			Stmt: "TRUNCATE t",

pkg/ccl/changefeedccl/sink_test.go

@@ -520,7 +520,9 @@ func TestSQLSink(t *testing.T) {
 	sqlDB.CheckQueryResults(t, `SELECT key, value FROM sink ORDER BY PRIMARY KEY sink`,
 		[][]string{{`k1`, `v0`}},
 	)
+	sqlDB.Exec(t, `ALTER TABLE sink SET (schema_locked=false)`)
 	sqlDB.Exec(t, `TRUNCATE sink`)
+	sqlDB.Exec(t, `ALTER TABLE sink SET (schema_locked=true)`)
 
 	// Verify the implicit flushing
 	sqlDB.CheckQueryResults(t, `SELECT count(*) FROM sink`, [][]string{{`0`}})
@@ -532,7 +534,9 @@ func TestSQLSink(t *testing.T) {
 	sqlDB.CheckQueryResults(t, `SELECT count(*) FROM sink`, [][]string{{`3`}})
 	require.NoError(t, sink.Flush(ctx))
 	sqlDB.CheckQueryResults(t, `SELECT count(*) FROM sink`, [][]string{{`4`}})
+	sqlDB.Exec(t, `ALTER TABLE sink SET (schema_locked=false)`)
 	sqlDB.Exec(t, `TRUNCATE sink`)
+	sqlDB.Exec(t, `ALTER TABLE sink SET (schema_locked=true)`)
 
 	// Two tables interleaved in time
 	var pool testAllocPool
@@ -544,7 +548,9 @@ func TestSQLSink(t *testing.T) {
 	sqlDB.CheckQueryResults(t, `SELECT topic, key, value FROM sink ORDER BY PRIMARY KEY sink`,
 		[][]string{{`bar`, `kbar`, `v0`}, {`foo`, `kfoo`, `v0`}, {`foo`, `kfoo`, `v1`}},
 	)
+	sqlDB.Exec(t, `ALTER TABLE sink SET (schema_locked=false)`)
 	sqlDB.Exec(t, `TRUNCATE sink`)
+	sqlDB.Exec(t, `ALTER TABLE sink SET (schema_locked=true)`)
 
 	// Multiple keys interleaved in time. Use sqlSinkNumPartitions+1 keys to
 	// guarantee that at lease two of them end up in the same partition.
@@ -569,7 +575,9 @@ func TestSQLSink(t *testing.T) {
 			{`2`, `v0`, `v1`},
 		},
 	)
+	sqlDB.Exec(t, `ALTER TABLE sink SET (schema_locked=false)`)
 	sqlDB.Exec(t, `TRUNCATE sink`)
+	sqlDB.Exec(t, `ALTER TABLE sink SET (schema_locked=true)`)
 
 	// Emit resolved
 	var e testEncoder

pkg/ccl/logictestccl/testdata/logic_test/partitioning

@@ -1517,9 +1517,15 @@ bar  CREATE TABLE public.bar (
        gc.ttlseconds = 150;
 
 
+statement ok
+ALTER TABLE bar SET (schema_locked=false)
+
 statement ok
 TRUNCATE bar;
 
+statement ok
+ALTER TABLE bar RESET (schema_locked)
+
 query TT
 SHOW CREATE TABLE bar;
 ----

pkg/ccl/multiregionccl/unique_test.go

@@ -150,8 +150,10 @@ CREATE TABLE u (
 		r.Exec(t, `SELECT crdb_internal.revalidate_unique_constraint('u', 'u_v_key')`)
 
 		// Clean up.
-		_, err = db.Exec(`TRUNCATE test.t`)
-		require.NoError(t, err)
+		r.Exec(t, `ALTER TABLE test.t SET (schema_locked=false)`)
+		r.Exec(t, `TRUNCATE test.t`)
+		r.Exec(t, `ALTER TABLE test.t SET (schema_locked=true)`)
+
 	})
 
 	t.Run("validate_implicitly_partitioned_primary_index", func(t *testing.T) {
@@ -192,8 +194,9 @@ CREATE TABLE u (
 		r.Exec(t, `SELECT crdb_internal.revalidate_unique_constraint('u', 'u_v_key')`)
 
 		// Clean up.
-		_, err = db.Exec(`TRUNCATE test.u`)
-		require.NoError(t, err)
+		r.Exec(t, `ALTER TABLE test.u SET (schema_locked=false)`)
+		r.Exec(t, `TRUNCATE test.u`)
+		r.Exec(t, `ALTER TABLE test.u SET (schema_locked=true)`)
 	})
 
 	t.Run("validate_implicitly_partitioned_secondary_index", func(t *testing.T) {
@@ -234,8 +237,9 @@ CREATE TABLE u (
 		r.Exec(t, `SELECT crdb_internal.revalidate_unique_constraint('u', 'u_pkey')`)
 
 		// Clean up.
-		_, err = db.Exec(`TRUNCATE test.u`)
-		require.NoError(t, err)
+		r.Exec(t, `ALTER TABLE test.u SET (schema_locked=false)`)
+		r.Exec(t, `TRUNCATE test.u`)
+		r.Exec(t, `ALTER TABLE test.u SET (schema_locked=true)`)
 	})
 
 }

pkg/cli/interactive_tests/test_copy.tcl

@@ -10,7 +10,7 @@ eexpect root@
 send "drop table if exists t;\r"
 eexpect "DROP TABLE"
 eexpect root@
-send "create table t (id INT PRIMARY KEY, t TEXT);\r"
+send "create table t (id INT PRIMARY KEY, t TEXT) WITH (schema_locked=false);\r"
 eexpect "CREATE TABLE"
 eexpect root@
 

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -797,15 +797,16 @@ func TestLogicalReplicationWithPhantomDelete(t *testing.T) {
 	skip.UnderDeadlock(t)
 	defer log.Scope(t).Close(t)
 
-	ctx := context.Background()
-
-	tc, s, serverASQL, serverBSQL := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
-	defer tc.Stopper().Stop(ctx)
-
-	serverAURL := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("a"))
-
 	for _, mode := range []string{"validated", "immediate"} {
 		t.Run(mode, func(t *testing.T) {
+			ctx := context.Background()
+			tc, s, serverASQL, serverBSQL := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
+			defer tc.Stopper().Stop(ctx)
+
+			serverAURL := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("a"))
+			serverASQL.Exec(t, "ALTER TABLE tab SET (schema_locked = false)")
+			serverBSQL.Exec(t, "ALTER TABLE tab SET (schema_locked = false)")
+
 			serverASQL.Exec(t, "TRUNCATE tab")
 			serverBSQL.Exec(t, "TRUNCATE tab")
 			var jobBID jobspb.JobID

pkg/server/application_api/metrics_test.go

@@ -338,7 +338,7 @@ func TestStoreProcedureCallStatementMetrics(t *testing.T) {
 	)
 
 	_, err := db.Exec(`		
-		CREATE TABLE tbl (id SERIAL PRIMARY KEY, t text UNIQUE);
+		CREATE TABLE tbl (id SERIAL PRIMARY KEY, t text UNIQUE) WITH (schema_locked=false);
 		INSERT INTO tbl (t) VALUES ('d');`)
 	require.NoError(t, err)
 
@@ -529,7 +529,7 @@ func TestUDFStatementMetrics(t *testing.T) {
 	)
 
 	_, err := db.Exec(`		
-		CREATE TABLE tbl (id SERIAL PRIMARY KEY, t text UNIQUE);
+		CREATE TABLE tbl (id SERIAL PRIMARY KEY, t text UNIQUE) WITH (schema_locked=false);
 		INSERT INTO tbl (t) VALUES ('d');`)
 	require.NoError(t, err)
 

pkg/sql/comment_on_column_test.go

@@ -26,7 +26,7 @@ func TestCommentOnColumn(t *testing.T) {
 		if _, err := db.Exec(`
 		CREATE DATABASE d;
 		SET DATABASE = d;
-		CREATE TABLE t (c1 INT, c2 INT, c3 INT);
+		CREATE TABLE t (c1 INT, c2 INT, c3 INT) WITH (schema_locked=false);
 	`); err != nil {
 			t.Fatal(err)
 		}

pkg/sql/comment_on_index_test.go

@@ -22,7 +22,7 @@ func TestCommentOnIndex(t *testing.T) {
 		if _, err := db.Exec(`
 		CREATE DATABASE d;
 		SET DATABASE = d;
-		CREATE TABLE t (c INT, INDEX t_c_idx (c));
+		CREATE TABLE t (c INT, INDEX t_c_idx (c)) WITH (schema_locked=false);
 	`); err != nil {
 			t.Fatal(err)
 		}