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

@@ -13483,3 +13483,38 @@ 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, `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

@@ -323,6 +323,22 @@ func (w *Watcher) Start(ctx context.Context, initialTS hlc.Timestamp) (retErr er
 				if !ok {
 					return nil
 				}
+			} else {
+				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", w.dbName, w.filter.DatabaseID),
+					)
+				}
+
+				return nil
 			}
 
 			if log.V(2) {