changefeedccl: respect full_table_name for db-level feeds

Previously, DB-level feeds would emit messages with plain
table neames, even when full_table_name was specified. Now,
we respect this option.

Epic: none
Fixes: #158145

Release note: None

Author: aerfrei

pkg/ccl/changefeedccl/changefeed.go

@@ -79,7 +79,10 @@ func AllTargets(
 					if len(cd.TargetSpecifications) > 1 {
 						return changefeedbase.Targets{}, errors.AssertionFailedf("database-level changefeed is not supported with multiple targets")
 					}
-					targets, err = getTargetsFromDatabaseSpec(ctx, ts, execCfg, timestamp)
+					_, useFullTableName := cd.Opts[changefeedbase.OptFullTableName]
+					targets, err = getTargetsFromDatabaseSpec(
+						ctx, ts, execCfg, timestamp, useFullTableName,
+					)
 					if err != nil {
 						return changefeedbase.Targets{}, err
 					}
@@ -117,6 +120,7 @@ func getTargetsFromDatabaseSpec(
 	ts jobspb.ChangefeedTargetSpecification,
 	execCfg *sql.ExecutorConfig,
 	timestamp hlc.Timestamp,
+	useFullTableName bool,
 ) (targets changefeedbase.Targets, err error) {
 	err = sql.DescsTxn(ctx, execCfg, func(
 		ctx context.Context, txn isql.Txn, descs *descs.Collection,
@@ -172,15 +176,21 @@ func getTargetsFromDatabaseSpec(
 					tableType = jobspb.ChangefeedTargetSpecification_EACH_FAMILY
 				}
 
+				tableName := func() string {
+					if useFullTableName {
+						return fullyQualifiedTableName
+					}
+					return desc.GetName()
+				}()
 				targets.Add(changefeedbase.Target{
 					Type:              tableType,
 					DescID:            desc.GetID(),
-					StatementTimeName: changefeedbase.StatementTimeName(desc.GetName()),
+					StatementTimeName: changefeedbase.StatementTimeName(tableName),
 				})
 			}
 		case tree.IncludeFilter:
-			for name := range ts.FilterList.Tables {
-				tn, err := parser.ParseTableName(name)
+			for fullyQualifiedTableName := range ts.FilterList.Tables {
+				tn, err := parser.ParseTableName(fullyQualifiedTableName)
 				if err != nil {
 					return err
 				}
@@ -215,10 +225,16 @@ func getTargetsFromDatabaseSpec(
 					tableType = jobspb.ChangefeedTargetSpecification_EACH_FAMILY
 				}
 
+				tableName := func() string {
+					if useFullTableName {
+						return fullyQualifiedTableName
+					}
+					return desc.GetName()
+				}()
 				targets.Add(changefeedbase.Target{
 					Type:              tableType,
 					DescID:            tableID,
-					StatementTimeName: changefeedbase.StatementTimeName(desc.GetName()),
+					StatementTimeName: changefeedbase.StatementTimeName(tableName),
 				})
 			}
 		default:

pkg/ccl/changefeedccl/changefeed_test.go

@@ -1374,6 +1374,30 @@ func TestChangefeedFullTableName(t *testing.T) {
 	})
 }
 
+func TestDatabaseLevelChangefeedWithFullTableName(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
+
+		normal := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d WITH full_table_name`)
+		defer closeFeed(t, normal)
+		include := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo WITH full_table_name`)
+		defer closeFeed(t, include)
+		exclude := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES bar WITH full_table_name`)
+		defer closeFeed(t, exclude)
+
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1, 'a')`)
+		assertPayloads(t, normal, []string{`d.public.foo: [1]->{"after": {"a": 1, "b": "a"}}`})
+		assertPayloads(t, include, []string{`d.public.foo: [1]->{"after": {"a": 1, "b": "a"}}`})
+		assertPayloads(t, exclude, []string{`d.public.foo: [1]->{"after": {"a": 1, "b": "a"}}`})
+	}
+
+	cdcTest(t, testFn, feedTestEnterpriseSinks)
+}
+
 func TestChangefeedMultiTable(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)