changefeedccl: (wip to be squashed) add tracking in job frontier

Puts in a rough fix that involves updating the checkpoint
to reflect the new tableset of the DB level changefeed when
changing the filter by storing a resolved span for newly
watched tables at the timestamp of the alter changefeed
statement.

Epic: CRDB-1421

Release note: None

Author: aerfrei

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedvalidators"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobfrontier"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsauth"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -36,6 +37,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
 )
@@ -143,7 +145,7 @@ func alterChangefeedPlanHook(
 		if prevDetails.TargetSpecifications[0].Type == jobspb.ChangefeedTargetSpecification_DATABASE {
 			newChangefeedStmt.Level = tree.ChangefeedLevelDatabase
 			if len(prevDetails.TargetSpecifications) != 1 {
-				return errors.Errorf("database level changefeed must have exactly one target specification")
+				return errors.AssertionFailedf("database level changefeed must have exactly one target specification")
 			}
 		} else {
 			newChangefeedStmt.Level = tree.ChangefeedLevelTable
@@ -248,18 +250,48 @@ func alterChangefeedPlanHook(
 		}
 
 		newDetails := jobRecord.Details.(jobspb.ChangefeedDetails)
-		if newChangefeedStmt.Level != tree.ChangefeedLevelDatabase {
+		// newStatementTime will either be the StatementTime of the job prior to the
+		// alteration, or it will be the high watermark of the job.
+		newDetails.StatementTime = newStatementTime
+
+		if newChangefeedStmt.Level == tree.ChangefeedLevelDatabase {
+			// For database level changefeeds, we may have set a new value for
+			// the filter options. In this case, the set of tables we are watching
+			// may have changed. We need to store the new frontier so that we only
+			// emit events for the new tables from the time of the ALTER CHANGEFEED
+			// statement, but not before.
+
+			// NB: We do not need to remove spans for tables that are no longer
+			// watched. When the changefeed is resumed, we will make the frontier
+			// so that it will only watch the new set of watched tables. This means
+			// that if altering a filter results in us watching fewer tables,
+			// we will not emit ANY more events for that table, even if it was
+			// lagging, i.e. even if they corresponded to updates made before
+			// the ALTER CHANGEFEED statement.
+
+			// We compare the effect of the two filters at the time of the current
+			// high watermark. If the filter now includes a new table, but that
+			// table was created between the previous high watermark and the
+			// ALTER CHANGEFEED statement, then we will not store progress for it.
+			// Instead, the tableset watcher will pick it up once the high water
+			// advances to its creation time.
+			targetTS := newProgress.GetHighWater()
+			if targetTS == nil || targetTS.IsEmpty() {
+				targetTS = &prevDetails.StatementTime
+			}
+			prevTargets, err := AllTargets(ctx, prevDetails, p.ExecCfg(), *targetTS)
+			if err != nil {
+				return err
+			}
+			storeNewTablesJobFrontier(ctx, p, jobID, *targetTS, targets, prevTargets, newDetails, prevDetails)
+		} else {
 			// If the changefeed is a table-level changefeed, we set the initial
 			// scan option to its default value of "yes" so that we can do an
 			// initial scan for new targets. We do not allow initial scans on
 			// database level changefeeds.
 			newDetails.Opts[changefeedbase.OptInitialScan] = ``
 		}
 
-		// newStatementTime will either be the StatementTime of the job prior to the
-		// alteration, or it will be the high watermark of the job.
-		newDetails.StatementTime = newStatementTime
-
 		newPayload := job.Payload()
 		newPayload.Details = jobspb.WrapPayloadDetails(newDetails)
 		newPayload.Description = jobRecord.Description
@@ -346,10 +378,10 @@ func processFiltersForChangefeed(
 	alterCmds tree.AlterChangefeedCmds,
 	prevDetails jobspb.ChangefeedDetails,
 ) error {
-	filters := generateNewFilters(alterCmds)
+	filterCommands := generateNewFilters(alterCmds)
 
 	// Only DB-level feeds should have filters.
-	if len(filters) > 0 && stmt.Level != tree.ChangefeedLevelDatabase {
+	if len(filterCommands) > 0 && stmt.Level != tree.ChangefeedLevelDatabase {
 		return errors.Errorf("filters are only supported for database level changefeeds")
 	}
 
@@ -367,8 +399,8 @@ func processFiltersForChangefeed(
 	}
 
 	// For table-level feeds, this should be a no-op since there are no
-	// filters as asserted above. Otherwise, apply each filter in order.
-	for _, filter := range filters {
+	// filter commands as asserted above. Otherwise, apply each filter in order.
+	for _, filter := range filterCommands {
 		currentFilter := stmt.FilterOption
 		// Note that the no-filters state is represented by FilterType = exclude
 		// with an empty tables list. In that state, it's perfectly valid to
@@ -527,9 +559,9 @@ func generateAndValidateNewTargets(
 
 	// Since we do not allow initial scans on database level changefeeds,
 	// we do not need to call generateAndValidateNewTargetsForTableLevelFeed.
-	// When filter changes result in changes to which tables are being tracked,
-	// we can rely on startDistChangefeed to handle removing/adding tables to
-	// the progress.
+	// When filter changes result in our set of watched tables changing, though,
+	// we will also need to modify the progress, but we do that once we've created
+	// the new changefeed job record so that we have access to the new table targets.
 	return tree.ChangefeedTableTargets{}, &prevProgress, prevDetails.StatementTime,
 		make(map[tree.ChangefeedTableTarget]jobspb.ChangefeedTargetSpecification), nil
 }
@@ -978,7 +1010,6 @@ func generateNewProgress(
 		// TODO(#142376): Consider whether we want to set the new statement time
 		// to the actual new statement time (ALTER CHANGEFEED statement time).
 		newStatementTime := *prevHighWater
-
 		newProgress := jobspb.Progress{
 			Progress: &jobspb.Progress_HighWater{},
 			Details: &jobspb.Progress_Changefeed{
@@ -1032,6 +1063,51 @@ func generateNewProgress(
 	return newProgress, prevStatementTime, nil
 }
 
+// storeNewTablesJobFrontier uses the persistentjob frontier to ensure that when
+// we resume the changefeed, new tables being watched by the database level feed
+// (due to ALTER CHANGEFEED changing the filter options) are watched only from
+// the time of the ALTER CHANGEFEED statement.
+func storeNewTablesJobFrontier(
+	ctx context.Context, p sql.PlanHookState, jobID jobspb.JobID, ts hlc.Timestamp,
+	targets changefeedbase.Targets, prevTargets changefeedbase.Targets,
+	newDetails jobspb.ChangefeedDetails, prevDetails jobspb.ChangefeedDetails,
+) error {
+	getSpansForTargets := func(targets changefeedbase.Targets, details jobspb.ChangefeedDetails) ([]roachpb.Span, error) {
+		tableDescs, err := fetchTableDescriptors(ctx, p.ExecCfg(), targets, ts)
+		if err != nil {
+			return nil, err
+		}
+		spans, err := fetchSpansForTables(ctx, p, tableDescs, details, ts)
+		if err != nil {
+			return nil, err
+		}
+		return spans, nil
+	}
+
+	existingTargetSpans, err := getSpansForTargets(prevTargets, prevDetails)
+	if err != nil {
+		return err
+	}
+	trackedSpans, err := getSpansForTargets(targets, newDetails)
+	if err != nil {
+		return err
+	}
+
+	statementTime := hlc.Timestamp{WallTime: p.ExtendedEvalContext().GetStmtTimestamp().UnixNano()}
+	var newTrackedSpans roachpb.SpanGroup
+	newTrackedSpans.Add(trackedSpans...)
+	newTrackedSpans.Sub(existingTargetSpans...)
+
+	frontier, err := span.MakeFrontierAt(statementTime, newTrackedSpans.Slice()...)
+	if err != nil {
+		return err
+	}
+
+	if err := jobfrontier.Store(ctx, p.InternalSQLTxn(), jobID, `alter_changefeed`, frontier); err != nil {
+		return err
+	}
+	return nil
+}
 func removeSpansFromProgress(progress jobspb.Progress, spansToRemove []roachpb.Span) error {
 	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(progress)
 	if err != nil {

pkg/ccl/changefeedccl/alter_changefeed_test.go

@@ -1574,6 +1574,12 @@ func TestAlterChangefeedDatabaseLevelChangefeedFilterSemantics(t *testing.T) {
 			// since the filter check passes when tables list is empty.
 			expectPayloads: map[string]bool{"foo": true, "bar": false, "baz": true},
 		},
+		{
+			name:           "unset_include_then_set_exclude",
+			createStmt:     "CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo",
+			alters:         []alterSpec{{stmt: "UNSET INCLUDE TABLES"}, {stmt: "SET EXCLUDE TABLES bar"}},
+			expectPayloads: map[string]bool{"foo": true, "bar": false, "baz": true},
+		},
 		{
 			name:       "set_exclude_then_unset_include_fails",
 			createStmt: "CREATE CHANGEFEED FOR DATABASE d",
@@ -1642,6 +1648,63 @@ func TestAlterChangefeedDatabaseLevelChangefeedFilterSemantics(t *testing.T) {
 	}
 }
 
+// TestAlterChangefeedFilterChangesDontEmitPreAlterEvents tests that if
+// we alter a changefeed's filter options in a way that means new tables are
+// being watched that we do not emit events for the new tables from before the
+// ALTER CHANGEFEED statement.
+func TestAlterChangefeedFilterChangesDontEmitPreAlterEvents(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 TABLE foo (a INT PRIMARY KEY, b STRING)`)
+		sqlDB.Exec(t, `CREATE TABLE bar (a INT PRIMARY KEY, b STRING)`)
+
+		// If the checkpoint advances, that will be the point we resume from for
+		// all tables, including 'bar' skipping the pre-alter event. This makes
+		// the test fail more consistently if we didn't add 'bar' to the frontier.
+		knobs := s.TestingKnobs.
+			DistSQL.(*execinfra.TestingKnobs).
+			Changefeed.(*TestingKnobs)
+		knobs.ShouldCheckpointToJobRecord = func(hw hlc.Timestamp) bool {
+			return false
+		}
+
+		testFeed := feed(t, f, `CREATE CHANGEFEED FOR DATABASE d INCLUDE TABLES foo`)
+		defer closeFeed(t, testFeed)
+
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1, 'before alter')`)
+		sqlDB.Exec(t, `INSERT INTO bar VALUES (1, 'before alter')`)
+
+		assertPayloads(t, testFeed, []string{
+			`foo: [1]->{"after": {"a": 1, "b": "before alter"}}`,
+		})
+
+		sqlDB.Exec(t, fmt.Sprintf(`PAUSE JOB %d`, testFeed.(cdctest.EnterpriseTestFeed).JobID()))
+		waitForJobState(sqlDB, t, testFeed.(cdctest.EnterpriseTestFeed).JobID(), `paused`)
+
+		sqlDB.Exec(t, fmt.Sprintf(`ALTER CHANGEFEED %d UNSET INCLUDE TABLES`,
+			testFeed.(cdctest.EnterpriseTestFeed).JobID()))
+
+		sqlDB.Exec(t, fmt.Sprintf(`RESUME JOB %d`, testFeed.(cdctest.EnterpriseTestFeed).JobID()))
+		waitForJobState(sqlDB, t, testFeed.(cdctest.EnterpriseTestFeed).JobID(), `running`)
+
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (2, 'after alter')`)
+		sqlDB.Exec(t, `INSERT INTO bar VALUES (2, 'after alter')`)
+
+		// We should only see the event from bar that was inserted after the alter,
+		// but we should see both events from foo. This test may falsely pass if
+		// we "get lucky" and the pre-alter event for bar is the last one emitted.
+		assertPayloads(t, testFeed, []string{
+			`foo: [2]->{"after": {"a": 2, "b": "after alter"}}`,
+			`bar: [2]->{"after": {"a": 2, "b": "after alter"}}`,
+		})
+	}
+
+	cdcTest(t, testFn, feedTestEnterpriseSinks)
+}
+
 func TestAlterChangefeedDatabaseScopeUnqualifiedName(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -254,9 +254,11 @@ func startDistChangefeed(
 
 	dsp := execCtx.DistSQLPlanner()
 
+	// ??
 	var spanLevelCheckpoint *jobspb.TimestampSpansMap
 	if progress := localState.progress.GetChangefeed(); progress != nil && progress.SpanLevelCheckpoint != nil {
 		spanLevelCheckpoint = progress.SpanLevelCheckpoint
+		fmt.Println("spanLevelCheckpoint starting for changefeed", spanLevelCheckpoint)
 		if log.V(2) {
 			log.Changefeed.Infof(ctx, "span-level checkpoint: %s", spanLevelCheckpoint)
 		}