changefeedccl: support EXCLUDE TABLES in DB-level feeds

KeithCh · KeithCh · commit 35d0006bc821 · 2025-09-24T11:48:55.000-04:00
Support the ability to exclude tables from being watched in a database-level feed. An example usage is: CREATE DATABASE FOR CHANGEFEED defaultdb EXCLUDE TABLES foo,fizz.buzz,defaultdb.bar.tab,notdefaultdb.bar.tab; The table names above in the EXCLUDE TABLES list will be resolved as such: Non-qualified table name foo: - defaultdb.public.foo Partially-qualified table name fizz.buzz: - defaultdb.fizz.buzz Fully-qualified table name defaultdb.bar.tab: - defaultdb.bar.tab Fully-qualified table name notdefaultdb.bar.tab: - Statement error; changefeed job will not be created Note that partially qualified table names in the form <database>.<table> are not supported. Partially qualfied table names are always assumed to be <schema>.<table>. Resolves: #147424 Release note: none
diff --git a/pkg/ccl/changefeedccl/BUILD.bazel b/pkg/ccl/changefeedccl/BUILD.bazel
@@ -120,6 +120,7 @@ go_library(
         "//pkg/sql/rowenc",
         "//pkg/sql/rowexec",
         "//pkg/sql/sem/asof",
+        "//pkg/sql/sem/catconstants",
         "//pkg/sql/sem/catid",
         "//pkg/sql/sem/eval",
         "//pkg/sql/sem/tree",
diff --git a/pkg/ccl/changefeedccl/changefeed.go b/pkg/ccl/changefeedccl/changefeed.go
@@ -8,6 +8,7 @@ package changefeedccl
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
@@ -113,10 +114,12 @@ func getTargetsFromDatabaseSpec(
 		if err != nil {
 			return err
 		}
+		tableDescToSchemaName := make(map[catalog.TableDescriptor]string)
 		tables, err := descs.GetAllTablesInDatabase(ctx, txn.KV(), databaseDescriptor)
 		if err != nil {
 			return err
 		}
+		dbName := databaseDescriptor.GetName()
 		for _, desc := range tables.OrderedDescriptors() {
 			tableDesc, ok := desc.(catalog.TableDescriptor)
 			if !ok {
@@ -126,6 +129,20 @@ func getTargetsFromDatabaseSpec(
 			if !tableDesc.IsPhysicalTable() {
 				continue
 			}
+			if ts.FilterList != nil && ts.FilterList.FilterType == jobspb.FilterList_EXCLUDE_TABLES {
+				if _, ok := tableDescToSchemaName[tableDesc]; !ok {
+					schemaID := tableDesc.GetParentSchemaID()
+					schema, err := descs.ByIDWithLeased(txn.KV()).Get().Schema(ctx, schemaID)
+					if err != nil {
+						return err
+					}
+					tableDescToSchemaName[tableDesc] = schema.GetName()
+				}
+				fullyQualifiedTableName := fmt.Sprintf("%s.%s.%s", dbName, tableDescToSchemaName[tableDesc], tableDesc.GetName())
+				if _, ok := ts.FilterList.Tables[fullyQualifiedTableName]; ok {
+					continue
+				}
+			}
 			var tableType jobspb.ChangefeedTargetSpecification_TargetType
 			if len(tableDesc.GetFamilies()) == 1 {
 				tableType = jobspb.ChangefeedTargetSpecification_PRIMARY_FAMILY_ONLY
diff --git a/pkg/ccl/changefeedccl/changefeed_stmt.go b/pkg/ccl/changefeedccl/changefeed_stmt.go
@@ -46,6 +46,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgnotice"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/asof"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/catconstants"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
@@ -65,6 +66,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
+	pbtypes "github.com/gogo/protobuf/types"
 )
 
 // featureChangefeedEnabled is used to enable and disable the CHANGEFEED feature.
@@ -631,11 +633,22 @@ func createChangefeedJobRecord(
 		if len(targetDatabaseDescs) == 0 || len(targetDatabaseDescs) > 1 {
 			return nil, changefeedbase.Targets{}, errors.Errorf("changefeed only supports one database target")
 		}
-		if targetDatabaseDescs[0].GetID() == keys.SystemDatabaseID {
+		targetDatabaseDesc := targetDatabaseDescs[0]
+		if targetDatabaseDesc.GetID() == keys.SystemDatabaseID {
 			return nil, changefeedbase.Targets{}, errors.Errorf("changefeed cannot target the system database")
 		}
+		if changefeedStmt.FilterOption != nil {
+			fqTableNames, err := getFullyQualifiedTableNames(
+				targetDatabaseDesc.GetName(), changefeedStmt.FilterOption.Tables,
+			)
+			if err != nil {
+				return nil, changefeedbase.Targets{}, err
+			}
+			changefeedStmt.FilterOption.Tables = fqTableNames
+		}
+		targetSpec := getDatabaseTargetSpec(targetDatabaseDesc, changefeedStmt.FilterOption)
 		details = jobspb.ChangefeedDetails{
-			TargetSpecifications: getDatabaseTargets(targetDatabaseDescs),
+			TargetSpecifications: []jobspb.ChangefeedTargetSpecification{targetSpec},
 			SinkURI:              sinkURI,
 			StatementTime:        statementTime,
 			EndTime:              endTime,
@@ -1144,19 +1157,53 @@ func getTargetsAndTables(
 	return targets, tables, nil
 }
 
-func getDatabaseTargets(
-	targetDatabaseDescs []catalog.DatabaseDescriptor,
-) []jobspb.ChangefeedTargetSpecification {
-	targets := make([]jobspb.ChangefeedTargetSpecification, len(targetDatabaseDescs))
+func getDatabaseTargetSpec(
+	targetDatabaseDesc catalog.DatabaseDescriptor, filterOpt *tree.ChangefeedFilterOption,
+) jobspb.ChangefeedTargetSpecification {
+	target := jobspb.ChangefeedTargetSpecification{
+		DescID:            targetDatabaseDesc.GetID(),
+		Type:              jobspb.ChangefeedTargetSpecification_DATABASE,
+		StatementTimeName: targetDatabaseDesc.GetName(),
+	}
+	if filterOpt != nil {
+		filterTables := make(map[string]pbtypes.Empty)
+		for _, table := range filterOpt.Tables {
+			filterTables[table.FQString()] = pbtypes.Empty{}
+		}
+		target.FilterList = &jobspb.FilterList{
+			FilterType: jobspb.FilterList_FilterType(filterOpt.FilterType),
+			Tables:     filterTables,
+		}
+	}
+	return target
+}
 
-	for i, desc := range targetDatabaseDescs {
-		targets[i] = jobspb.ChangefeedTargetSpecification{
-			DescID:            desc.GetID(),
-			Type:              jobspb.ChangefeedTargetSpecification_DATABASE,
-			StatementTimeName: desc.GetName(),
+func getFullyQualifiedTableNames(
+	targetDatabase string, tableNames tree.TableNames,
+) (tree.TableNames, error) {
+	var fqTableNames tree.TableNames
+
+	for _, tableName := range tableNames {
+		if tableName.SchemaName == "" {
+			// The table name is non-qualified e.g. foo. This will resolve to <targetDatabase>.public.foo.
+			tableName.SchemaName = catconstants.PublicSchemaName
+			tableName.CatalogName = tree.Name(targetDatabase)
+		} else if tableName.CatalogName == "" {
+			// The table name is partially qualified e.g. foo.bar. This will resolve to
+			// <targetDatabase>.foo.bar.
+			tableName.CatalogName = tree.Name(targetDatabase)
+		} else {
+			// Table name is fully qualfied e.g. foo.bar.fizz. This will resolve to
+			// foo.bar.fizz unless foo != <targetDatabase>, in which case it would fail.
+			if tableName.CatalogName != tree.Name(targetDatabase) {
+				return nil, errors.AssertionFailedf(
+					"table %q must be in target database %q", tableName.FQString(), targetDatabase,
+				)
+			}
 		}
+		fqTableNames = append(fqTableNames, tableName)
 	}
-	return targets
+	return fqTableNames, nil
 }
 
 func validateSink(
diff --git a/pkg/ccl/changefeedccl/changefeed_test.go b/pkg/ccl/changefeedccl/changefeed_test.go
@@ -245,7 +245,7 @@ func TestDatabaseLevelChangefeedWithIncludeFilter(t *testing.T) {
 
 		expectSuccess(`CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo`)
 		expectSuccess(`CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo,foo2`)
-		expectSuccess(`CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo.bar.fizz, foo.foo2, foo`)
+		expectSuccess(`CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES d.bar.fizz, foo.foo2, foo`)
 		expectErrCreatingFeed(t, f, `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo.*`,
 			`at or near "*": syntax error`)
 		// TODO(#147421): Assert payload once the filter works
@@ -258,26 +258,85 @@ func TestDatabaseLevelChangefeedWithExcludeFilter(t *testing.T) {
 	defer log.Scope(t).Close(t)
 
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
-		expectSuccess := func(stmt string) {
-			successfulFeed := feed(t, f, stmt)
-			defer closeFeed(t, successfulFeed)
-			_, err := successfulFeed.Next()
-			require.NoError(t, err)
-		}
 		sqlDB := sqlutils.MakeSQLRunner(s.DB)
-		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
-		sqlDB.Exec(t, `INSERT INTO foo VALUES (0, 'initial')`)
-		sqlDB.Exec(t, `UPSERT INTO foo VALUES (0, 'updated')`)
-		sqlDB.Exec(t, `CREATE TABLE foo2 (a INT PRIMARY KEY, b STRING)`)
-		sqlDB.Exec(t, `INSERT INTO foo2 VALUES (0, 'initial')`)
-		sqlDB.Exec(t, `UPSERT INTO foo2 VALUES (0, 'updated')`)
+		for i := range 4 {
+			name := fmt.Sprintf("foo%d", i+1)
+			sqlDB.Exec(t, fmt.Sprintf(`CREATE TABLE %s (a INT PRIMARY KEY, b STRING)`, name))
+			sqlDB.Exec(t, fmt.Sprintf(`INSERT INTO %s VALUES (0, 'initial')`, name))
+			sqlDB.Exec(t, fmt.Sprintf(`UPSERT INTO %s VALUES (0, 'updated')`, name))
+		}
+
+		sqlDB.Exec(t, `CREATE SCHEMA private`)
+		sqlDB.Exec(t, `CREATE TABLE private.foo1 (a INT PRIMARY KEY, b STRING)`)
+		sqlDB.Exec(t, `INSERT INTO private.foo1 VALUES (0, 'initial')`)
+		// Test that if there are multiple tables with the same name the correct
+		// one will still be found using the default schema of public.
+		feed1 := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES foo1`)
+		defer closeFeed(t, feed1)
+		assertPayloads(t, feed1, []string{
+			`foo1: [0]->{"after": {"a": 0, "b": "initial"}}`,
+			`foo2: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo3: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo4: [0]->{"after": {"a": 0, "b": "updated"}}`,
+		})
+
+		feed2 := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES foo2`)
+		defer closeFeed(t, feed2)
+		assertPayloads(t, feed2, []string{
+			`foo1: [0]->{"after": {"a": 0, "b": "initial"}}`,
+			`foo1: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo3: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo4: [0]->{"after": {"a": 0, "b": "updated"}}`,
+		})
+
+		// Test that we can exclude multiple tables.
+		feed3 := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES foo2,foo3`)
+		defer closeFeed(t, feed3)
+		assertPayloads(t, feed3, []string{
+			`foo1: [0]->{"after": {"a": 0, "b": "initial"}}`,
+			`foo1: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo4: [0]->{"after": {"a": 0, "b": "updated"}}`,
+		})
+
+		// Test that we can handle fully qualfied, partially qualified, and unqualified table names.
+		feed4 := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES d.public.foo2, public.foo3, foo4`)
+		defer closeFeed(t, feed4)
+		assertPayloads(t, feed4, []string{
+			`foo1: [0]->{"after": {"a": 0, "b": "initial"}}`,
+			`foo1: [0]->{"after": {"a": 0, "b": "updated"}}`,
+		})
+
+		// Test that we can handle tables that don't exist.
+		feed5 := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES bob`)
+		defer closeFeed(t, feed5)
+		assertPayloads(t, feed5, []string{
+			`foo1: [0]->{"after": {"a": 0, "b": "initial"}}`,
+			`foo1: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo2: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo3: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo4: [0]->{"after": {"a": 0, "b": "updated"}}`,
+		})
 
-		expectSuccess(`CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES foo`)
-		expectSuccess(`CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES foo,foo2`)
-		expectSuccess(`CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES foo.bar.fizz, foo.foo2, foo`)
+		// Test that fully qualified table names outside of the target database will
+		// cause an error.
+		expectErrCreatingFeed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES fizz.buzz.foo`,
+			`table "fizz.buzz.foo" must be in target database "d"`)
+
+		// Table patterns are not supported.
 		expectErrCreatingFeed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES foo.*`,
 			`at or near "*": syntax error`)
-		// TODO(#147421): Assert payload once the filter works
+
+		// Test that name resolution is not dependent on search_path() or current DB
+		sqlDB.Exec(t, `CREATE DATABASE notd`)
+		sqlDB.Exec(t, `USE notd`)
+		sqlDB.Exec(t, `SET search_path TO notpublic`)
+		feed6 := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES foo1, private.foo1`)
+		defer closeFeed(t, feed6)
+		assertPayloads(t, feed6, []string{
+			`foo2: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo3: [0]->{"after": {"a": 0, "b": "updated"}}`,
+			`foo4: [0]->{"after": {"a": 0, "b": "updated"}}`,
+		})
 	}
 	cdcTest(t, testFn, feedTestEnterpriseSinks)
 }
diff --git a/pkg/cmd/protoc-gen-gogoroach/BUILD.bazel b/pkg/cmd/protoc-gen-gogoroach/BUILD.bazel
@@ -36,6 +36,7 @@ go_proto_compiler(
         "Mgoogle/protobuf/any.proto=github.com/gogo/protobuf/types",
         "Mgoogle/protobuf/descriptor.proto=github.com/gogo/protobuf/protoc-gen-gogo/descriptor",
         "Mgoogle/protobuf/duration.proto=github.com/gogo/protobuf/types",
+        "Mgoogle/protobuf/empty.proto=github.com/gogo/protobuf/types",
     ],
     plugin = ":protoc-gen-gogoroach",
     valid_archive = False,
@@ -58,6 +59,7 @@ go_proto_compiler(
         "Mgoogle/protobuf/any.proto=github.com/gogo/protobuf/types",
         "Mgoogle/protobuf/descriptor.proto=github.com/gogo/protobuf/protoc-gen-gogo/descriptor",
         "Mgoogle/protobuf/duration.proto=github.com/gogo/protobuf/types",
+        "Mgoogle/protobuf/empty.proto=github.com/gogo/protobuf/types",
     ],
     plugin = ":protoc-gen-gogoroach",
     valid_archive = False,
diff --git a/pkg/jobs/jobspb/BUILD.bazel b/pkg/jobs/jobspb/BUILD.bazel
@@ -50,6 +50,7 @@ proto_library(
         "@com_github_cockroachdb_errors//errorspb:errorspb_proto",
         "@com_github_gogo_protobuf//gogoproto:gogo_proto",
         "@com_google_protobuf//:any_proto",
+        "@com_google_protobuf//:empty_proto",
         "@com_google_protobuf//:timestamp_proto",
     ],
 )
diff --git a/pkg/jobs/jobspb/jobs.proto b/pkg/jobs/jobspb/jobs.proto
@@ -22,6 +22,7 @@ import "clusterversion/cluster_version.proto";
 import "google/protobuf/timestamp.proto";
 import "util/tracing/tracingpb/recorded_span.proto";
 import "sql/catalog/externalcatalog/externalpb/external.proto";
+import "google/protobuf/empty.proto";
 
 enum EncryptionMode {
   Passphrase = 0;
@@ -1107,7 +1108,15 @@ message ChangefeedTargetSpecification {
     (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.ID"];
   string family_name = 3;
   string statement_time_name = 4;
+  FilterList filter_list = 5;
+}
 
+message FilterList {
+  enum FilterType {
+    EXCLUDE_TABLES = 0;
+  }
+  FilterType filter_type = 1;
+  map<string, google.protobuf.Empty> tables = 2 [(gogoproto.nullable) = false];
 }
 
 message ChangefeedDetails {
diff --git a/pkg/protos.bzl b/pkg/protos.bzl
@@ -58,6 +58,7 @@ SERVER_PROTOS = [
     "@com_google_protobuf//:any_proto",
     "@com_google_protobuf//:descriptor_proto",
     "@com_google_protobuf//:duration_proto",
+    "@com_google_protobuf//:empty_proto",
     "@com_google_protobuf//:timestamp_proto",
     "@go_googleapis//google/api:annotations_proto",
     "@go_googleapis//google/api:http_proto",

Original file line number	Diff line number	Diff line change
`@@ -50,6 +50,7 @@ proto_library(`
`50`	`50`	`"@com_github_cockroachdb_errors//errorspb:errorspb_proto",`
`51`	`51`	`"@com_github_gogo_protobuf//gogoproto:gogo_proto",`
`52`	`52`	`"@com_google_protobuf//:any_proto",`
	`53`	`+ "@com_google_protobuf//:empty_proto",`
`53`	`54`	`"@com_google_protobuf//:timestamp_proto",`
`54`	`55`	`],`
`55`	`56`	`)`