changefeedccl: implement INCLUDE TABLES syntax in DB-level feeds

The syntax for INCLUDE TABLES will be
CREATE CHANGEFEED FOR DATABASE FOO EXCLUDE TABLES fizz,buzz;

Resolves #147420
Release note: None

Author: KeithCh

docs/generated/sql/bnf/stmt_block.bnf

@@ -1953,6 +1953,7 @@ opt_changefeed_sink ::=
 
 db_level_changefeed_filter_option ::=
 	'EXCLUDE' 'TABLES' table_name_list
+	| 'INCLUDE' 'TABLES' table_name_list
 	| 
 
 target_list ::=

pkg/ccl/changefeedccl/changefeed_test.go

@@ -224,7 +224,36 @@ func TestDatabaseLevelChangefeedBasics(t *testing.T) {
 	cdcTest(t, testFn)
 }
 
-func TestDatabaseLevelChangefeedWithFilter(t *testing.T) {
+func TestDatabaseLevelChangefeedWithIncludeFilter(t *testing.T) {
+	defer leaktest.AfterTest(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')`)
+
+		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`)
+		expectErrCreatingFeed(t, f, `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo.*`,
+			`at or near "*": syntax error`)
+		// TODO(#147421): Assert payload once the filter works
+	}
+	cdcTest(t, testFn, feedTestEnterpriseSinks)
+}
+
+func TestDatabaseLevelChangefeedWithExcludeFilter(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 

pkg/sql/parser/sql.y

@@ -6501,6 +6501,10 @@ db_level_changefeed_filter_option:
   {
     $$.val = &tree.ChangefeedFilterOption{Tables: $3.tableNames(), FilterType: tree.ExcludeFilter}
   }
+| INCLUDE TABLES table_name_list
+  {
+    $$.val = &tree.ChangefeedFilterOption{Tables: $3.tableNames(), FilterType: tree.IncludeFilter}
+  }
 | /* EMPTY */ 
   {
     $$.val = nil

pkg/sql/parser/testdata/changefeed

@@ -196,3 +196,21 @@ CREATE CHANGEFEED FOR DATABASE d1 EXCLUDE TABLES foo, bar INTO ('*****') -- full
 CREATE CHANGEFEED FOR DATABASE d1 EXCLUDE TABLES foo, bar INTO '_' -- literals removed
 CREATE CHANGEFEED FOR DATABASE _ EXCLUDE TABLES _, _ INTO '*****' -- identifiers removed
 CREATE CHANGEFEED FOR DATABASE d1 EXCLUDE TABLES foo, bar INTO 'foo' -- passwords exposed
+
+parse
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES bar INTO 'foo'
+----
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES bar INTO '*****' -- normalized!
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES bar INTO ('*****') -- fully parenthesized
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES bar INTO '_' -- literals removed
+CREATE CHANGEFEED FOR DATABASE _ INCLUDE TABLES _ INTO '*****' -- identifiers removed
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES bar INTO 'foo' -- passwords exposed
+
+parse
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES foo,bar INTO 'foo'
+----
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES foo, bar INTO '*****' -- normalized!
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES foo, bar INTO ('*****') -- fully parenthesized
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES foo, bar INTO '_' -- literals removed
+CREATE CHANGEFEED FOR DATABASE _ INCLUDE TABLES _, _ INTO '*****' -- identifiers removed
+CREATE CHANGEFEED FOR DATABASE d1 INCLUDE TABLES foo, bar INTO 'foo' -- passwords exposed

pkg/sql/sem/tree/changefeed.go

@@ -23,9 +23,9 @@ const (
 
 type FilterType int
 
-// Needs to be the same as
 const (
 	ExcludeFilter FilterType = iota
+	IncludeFilter
 )
 
 type ChangefeedFilterOption struct {
@@ -38,7 +38,7 @@ func (l ChangefeedLevel) String() string {
 }
 
 func (l FilterType) String() string {
-	return []string{"EXCLUDE"}[l]
+	return []string{"EXCLUDE", "INCLUDE"}[l]
 }
 
 // CreateChangefeed represents a CREATE CHANGEFEED statement.