ttl: allow outbound foreign keys on TTL tables

Fixes #101225

This change allows TTL tables to have outbound foreign keys. Outbound foreign
keys are unaffected by rows being deleted from the TTL table itself.

Release note (sql change): Allow outbound foreign keys on TTL tables.

pkg/sql/catalog/tabledesc/validate.go

@@ -198,23 +198,14 @@ func (desc *wrapper) ValidateForwardReferences(
 		}
 	}
 
-	// Row-level TTL is not compatible with foreign keys.
+	// Row-level TTL is not compatible with inbound foreign keys.
 	// This check should be in ValidateSelf but interferes with AllocateIDs.
-	if desc.HasRowLevelTTL() {
-		if len(desc.OutboundForeignKeys()) > 0 {
-			vea.Report(unimplemented.NewWithIssuef(
-				76407,
-				`foreign keys from table with TTL %q are not permitted`,
-				desc.GetName(),
-			))
-		}
-		if len(desc.InboundForeignKeys()) > 0 {
-			vea.Report(unimplemented.NewWithIssuef(
-				76407,
-				`foreign keys to table with TTL %q are not permitted`,
-				desc.GetName(),
-			))
-		}
+	if desc.HasRowLevelTTL() && len(desc.InboundForeignKeys()) > 0 {
+		vea.Report(unimplemented.NewWithIssuef(
+			76407,
+			`foreign keys to table with TTL %q are not permitted`,
+			desc.GetName(),
+		))
 	}
 
 	// Check enforced outbound foreign keys.

pkg/sql/logictest/testdata/logic_test/row_level_ttl

@@ -962,28 +962,25 @@ subtest end
 subtest foreign_key_constraint
 
 statement ok
-CREATE TABLE ref_table (id INT PRIMARY KEY, ref INT)
+CREATE TABLE ttl_table (id INT PRIMARY KEY) WITH (ttl_expire_after = '10 mins')
 
-statement error foreign keys from table with TTL "ttl_table" are not permitted
-CREATE TABLE ttl_table (id INT PRIMARY KEY, ref INT REFERENCES ref_table(id)) WITH (ttl_expire_after = '10 mins')
+statement error foreign keys to table with TTL "ttl_table" are not permitted
+CREATE TABLE ref_table (id INT PRIMARY KEY, ref INT REFERENCES ttl_table(id))
 
 statement ok
-CREATE TABLE ttl_table (id INT PRIMARY KEY, ref INT) WITH (ttl_expire_after = '10 mins')
-
-statement error foreign keys to table with TTL "ttl_table" are not permitted
-CREATE TABLE new_ref_table (id INT PRIMARY KEY, ref INT REFERENCES ttl_table(id))
+CREATE TABLE ref_table (id INT PRIMARY KEY, ref INT)
 
 statement error foreign keys to table with TTL "ttl_table" are not permitted
 ALTER TABLE ref_table ADD CONSTRAINT fk FOREIGN KEY (ref) REFERENCES ttl_table (id)
 
-statement error foreign keys from table with TTL "ttl_table" are not permitted
-ALTER TABLE ttl_table ADD CONSTRAINT fk FOREIGN KEY (ref) REFERENCES ttl_table (id)
+statement ok
+CREATE TABLE become_ttl_table (id INT PRIMARY KEY)
 
 statement ok
-CREATE TABLE ttl_become_table (id INT PRIMARY KEY, ref INT REFERENCES ref_table (id))
+CREATE TABLE become_ref_table (id INT PRIMARY KEY, ref INT REFERENCES become_ttl_table(id))
 
-statement error foreign keys from table with TTL "ttl_become_table" are not permitted
-ALTER TABLE ttl_become_table SET (ttl_expire_after = '10 minutes')
+statement error foreign keys to table with TTL "become_ttl_table" are not permitted
+ALTER TABLE become_ttl_table SET (ttl_expire_after = '10 mins')
 
 subtest end
 

pkg/sql/schemachanger/scbuild/internal/scbuildstmt/alter_table_alter_primary_key.go

@@ -425,15 +425,11 @@ func fallBackIfDescColInRowLevelTTLTables(b BuildCtx, tableID catid.DescID, t al
 	}
 
 	_, _, ns := scpb.FindNamespace(b.QueryByID(tableID))
-	// Panic if there is any inbound/outbound FK constraints.
+	// Panic if there are any inbound FK constraints.
 	if _, _, inboundFKElem := scpb.FindForeignKeyConstraint(b.BackReferences(tableID)); inboundFKElem != nil {
 		panic(scerrors.NotImplementedErrorf(t.n,
 			`foreign keys to table with TTL %q are not permitted`, ns.Name))
 	}
-	if _, _, outboundFKElem := scpb.FindForeignKeyConstraint(b.QueryByID(tableID)); outboundFKElem != nil {
-		panic(scerrors.NotImplementedErrorf(t.n,
-			`foreign keys from table with TTL %q are not permitted`, ns.Name))
-	}
 }
 
 func mustRetrievePrimaryIndexElement(b BuildCtx, tableID catid.DescID) (res *scpb.PrimaryIndex) {

pkg/sql/ttl/ttljob/ttljob_test.go

@@ -211,7 +211,7 @@ func (h *rowLevelTTLTestJobTestHelper) verifyExpiredRowsJobOnly(
 		var progressBytes []byte
 		require.NoError(t, rows.Scan(&status, &progressBytes))
 
-		require.Equal(t, "succeeded", status)
+		require.Equal(t, string(jobs.StatusSucceeded), status)
 
 		var progress jobspb.Progress
 		require.NoError(t, protoutil.Unmarshal(progressBytes, &progress))
@@ -242,7 +242,7 @@ func (h *rowLevelTTLTestJobTestHelper) verifyExpiredRows(
 		var progressBytes []byte
 		require.NoError(t, rows.Scan(&status, &progressBytes))
 
-		require.Equal(t, "succeeded", status)
+		require.Equal(t, string(jobs.StatusSucceeded), status)
 
 		var progress jobspb.Progress
 		require.NoError(t, protoutil.Unmarshal(progressBytes, &progress))
@@ -880,3 +880,32 @@ func TestRowLevelTTLJobRandomEntries(t *testing.T) {
 		})
 	}
 }
+
+func TestOutboundForeignKey(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	th, cleanupFunc := newRowLevelTTLTestJobTestHelper(
+		t,
+		&sql.TTLTestingKnobs{
+			AOSTDuration:     &zeroDuration,
+			ReturnStatsError: true,
+		},
+		false, /* testMultiTenant */
+		1,     /* numNodes */
+	)
+	defer cleanupFunc()
+
+	sqlDB := th.sqlDB
+	sqlDB.Exec(t, "CREATE TABLE parent (id INT PRIMARY KEY)")
+	sqlDB.Exec(t, "CREATE TABLE tbl (id INT PRIMARY KEY, expire_at TIMESTAMPTZ, parent_id INT REFERENCES parent (id)) WITH (ttl_expiration_expression = 'expire_at')")
+
+	sqlDB.Exec(t, "INSERT INTO parent VALUES (1)")
+	sqlDB.Exec(t, "INSERT INTO tbl VALUES (1, '2020-01-01', 1)")
+
+	// Force the schedule to execute.
+	th.waitForScheduledJob(t, jobs.StatusSucceeded, "")
+
+	results := sqlDB.QueryStr(t, "SELECT * FROM tbl")
+	require.Empty(t, results)
+}