changefeedccl: support altering filters on db-level feeds

This commit allows users to change filters on existing
DB-level changefeeds with commands like ALTER CHANGEFEED
<job_id> SET INCLUDE TABLES (including EXCLUDE or UNSET).

This also implements errors when trying to set filters
on non-DB-level feeds, ADD/DROP targets on DB-level feeds,
setting both INCLUDE and EXCLUDE at the same time, and
switching between them without first doing an explicit
unset.

Note that, when setting an include filter on a feed that
already has a filter set, tables not explicitly
respecified in the ALTER CHANGEFEED command will no longer
be included (or excluded) by the filter.

When filter changes result in new tables being watched,
they will only emit changes from the time the filter was
altered.

Epic: CRDB-1421
Fixes: #156484
Fixes: #155986
Informs: #155708
Release note: none

Author: aerfrei

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -9,13 +9,15 @@ import (
 	"context"
 	"maps"
 	"net/url"
+	"slices"
 
 	"github.com/cockroachdb/cockroach/pkg/backup/backupresolver"
 	"github.com/cockroachdb/cockroach/pkg/build"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdceval"
 	"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"
@@ -35,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"
 )
@@ -55,6 +58,9 @@ func alterChangefeedTypeCheck(
 	toCheck := []exprutil.ToTypeCheck{
 		exprutil.Ints{alterChangefeedStmt.Jobs},
 	}
+	// TODO(#156806): Validate the options for ADD and UNSET alter commands
+	// to fail when 'PREPARE'ing an ALTER CHANGEFEED statement specifying an
+	// invalid option.
 	for _, cmd := range alterChangefeedStmt.Cmds {
 		switch v := cmd.(type) {
 		case *tree.AlterChangefeedSetOptions:
@@ -129,6 +135,14 @@ func alterChangefeedPlanHook(
 		}
 
 		newChangefeedStmt := &tree.CreateChangefeed{}
+		if isDBLevelChangefeed(prevDetails) {
+			newChangefeedStmt.Level = tree.ChangefeedLevelDatabase
+			if len(prevDetails.TargetSpecifications) != 1 {
+				return errors.AssertionFailedf("database level changefeed must have exactly one target specification")
+			}
+		} else {
+			newChangefeedStmt.Level = tree.ChangefeedLevelTable
+		}
 
 		prevOpts, err := getPrevOpts(job.Payload().Description, prevDetails.Opts)
 		if err != nil {
@@ -152,6 +166,7 @@ func alterChangefeedPlanHook(
 
 		newTargets, newProgress, newStatementTime, originalSpecs, err := generateAndValidateNewTargets(
 			ctx, exprEval, p,
+			newChangefeedStmt.Level,
 			alterChangefeedStmt.Cmds,
 			newOptions,
 			prevDetails, job.Progress(),
@@ -160,7 +175,14 @@ func alterChangefeedPlanHook(
 		if err != nil {
 			return err
 		}
-		newChangefeedStmt.TableTargets = newTargets
+
+		if err := setChangefeedTargets(ctx, p, newChangefeedStmt, prevDetails, newTargets); err != nil {
+			return err
+		}
+
+		if err := processFiltersForChangefeed(newChangefeedStmt, alterChangefeedStmt.Cmds, prevDetails); err != nil {
+			return err
+		}
 
 		if prevDetails.Select != "" {
 			query, err := cdceval.ParseChangefeedExpression(prevDetails.Select)
@@ -218,12 +240,31 @@ func alterChangefeedPlanHook(
 		}
 
 		newDetails := jobRecord.Details.(jobspb.ChangefeedDetails)
-		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
 
+		if newChangefeedStmt.Level == tree.ChangefeedLevelDatabase {
+			// NB: We do not need to remove spans for tables that are no longer
+			// watched from job progress, only to add spans for new tables.
+			// 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.
+			if err := storeAlterChangefeedFrontier(
+				ctx, p, jobID, newDetails, prevDetails,
+			); err != nil {
+				return err
+			}
+		} 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.
+			newDetails.Opts[changefeedbase.OptInitialScan] = ``
+		}
+
 		newPayload := job.Payload()
 		newPayload.Details = jobspb.WrapPayloadDetails(newDetails)
 		newPayload.Description = jobRecord.Description
@@ -268,6 +309,80 @@ func alterChangefeedPlanHook(
 	return fn, alterChangefeedHeader, false, nil
 }
 
+// setChangefeedTargets sets the appropriate target fields on the changefeed
+// statement based on the feed level. Database-level feeds get a DatabaseTarget,
+// while table-level feeds get TableTargets.
+func setChangefeedTargets(
+	ctx context.Context,
+	p sql.PlanHookState,
+	stmt *tree.CreateChangefeed,
+	prevDetails jobspb.ChangefeedDetails,
+	newTargets tree.ChangefeedTableTargets,
+) error {
+	if stmt.Level == tree.ChangefeedLevelDatabase {
+		targetSpec := prevDetails.TargetSpecifications[0]
+		txn := p.InternalSQLTxn()
+		databaseDescriptor, err := txn.Descriptors().ByIDWithLeased(txn.KV()).Get().Database(ctx, targetSpec.DescID)
+		if err != nil {
+			return err
+		}
+		stmt.DatabaseTarget = tree.ChangefeedDatabaseTarget(databaseDescriptor.GetName())
+	} else {
+		stmt.TableTargets = newTargets
+	}
+	return nil
+}
+
+// processFiltersForChangefeed validates and applies filter changes to the
+// changefeed statement. Filters are only supported for database-level feeds.
+func processFiltersForChangefeed(
+	stmt *tree.CreateChangefeed,
+	alterCmds tree.AlterChangefeedCmds,
+	prevDetails jobspb.ChangefeedDetails,
+) error {
+	filterCommands := generateNewFilters(alterCmds)
+
+	// Only DB-level feeds should have filters.
+	if len(filterCommands) > 0 && stmt.Level != tree.ChangefeedLevelDatabase {
+		return errors.Errorf("filters are only supported for database level changefeeds")
+	}
+
+	// We add the existing filter state to the changefeed statement for
+	// DB-level feeds. This ensures that existing filters are preserved when
+	// we alter options unrelated to filters and that we can error if we try
+	// to set both include and exclude filters at the same time.
+	if stmt.Level == tree.ChangefeedLevelDatabase {
+		targetSpec := prevDetails.TargetSpecifications[0]
+		filterOpt, err := parseFilterOptionFromTargetSpec(targetSpec)
+		if err != nil {
+			return err
+		}
+		stmt.FilterOption = filterOpt
+	}
+
+	// For table-level feeds, this should be a no-op since there are no
+	// 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
+		// set or unset an Include filter.
+		if len(currentFilter.Tables) > 0 && currentFilter.FilterType != filter.FilterType {
+			return errors.Errorf("cannot alter filter type from %s to %s", currentFilter.FilterType, filter.FilterType)
+		}
+		if len(filter.Tables) > 0 {
+			stmt.FilterOption = tree.ChangefeedFilterOption{
+				FilterType: filter.FilterType,
+				Tables:     filter.Tables,
+			}
+		} else {
+			stmt.FilterOption = tree.DefaultChangefeedFilterOption()
+		}
+	}
+
+	return nil
+}
+
 func getTargetDesc(
 	ctx context.Context,
 	p sql.PlanHookState,
@@ -375,6 +490,7 @@ func generateAndValidateNewTargets(
 	ctx context.Context,
 	exprEval exprutil.Evaluator,
 	p sql.PlanHookState,
+	level tree.ChangefeedLevel,
 	alterCmds tree.AlterChangefeedCmds,
 	opts changefeedbase.StatementOptions,
 	prevDetails jobspb.ChangefeedDetails,
@@ -387,7 +503,42 @@ func generateAndValidateNewTargets(
 	map[tree.ChangefeedTableTarget]jobspb.ChangefeedTargetSpecification,
 	error,
 ) {
+	if level == tree.ChangefeedLevelTable {
+		return generateAndValidateNewTargetsForTableLevelFeed(
+			ctx, exprEval, p, alterCmds, opts, prevDetails, prevProgress, sinkURI,
+		)
+	}
+	// Database-level feeds do not support ADD/DROP target commands, so we validate
+	// that we haven't set any, otherwise, this is a no-op.
+	isAlteringTargets := slices.ContainsFunc(alterCmds, func(cmd tree.AlterChangefeedCmd) bool {
+		_, isAdd := cmd.(*tree.AlterChangefeedAddTarget)
+		_, isDrop := cmd.(*tree.AlterChangefeedDropTarget)
+		return isAdd || isDrop
+	})
+	if isAlteringTargets {
+		return nil, nil, hlc.Timestamp{}, nil, errors.Errorf("cannot alter targets for a database level changefeed")
+	}
 
+	return tree.ChangefeedTableTargets{}, &prevProgress, prevDetails.StatementTime,
+		make(map[tree.ChangefeedTableTarget]jobspb.ChangefeedTargetSpecification), nil
+}
+
+func generateAndValidateNewTargetsForTableLevelFeed(
+	ctx context.Context,
+	exprEval exprutil.Evaluator,
+	p sql.PlanHookState,
+	alterCmds tree.AlterChangefeedCmds,
+	opts changefeedbase.StatementOptions,
+	prevDetails jobspb.ChangefeedDetails,
+	prevProgress jobspb.Progress,
+	sinkURI string,
+) (
+	tree.ChangefeedTableTargets,
+	*jobspb.Progress,
+	hlc.Timestamp,
+	map[tree.ChangefeedTableTarget]jobspb.ChangefeedTargetSpecification,
+	error,
+) {
 	type targetKey struct {
 		TableID    descpb.ID
 		FamilyName tree.Name
@@ -870,6 +1021,88 @@ func generateNewProgress(
 	return newProgress, prevStatementTime, nil
 }
 
+// storeAlterChangefeedFrontier 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 storeAlterChangefeedFrontier(
+	ctx context.Context,
+	p sql.PlanHookState,
+	jobID jobspb.JobID,
+	newDetails jobspb.ChangefeedDetails,
+	prevDetails jobspb.ChangefeedDetails,
+) error {
+	// 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.
+	statementTime := hlc.Timestamp{
+		WallTime: p.ExtendedEvalContext().GetStmtTimestamp().UnixNano(),
+	}
+
+	getSpans := func(details jobspb.ChangefeedDetails) ([]roachpb.Span, error) {
+		targets, err := AllTargets(ctx, details, p.ExecCfg(), statementTime)
+		if err != nil {
+			return nil, err
+		}
+		tableDescs, err := fetchTableDescriptors(ctx, p.ExecCfg(), targets, statementTime)
+		if err != nil {
+			return nil, err
+		}
+		spans, err := fetchSpansForTables(ctx, p, tableDescs, details, statementTime)
+		if err != nil {
+			return nil, err
+		}
+		return spans, nil
+	}
+
+	// We compute the difference between the spans we would have been tracking
+	// at the time of the ALTER CHANGEFEED statement and the spans we expect to
+	// be tracking at that time with the new filter.
+	trackedSpansBeforeAlter, err := getSpans(prevDetails)
+	if err != nil {
+		return err
+	}
+	// Note that these spans are computed based on the spans we expect to be
+	// tracking when processing events from the time of the ALTER CHANGEFEED
+	// statement. This may be different than the spans/targets we will be
+	// tracking when the changefeed is resumed (which will be computed based on
+	// the highwater). This will happen if a table that is matched by the filter
+	// is created between the highwater and the ALTER CHANGEFEED statement.
+	trackedSpansAfterAlter, err := getSpans(newDetails)
+	if err != nil {
+		return err
+	}
+
+	var newTrackedSpans roachpb.SpanGroup
+	newTrackedSpans.Add(trackedSpansAfterAlter...)
+	newTrackedSpans.Sub(trackedSpansBeforeAlter...)
+
+	frontier, err := span.MakeFrontierAt(statementTime, newTrackedSpans.Slice()...)
+	if err != nil {
+		return err
+	}
+
+	// If an alter_changefeed frontier already exists, we add those resolved spans
+	// back to the frontier so as not to lose that progress. This could happen if
+	// we execute multiple ALTER CHANGEFEED statements that change the filter
+	// options to track new tables.
+	resolvedSpans, _, err := jobfrontier.GetResolvedSpans(ctx, p.InternalSQLTxn(), jobID, `alter_changefeed`)
+	if err != nil {
+		return err
+	}
+	for _, resolvedSpan := range resolvedSpans {
+		if err := frontier.AddSpansAt(resolvedSpan.Timestamp, resolvedSpan.Span); 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 {
@@ -934,6 +1167,55 @@ func fetchSpansForDescs(p sql.PlanHookState, spanIDs []spanID) (primarySpans []r
 	return primarySpans
 }
 
+// parseFilterOptionFromTargetSpec parses the existing filter option from the
+// target specification so that it can be used to generate the new filter option.
+func parseFilterOptionFromTargetSpec(
+	targetSpec jobspb.ChangefeedTargetSpecification,
+) (tree.ChangefeedFilterOption, error) {
+	var tables tree.TableNames
+	if targetSpec.FilterList == nil {
+		return tree.ChangefeedFilterOption{}, errors.AssertionFailedf("filter list is nil")
+	}
+	for table := range targetSpec.FilterList.Tables {
+		// Parse the fully-qualified table name string back into a TableName object.
+		unresolvedName, err := parser.ParseTableName(table)
+		if err != nil {
+			return tree.ChangefeedFilterOption{}, err
+		}
+		tableName := unresolvedName.ToTableName()
+		tables = append(tables, tableName)
+	}
+	return tree.ChangefeedFilterOption{
+		FilterType: targetSpec.FilterList.FilterType,
+		Tables:     tables,
+	}, nil
+}
+
+// generateNewFilters processes alter changefeed commands and extracts filter changes.
+// It returns a slice of Filter structs containing all filter modifications.
+func generateNewFilters(
+	alterCmds tree.AlterChangefeedCmds,
+) (filters []tree.ChangefeedFilterOption) {
+	for _, cmd := range alterCmds {
+		switch v := cmd.(type) {
+		case *tree.AlterChangefeedSetFilterOption:
+			filters = append(filters, tree.ChangefeedFilterOption{
+				FilterType: v.ChangefeedFilterOption.FilterType,
+				Tables:     v.ChangefeedFilterOption.Tables,
+			})
+		case *tree.AlterChangefeedUnsetFilterOption:
+			// Here we do not return the DefaultChangefeedFilterOption (Exclude
+			// with an empty tables list) so we can treat UNSET INCLUDE TABLES and
+			// UNSET EXCLUDE TABLES differently.
+			filters = append(filters, tree.ChangefeedFilterOption{
+				FilterType: v.ChangefeedFilterOption.FilterType,
+			})
+		}
+	}
+
+	return filters
+}
+
 func getPrevOpts(prevDescription string, opts map[string]string) (map[string]string, error) {
 	prevStmt, err := parser.ParseOne(prevDescription)
 	if err != nil {