@@ -12913,3 +12913,170 @@ func TestDatabaseLevelChangefeedChangingTableset(t *testing.T) {
1291312913 }
1291412914 })
1291512915}
12916+
12917+ // TestDatabaseLevelChangefeedEmptyTableset tests that a database-level changefeed
12918+ // hibernates while there are no tables in the database.
12919+ func TestDatabaseLevelChangefeedEmptyTableset (t * testing.T ) {
12920+ defer leaktest .AfterTest (t )()
12921+ defer log .Scope (t ).Close (t )
12922+
12923+ testFnNoWait := func (t * testing.T , s TestServer , f cdctest.TestFeedFactory ) {
12924+ sqlDB := sqlutils .MakeSQLRunner (s .DB )
12925+ sqlDB .Exec (t , `CREATE DATABASE db` )
12926+ sqlDB .Exec (t , `GRANT CHANGEFEED ON DATABASE db TO enterprisefeeduser` )
12927+ dbcf := feed (t , f , `CREATE CHANGEFEED FOR DATABASE db` )
12928+ defer closeFeed (t , dbcf )
12929+
12930+ // create a table
12931+ sqlDB .Exec (t , `CREATE TABLE db.foo (a INT PRIMARY KEY, b STRING)` )
12932+ sqlDB .Exec (t , `INSERT INTO db.foo VALUES (0, 'initial')` )
12933+
12934+ assertPayloads (t , dbcf , []string {
12935+ `foo: [0]->{"after": {"a": 0, "b": "initial"}}` ,
12936+ })
12937+ }
12938+ cdcTest (t , testFnNoWait , feedTestEnterpriseSinks )
12939+
12940+ testFnWait := func (t * testing.T , s TestServer , f cdctest.TestFeedFactory ) {
12941+ sqlDB := sqlutils .MakeSQLRunner (s .DB )
12942+ sqlDB .Exec (t , `CREATE DATABASE db` )
12943+ sqlDB .Exec (t , `GRANT CHANGEFEED ON DATABASE db TO enterprisefeeduser` )
12944+ dbcf := feed (t , f , `CREATE CHANGEFEED FOR DATABASE db` )
12945+ defer closeFeed (t , dbcf )
12946+
12947+ time .Sleep (5 * time .Second )
12948+
12949+ // create a table
12950+ sqlDB .Exec (t , `CREATE TABLE db.foo (a INT PRIMARY KEY, b STRING)` )
12951+ sqlDB .Exec (t , `INSERT INTO db.foo VALUES (0, 'initial')` )
12952+
12953+ assertPayloads (t , dbcf , []string {
12954+ `foo: [0]->{"after": {"a": 0, "b": "initial"}}` ,
12955+ })
12956+ }
12957+ cdcTest (t , testFnWait , feedTestEnterpriseSinks )
12958+ }
12959+
12960+ // TestDatabaseLevelChangefeedFiltersHibernation tests that a database-level changefeed
12961+ // with include/exclude filters correctly handles hibernation:
12962+ // - With EXCLUDE filter: creating an excluded table should not wake the changefeed,
12963+ // but creating a non-excluded table should wake it.
12964+ // - With INCLUDE filter: creating a non-included table should not wake the changefeed,
12965+ // but creating an included table should wake it.
12966+ func TestDatabaseLevelChangefeedFiltersHibernation (t * testing.T ) {
12967+ defer leaktest .AfterTest (t )()
12968+ defer log .Scope (t ).Close (t )
12969+
12970+ full_filter := map [string ]string {
12971+ "include" : "EXCLUDE TABLES excluded_table" ,
12972+ "exclude" : "INCLUDE TABLES included_table" ,
12973+ }
12974+ testFn := func (t * testing.T , s TestServer , f cdctest.TestFeedFactory , filterType string ) {
12975+ sqlDB := sqlutils .MakeSQLRunner (s .DB )
12976+ sqlDB .Exec (t , `CREATE DATABASE db` )
12977+ sqlDB .Exec (t , `GRANT CHANGEFEED ON DATABASE db TO enterprisefeeduser` )
12978+
12979+ // Create changefeed with exclude filter - should hibernate since no tables exist
12980+ createStmt := fmt .Sprintf (`CREATE CHANGEFEED FOR DATABASE db %s` , full_filter [filterType ])
12981+ dbcf := feed (t , f , createStmt )
12982+ defer closeFeed (t , dbcf )
12983+
12984+ var jobID jobspb.JobID
12985+ if ef , ok := dbcf .(cdctest.EnterpriseTestFeed ); ok {
12986+ jobID = ef .JobID ()
12987+ } else {
12988+ t .Fatal ("expected EnterpriseTestFeed" )
12989+ }
12990+
12991+ // Get initial diagram count (should be 0 when hibernating, as no diagram is written yet)
12992+ getDiagramCount := func () int {
12993+ var count int
12994+ sqlDB .QueryRow (t ,
12995+ `SELECT count(*) FROM system.job_info WHERE job_id = $1 AND info_key LIKE '~dsp-diag-url-%'` ,
12996+ jobID ,
12997+ ).Scan (& count )
12998+ return count
12999+ }
13000+
13001+ testutils .SucceedsSoon (t , func () error {
13002+ var count int
13003+ sqlDB .QueryRow (t ,
13004+ `SELECT count(*) FROM [SHOW CHANGEFEED JOB $1] WHERE running_status = 'running'` ,
13005+ jobID ,
13006+ ).Scan (& count )
13007+ return nil
13008+ })
13009+ require .Equal (t , 0 , getDiagramCount (), "changefeed should be hibernating (no diagram written)" )
13010+ time .Sleep (20 * time .Second )
13011+ require .Equal (t , 0 , getDiagramCount (), "changefeed should stay hibernating (no diagram written)" )
13012+
13013+ // Create a table that is excluded - changefeed should stay hibernating
13014+ sqlDB .Exec (t , `CREATE TABLE db.excluded_table (a INT PRIMARY KEY, b STRING)` )
13015+ sqlDB .Exec (t , `INSERT INTO db.excluded_table VALUES (0, 'excluded')` )
13016+
13017+ // // Verify changefeed stays hibernating (no new diagram written)
13018+ time .Sleep (20 * time .Second )
13019+ require .Equal (t , 0 , getDiagramCount (), "changefeed should stay hibernating (no new diagram written)" )
13020+
13021+ // Create a table that is NOT excluded - changefeed should wake up
13022+ sqlDB .Exec (t , `CREATE TABLE db.included_table (a INT PRIMARY KEY, b STRING)` )
13023+ sqlDB .Exec (t , `INSERT INTO db.included_table VALUES (0, 'included')` )
13024+
13025+ // Wait for a new diagram to be written (indicating changefeed woke up)
13026+ time .Sleep (20 * time .Second )
13027+ require .Equal (t , 1 , getDiagramCount (), "changefeed should wake up (new diagram written)" )
13028+
13029+ // Should only receive events from the included table
13030+ assertPayloads (t , dbcf , []string {
13031+ `included_table: [0]->{"after": {"a": 0, "b": "included"}}` ,
13032+ })
13033+ }
13034+ testutils .RunValues (t , "filterType" , []string {"include" , "exclude" }, func (t * testing.T , filterType string ) {
13035+ runTestFn := func (t * testing.T , s TestServer , f cdctest.TestFeedFactory ) {
13036+ testFn (t , s , f , filterType )
13037+ }
13038+ cdcTest (t , runTestFn , feedTestEnterpriseSinks )
13039+ })
13040+ }
13041+
13042+ // TestChangefeedWatcherCleanupOnStop verifies that the watcher context is properly
13043+ // cleaned up when a changefeed is stopped before receiving any table diffs.
13044+ // This is a regression test for context leaks in the watcher lifecycle.
13045+ func TestChangefeedWatcherCleanupOnStop (t * testing.T ) {
13046+ defer leaktest .AfterTest (t )()
13047+ defer log .Scope (t ).Close (t )
13048+
13049+ testFn := func (t * testing.T , s TestServer , f cdctest.TestFeedFactory ) {
13050+ sqlDB := sqlutils .MakeSQLRunner (s .DB )
13051+
13052+ // Create a database with no tables. This will trigger the watcher
13053+ // since the changefeed targets the database but there are no tables yet.
13054+ sqlDB .Exec (t , `CREATE DATABASE db` )
13055+ sqlDB .Exec (t , `GRANT CHANGEFEED ON DATABASE db TO enterprisefeeduser` )
13056+
13057+ // Create a changefeed on the empty database. This will start the watcher
13058+ // but won't receive any diffs until a table is created.
13059+ feed , err := f .Feed (`CREATE CHANGEFEED FOR DATABASE db` )
13060+ require .NoError (t , err )
13061+ enterpriseFeed := feed .(cdctest.EnterpriseTestFeed )
13062+
13063+ // Wait for the changefeed job to be running to ensure the watcher
13064+ // goroutine has started.
13065+ waitForJobState (sqlDB , t , enterpriseFeed .JobID (), jobs .StateRunning )
13066+
13067+ // Cancel the changefeed job before any tables are created. This tests the
13068+ // scenario where the changefeed is stopped before the watcher receives diffs,
13069+ // which previously could lead to a context leak.
13070+ sqlDB .Exec (t , `CANCEL JOB $1` , enterpriseFeed .JobID ())
13071+
13072+ // Wait for the job to be canceled.
13073+ waitForJobState (sqlDB , t , enterpriseFeed .JobID (), jobs .StateCanceled )
13074+
13075+ // Close the feed to ensure all resources are released.
13076+ require .NoError (t , feed .Close ())
13077+
13078+ // The leaktest.AfterTest will verify that no goroutines or contexts leaked.
13079+ }
13080+
13081+ cdcTest (t , testFn , feedTestEnterpriseSinks )
13082+ }
0 commit comments