@@ -229,26 +229,78 @@ func TestDatabaseLevelChangefeedWithIncludeFilter(t *testing.T) {
229229 defer log .Scope (t ).Close (t )
230230
231231 testFn := func (t * testing.T , s TestServer , f cdctest.TestFeedFactory ) {
232- expectSuccess := func (stmt string ) {
233- successfulFeed := feed (t , f , stmt )
234- defer closeFeed (t , successfulFeed )
235- _ , err := successfulFeed .Next ()
236- require .NoError (t , err )
237- }
238232 sqlDB := sqlutils .MakeSQLRunner (s .DB )
239- sqlDB .Exec (t , `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)` )
240- sqlDB .Exec (t , `INSERT INTO foo VALUES (0, 'initial')` )
241- sqlDB .Exec (t , `UPSERT INTO foo VALUES (0, 'updated')` )
242- sqlDB .Exec (t , `CREATE TABLE foo2 (a INT PRIMARY KEY, b STRING)` )
243- sqlDB .Exec (t , `INSERT INTO foo2 VALUES (0, 'initial')` )
244- sqlDB .Exec (t , `UPSERT INTO foo2 VALUES (0, 'updated')` )
233+ for i := range 4 {
234+ name := fmt .Sprintf ("foo%d" , i + 1 )
235+ sqlDB .Exec (t , fmt .Sprintf (`CREATE TABLE %s (a INT PRIMARY KEY, b STRING)` , name ))
236+ sqlDB .Exec (t , fmt .Sprintf (`INSERT INTO %s VALUES (0, 'initial')` , name ))
237+ sqlDB .Exec (t , fmt .Sprintf (`UPSERT INTO %s VALUES (0, 'updated')` , name ))
238+ }
239+
240+ sqlDB .Exec (t , `CREATE SCHEMA private` )
241+ sqlDB .Exec (t , `CREATE TABLE private.foo1 (a INT PRIMARY KEY, b STRING)` )
242+ sqlDB .Exec (t , `INSERT INTO private.foo1 VALUES (0, 'initial')` )
243+ // Test that if there are multiple tables with the same name the correct
244+ // one will still be found using the default schema of public.
245+ feed1 := feed (t , f , `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo1` )
246+ defer closeFeed (t , feed1 )
247+ assertPayloads (t , feed1 , []string {
248+ `foo1: [0]->{"after": {"a": 0, "b": "updated"}}` ,
249+ })
250+
251+ // Test that we can include multiple tables.
252+ feed2 := feed (t , f , `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo1,foo2` )
253+ defer closeFeed (t , feed2 )
254+ assertPayloads (t , feed2 , []string {
255+ `foo1: [0]->{"after": {"a": 0, "b": "updated"}}` ,
256+ `foo2: [0]->{"after": {"a": 0, "b": "updated"}}` ,
257+ })
245258
246- expectSuccess (`CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo` )
247- expectSuccess (`CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo,foo2` )
248- expectSuccess (`CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES d.bar.fizz, foo.foo2, foo` )
259+ // Test that we can handle fully qualified, partially qualified, and unqualified
260+ // table names.
261+ feed3 := feed (t , f ,
262+ `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES d.public.foo2, public.foo3, foo4` )
263+ defer closeFeed (t , feed3 )
264+ assertPayloads (t , feed3 , []string {
265+ `foo2: [0]->{"after": {"a": 0, "b": "updated"}}` ,
266+ `foo3: [0]->{"after": {"a": 0, "b": "updated"}}` ,
267+ `foo4: [0]->{"after": {"a": 0, "b": "updated"}}` ,
268+ })
269+
270+ // Test that we can handle tables that don't exist.
271+ feed4 := feed (t , f , `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo1,bob` )
272+ defer closeFeed (t , feed4 )
273+ assertPayloads (t , feed4 , []string {
274+ `foo1: [0]->{"after": {"a": 0, "b": "updated"}}` ,
275+ })
276+
277+ // Test that fully qualified table names outside of the target database will
278+ // cause an error.
279+ expectErrCreatingFeed (t , f , `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES fizz.buzz.foo` ,
280+ `table "fizz.buzz.foo" must be in target database "d"` )
281+
282+ // Table patterns are not supported.
249283 expectErrCreatingFeed (t , f , `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo.*` ,
250284 `at or near "*": syntax error` )
251- // TODO(#147421): Assert payload once the filter works
285+
286+ // Test that name resolution is not dependent on search_path() or current DB.
287+ sqlDB .Exec (t , `CREATE DATABASE notd` )
288+ sqlDB .Exec (t , `USE notd` )
289+ sqlDB .Exec (t , `SET search_path TO notpublic` )
290+ feed5 := feed (t , f , `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo1, private.foo1` )
291+ defer closeFeed (t , feed5 )
292+ assertPayloads (t , feed5 , []string {
293+ `foo1: [0]->{"after": {"a": 0, "b": "updated"}}` ,
294+ `foo1: [0]->{"after": {"a": 0, "b": "initial"}}` ,
295+ })
296+
297+ // Test that partially qualified table names are always assumed to be
298+ // <schema>.<table>.
299+ feed6 := feed (t , f , `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES d.foo1, public.foo2` )
300+ defer closeFeed (t , feed6 )
301+ assertPayloads (t , feed6 , []string {
302+ `foo2: [0]->{"after": {"a": 0, "b": "updated"}}` ,
303+ })
252304 }
253305 cdcTest (t , testFn , feedTestEnterpriseSinks )
254306}
@@ -298,8 +350,10 @@ func TestDatabaseLevelChangefeedWithExcludeFilter(t *testing.T) {
298350 `foo4: [0]->{"after": {"a": 0, "b": "updated"}}` ,
299351 })
300352
301- // Test that we can handle fully qualfied, partially qualified, and unqualified table names.
302- feed4 := feed (t , f , `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES d.public.foo2, public.foo3, foo4` )
353+ // Test that we can handle fully qualified, partially qualified, and unqualified
354+ // table names.
355+ feed4 := feed (t , f ,
356+ `CREATE CHANGEFEED FOR DATABASE d EXCLUDE TABLES d.public.foo2, public.foo3, foo4` )
303357 defer closeFeed (t , feed4 )
304358 assertPayloads (t , feed4 , []string {
305359 `foo1: [0]->{"after": {"a": 0, "b": "initial"}}` ,
0 commit comments