changefeedccl: block db-level changefeed creation for non-empty tableset

On startup, the changefeed should not be started until there exist
tables in the database to watch. If no tables exist at the time of
startup, then a tableset watcher is created and polled; the changefeed
will be started from where there is first a target table.

Fixes: #147371
Epic: CRDB-1421

Release note: none

Author: log-head

pkg/ccl/changefeedccl/BUILD.bazel

@@ -64,6 +64,7 @@ go_library(
         "//pkg/ccl/changefeedccl/kvfeed",
         "//pkg/ccl/changefeedccl/resolvedspan",
         "//pkg/ccl/changefeedccl/schemafeed",
+        "//pkg/ccl/changefeedccl/tableset",
         "//pkg/ccl/changefeedccl/timers",
         "//pkg/ccl/kvccl/kvfollowerreadsccl",
         "//pkg/ccl/utilccl",

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -18,6 +18,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedvalidators"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/checkpoint"
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/tableset"
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn"
@@ -40,6 +41,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/exprutil"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
@@ -59,6 +61,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log/channel"
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
 	"github.com/cockroachdb/cockroach/pkg/util/log/severity"
+	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
@@ -1724,6 +1727,9 @@ func (b *changefeedResumer) resumeWithRetries(
 
 	maxBackoff := changefeedbase.MaxRetryBackoff.Get(&execCfg.Settings.SV)
 	backoffReset := changefeedbase.RetryBackoffReset.Get(&execCfg.Settings.SV)
+
+	memMonitor := execinfra.NewMonitor(ctx, execCfg.DistSQLSrv.ChangefeedMonitor, mon.MakeName("changefeed-empty-watcher-mem"))
+	defer memMonitor.Stop(ctx)
 	for r := getRetry(ctx, maxBackoff, backoffReset); r.Next(); {
 		flowErr := maybeUpgradePreProductionReadyExpression(ctx, jobID, details, jobExec)
 
@@ -1737,7 +1743,7 @@ func (b *changefeedResumer) resumeWithRetries(
 				knobs.BeforeDistChangefeed()
 			}
 
-			confPoller := make(chan struct{})
+			runningChangefeedChan := make(chan struct{})
 			g := ctxgroup.WithContext(ctx)
 			initialHighWater, schemaTS, err := computeDistChangefeedTimestamps(ctx, jobExec, details, localState)
 			if err != nil {
@@ -1751,28 +1757,92 @@ func (b *changefeedResumer) resumeWithRetries(
 			if haveKnobs && maybeCfKnobs.TablesetChangingCallback != nil {
 				maybeCfKnobs.TablesetChangingCallback()
 			}
+
+			// Watcher is only used for db-level changefeeds with no watched tables.
+			var watcher *tableset.Watcher
+			watcherChan := make(chan []tableset.TableDiff, 1)
+			if len(details.TargetSpecifications) == 1 &&
+				details.TargetSpecifications[0].Type == jobspb.ChangefeedTargetSpecification_DATABASE &&
+				targets.NumUniqueTables() == 0 {
+
+				// Create watcher dependencies.
+				dbID := details.TargetSpecifications[0].DescID
+				filter := tableset.Filter{
+					DatabaseID: dbID,
+				}
+				var initialWatcherTS hlc.Timestamp
+				if h := localState.progress.GetHighWater(); h != nil && !h.IsEmpty() {
+					initialWatcherTS = *h
+				} else {
+					initialWatcherTS = details.StatementTime
+				}
+
+				// Create a watcher for the database.
+				watcher = tableset.NewWatcher(filter, execCfg, memMonitor, int64(jobID))
+				timestamp := initialWatcherTS
+				g.GoCtx(func(ctx context.Context) error {
+					// defer close(watcherChan)
+					return watcher.Start(ctx, timestamp)
+				})
+				// } else {
+				// 	close(watcherChan)
+			}
+
+			// Start the changefeed.
 			g.GoCtx(func(ctx context.Context) error {
-				defer close(confPoller)
+				defer close(runningChangefeedChan)
+				var schemaTSOverride hlc.Timestamp
+				if targets.NumUniqueTables() == 0 {
+					select {
+					case <-ctx.Done():
+						return ctx.Err()
+					case diffs := <-watcherChan:
+						// Get the diff with the earliest timestamp.
+						schemaTSOverride = diffs[0].AsOf
+						schemaTS, initialHighWater = schemaTSOverride, schemaTSOverride
+						targets, err = AllTargets(ctx, details, execCfg, schemaTSOverride)
+						if err != nil {
+							return err
+						}
+					}
+				}
+				// return distChangefeedFlow(ctx, jobExec, jobID, details, description, localState, startedCh, onTracingEvent, targets, schemaTSOverride)
 				return startDistChangefeed(ctx, jobExec, jobID, schemaTS, details, description, initialHighWater, localState, startedCh, onTracingEvent, targets)
-
 			})
+
+			// Poll for updated configuration or new database tables if hibernating.
 			g.GoCtx(func(ctx context.Context) error {
 				t := time.NewTicker(15 * time.Second)
 				defer t.Stop()
+				diffSent := false
 				for {
 					select {
 					case <-ctx.Done():
 						return ctx.Err()
-					case <-confPoller:
+					case <-runningChangefeedChan:
 						return nil
-					case <-t.C:
+					case tick := <-t.C:
 						newDest, err := reloadDest(ctx, jobID, execCfg)
 						if err != nil {
 							log.Changefeed.Warningf(ctx, "failed to check for updated configuration: %v", err)
 						} else if newDest != resolvedDest {
 							resolvedDest = newDest
 							return replanErr
 						}
+						if watcher != nil && !diffSent {
+							unchanged, diffs, err := watcher.PopUnchangedUpTo(ctx, hlc.Timestamp{WallTime: tick.UnixNano()})
+							if err != nil {
+								return err
+							}
+							if !unchanged {
+								select {
+								case watcherChan <- diffs:
+									diffSent = true
+								case <-ctx.Done():
+									return ctx.Err()
+								}
+							}
+						}
 					}
 				}
 			})

pkg/ccl/changefeedccl/changefeed_test.go

@@ -12933,3 +12933,44 @@ func TestDatabaseLevelChangefeedChangingTableset(t *testing.T) {
 	}
 	cdcTest(t, testFnAddEnriched, withKnobsFn(knobsFn), feedTestForceSink("kafka"))
 }
+
+func TestDatabaseLevelChangefeedEmptyTableset(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFnNoWait := 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`)
+		dbcf := feed(t, f, `CREATE CHANGEFEED FOR DATABASE db`)
+		defer closeFeed(t, dbcf)
+
+		// create a table
+		sqlDB.Exec(t, `CREATE TABLE db.foo (a INT PRIMARY KEY, b STRING)`)
+		sqlDB.Exec(t, `INSERT INTO db.foo VALUES (0, 'initial')`)
+
+		assertPayloads(t, dbcf, []string{
+			`foo: [0]->{"after": {"a": 0, "b": "initial"}}`,
+		})
+	}
+	cdcTest(t, testFnNoWait, feedTestForceSink("enterprise"))
+
+	testFnWait := 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`)
+		dbcf := feed(t, f, `CREATE CHANGEFEED FOR DATABASE db`)
+		defer closeFeed(t, dbcf)
+
+		time.Sleep(5 * time.Second)
+
+		// create a table
+		sqlDB.Exec(t, `CREATE TABLE db.foo (a INT PRIMARY KEY, b STRING)`)
+		sqlDB.Exec(t, `INSERT INTO db.foo VALUES (0, 'initial')`)
+
+		assertPayloads(t, dbcf, []string{
+			`foo: [0]->{"after": {"a": 0, "b": "initial"}}`,
+		})
+	}
+	cdcTest(t, testFnWait, feedTestForceSink("enterprise"))
+}