changefeedccl: db-level feeds: fail the feed if the watched db is dropped

This patch fails a db-level changefeed when its watched database is
dropped.

Epic: CRDB-55919
Fixes: #156780

Release note (sql change): A database-level changefeed will now fail
when its watched database is dropped.

Author: log-head

pkg/ccl/changefeedccl/changefeed_test.go

@@ -12842,21 +12842,15 @@ func TestDatabaseLevelChangefeedRenameDatabase(t *testing.T) {
 		sqlDB.Exec(t, `CREATE DATABASE foo;`)
 		sqlDB.Exec(t, `CREATE TABLE foo.bar (id INT PRIMARY KEY);`)
 
-		feed1 := feed(t, f, `CREATE CHANGEFEED FOR DATABASE foo`)
-		defer closeFeed(t, feed1)
+		feed := feed(t, f, `CREATE CHANGEFEED FOR DATABASE foo`)
 
-		sqlDB.Exec(t, `INSERT INTO foo.bar VALUES (1);`)
-		expectedRows := []string{
-			`bar: [1]->{"after": {"id": 1}}`,
-		}
-		assertPayloads(t, feed1, expectedRows)
+		enterpriseFeed := feed.(cdctest.EnterpriseTestFeed)
+		waitForJobState(sqlDB, t, enterpriseFeed.JobID(), jobs.StateRunning)
 
 		sqlDB.Exec(t, `ALTER DATABASE foo RENAME TO bar;`)
-		sqlDB.Exec(t, `INSERT INTO bar.bar VALUES (2);`)
-		expectedRows = []string{
-			`bar: [2]->{"after": {"id": 2}}`,
-		}
-		assertPayloads(t, feed1, expectedRows)
+		waitForJobState(sqlDB, t, enterpriseFeed.JobID(), jobs.StateFailed)
+		require.NoError(t, feed.Close())
+		require.Error(t, enterpriseFeed.FetchTerminalJobErr())
 	}
 	// TODO(#152196): Remove feedTestUseRootUserConnection once we have ALTER
 	// DEFAULT PRIVILEGES for databases
@@ -13488,3 +13482,42 @@ func TestChangefeedWatcherCleanupOnStop(t *testing.T) {
 
 	cdcTest(t, testFn, feedTestEnterpriseSinks)
 }
+
+func TestDatabaseLevelChangefeedDropDatabase(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 DATABASE db`)
+		sqlDB.Exec(t, `GRANT CHANGEFEED ON DATABASE db TO enterprisefeeduser`)
+		sqlDB.Exec(t, `CREATE TABLE db.foo (a INT PRIMARY KEY, b STRING)`)
+		feed := feed(t, f, `CREATE CHANGEFEED FOR DATABASE db`)
+		enterpriseFeed := feed.(cdctest.EnterpriseTestFeed)
+		waitForJobState(sqlDB, t, enterpriseFeed.JobID(), jobs.StateRunning)
+		sqlDB.Exec(t, `insert into db.foo values (1, 'test')`)
+		assertPayloads(t, feed, []string{
+			`foo: [1]->{"after": {"a": 1, "b": "test"}}`,
+		})
+		sqlDB.Exec(t, `DROP DATABASE db`)
+		waitForJobState(sqlDB, t, enterpriseFeed.JobID(), jobs.StateFailed)
+		require.NoError(t, feed.Close())
+		require.Error(t, enterpriseFeed.FetchTerminalJobErr())
+	}
+	cdcTest(t, testFn, feedTestEnterpriseSinks, withAllowChangefeedErr(("feed should fail when database is dropped")))
+
+	testEmptyTablesetFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		sqlDB.Exec(t, `CREATE DATABASE db`)
+		sqlDB.Exec(t, `GRANT CHANGEFEED ON DATABASE db TO enterprisefeeduser`)
+		feed := feed(t, f, `CREATE CHANGEFEED FOR DATABASE db WITH hibernation_polling_frequency='1s'`)
+		enterpriseFeed := feed.(cdctest.EnterpriseTestFeed)
+		waitForJobState(sqlDB, t, enterpriseFeed.JobID(), jobs.StateRunning)
+		time.Sleep(5 * time.Second)
+		sqlDB.Exec(t, `DROP DATABASE db`)
+		waitForJobState(sqlDB, t, enterpriseFeed.JobID(), jobs.StateFailed)
+		require.NoError(t, feed.Close())
+		require.Error(t, enterpriseFeed.FetchTerminalJobErr())
+	}
+	cdcTest(t, testEmptyTablesetFn, feedTestEnterpriseSinks, withAllowChangefeedErr(("feed should fail when database is dropped")))
+}

pkg/ccl/changefeedccl/tableset/watcher.go

@@ -315,6 +315,20 @@ func (w *Watcher) Start(ctx context.Context, initialTS hlc.Timestamp) (retErr er
 			var err error
 			var ok bool
 
+			nameInfo, err := catalogkeys.DecodeNameMetadataKey(w.execCfg.Codec, kv.Key)
+			if err != nil {
+				return err
+			}
+
+			// Database-level namespace entry tombstone.
+			if nameInfo.ParentID == 0 && nameInfo.ParentSchemaID == 0 &&
+				tree.Name(nameInfo.Name) == w.dbName {
+				// Fail the watcher/changefeed: the target database entry was removed.
+				return changefeedbase.WithTerminalError(
+					errors.Newf("target database %q (id %d) dropped or renamed", w.dbName, w.filter.DatabaseID),
+				)
+			}
+
 			if kv.Value.IsPresent() {
 				table, ok, err = w.kvToTable(ctx, roachpb.KeyValue{Key: kv.Key, Value: kv.Value}, dec)
 				if err != nil {

pkg/ccl/changefeedccl/tableset/watcher_test.go

@@ -381,12 +381,12 @@ func TestWatcherDropDatabase(t *testing.T) {
 
 	db.Exec(t, "DROP DATABASE defaultdb")
 
-	adds, err := watcher.PopUpTo(ctx, hlc.Timestamp{WallTime: timeutil.Now().UnixNano()})
-	require.NoError(t, err)
-	assert.Empty(t, adds)
+	_, err = watcher.PopUpTo(ctx, hlc.Timestamp{WallTime: timeutil.Now().UnixNano()})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "dropped")
 
-	cancel()
-	require.ErrorIs(t, eg.Wait(), context.Canceled)
+	err = eg.Wait()
+	require.Contains(t, err.Error(), "dropped")
 }
 
 func getDatabaseID(