Merge #149984

149984: changefeedccl: create db-level changefeed job r=andyyang890,asg0451 a=KeithCh

Support the creation of database-level changefeed
jobs e.g.
CREATE CHANGEFEED FOR DATABASE d;
    
We do this by modifying AllTargets to expand
database-level changefeed target spec to fetch
all tables in the database.

Resolves: #147371
Epic: CRDB-1421

Release note: None

Co-authored-by: Keith Chow <keith.chow@cockroachlabs.com>

Author: craig[bot]

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -202,7 +202,7 @@ func alterChangefeedPlanHook(
 			return err
 		}
 
-		jobRecord, err := createChangefeedJobRecord(
+		jobRecord, targets, err := createChangefeedJobRecord(
 			ctx,
 			p,
 			annotatedStmt,
@@ -251,7 +251,7 @@ func alterChangefeedPlanHook(
 
 		telemetry.Count(telemetryPath)
 
-		logAlterChangefeedTelemetry(ctx, j, jobPayload.Description)
+		logAlterChangefeedTelemetry(ctx, j, jobPayload.Description, targets.Size)
 
 		select {
 		case <-ctx.Done():
@@ -447,7 +447,10 @@ func generateAndValidateNewTargets(
 		return nil, nil, hlc.Timestamp{}, nil, err
 	}
 
-	prevTargets := AllTargets(prevDetails)
+	prevTargets, err := AllTargets(ctx, prevDetails, p.ExecCfg())
+	if err != nil {
+		return nil, nil, hlc.Timestamp{}, nil, err
+	}
 	noLongerExist := make(map[string]descpb.ID)
 	if err := prevTargets.EachTarget(func(targetSpec changefeedbase.Target) error {
 		k := targetKey{TableID: targetSpec.DescID, FamilyName: tree.Name(targetSpec.FamilyName)}

pkg/ccl/changefeedccl/changefeed.go

@@ -12,10 +12,15 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
+	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/protoreflect"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/errors"
 	"github.com/gogo/protobuf/jsonpb"
 )
 
@@ -37,33 +42,59 @@ type ChangefeedConfig struct {
 
 // makeChangefeedConfigFromJobDetails creates a ChangefeedConfig struct from any
 // version of the ChangefeedDetails protobuf.
-func makeChangefeedConfigFromJobDetails(d jobspb.ChangefeedDetails) ChangefeedConfig {
+func makeChangefeedConfigFromJobDetails(
+	ctx context.Context, d jobspb.ChangefeedDetails, execCfg *sql.ExecutorConfig,
+) (ChangefeedConfig, error) {
+	targets, err := AllTargets(ctx, d, execCfg)
+	if err != nil {
+		return ChangefeedConfig{}, err
+	}
 	return ChangefeedConfig{
 		SinkURI:  d.SinkURI,
 		Opts:     changefeedbase.MakeStatementOptions(d.Opts),
 		ScanTime: d.StatementTime,
 		EndTime:  d.EndTime,
-		Targets:  AllTargets(d),
-	}
+		Targets:  targets,
+	}, nil
 }
 
 // AllTargets gets all the targets listed in a ChangefeedDetails,
 // from the statement time name map in old protos
 // or the TargetSpecifications in new ones.
-func AllTargets(cd jobspb.ChangefeedDetails) (targets changefeedbase.Targets) {
+func AllTargets(
+	ctx context.Context, cd jobspb.ChangefeedDetails, execCfg *sql.ExecutorConfig,
+) (changefeedbase.Targets, error) {
+	targets := changefeedbase.Targets{}
+	var err error
+
 	// TODO: Use a version gate for this once we have CDC version gates
 	if len(cd.TargetSpecifications) > 0 {
 		for _, ts := range cd.TargetSpecifications {
 			if ts.DescID > 0 {
-				if ts.StatementTimeName == "" {
-					ts.StatementTimeName = cd.Tables[ts.DescID].StatementTimeName
+				switch ts.Type {
+				case jobspb.ChangefeedTargetSpecification_DATABASE:
+					if len(cd.TargetSpecifications) > 1 {
+						return changefeedbase.Targets{}, errors.AssertionFailedf("database-level changefeed is not supported with multiple targets")
+					}
+					targets, err = getTargetsFromDatabaseSpec(ctx, ts, execCfg)
+					if err != nil {
+						return changefeedbase.Targets{}, err
+					}
+				case jobspb.ChangefeedTargetSpecification_PRIMARY_FAMILY_ONLY,
+					jobspb.ChangefeedTargetSpecification_COLUMN_FAMILY,
+					jobspb.ChangefeedTargetSpecification_EACH_FAMILY:
+					if ts.StatementTimeName == "" {
+						ts.StatementTimeName = cd.Tables[ts.DescID].StatementTimeName
+					}
+					targets.Add(changefeedbase.Target{
+						Type:              ts.Type,
+						DescID:            ts.DescID,
+						FamilyName:        ts.FamilyName,
+						StatementTimeName: changefeedbase.StatementTimeName(ts.StatementTimeName),
+					})
+				default:
+					return changefeedbase.Targets{}, errors.AssertionFailedf("unsupported target type: %s", ts.Type)
 				}
-				targets.Add(changefeedbase.Target{
-					Type:              ts.Type,
-					DescID:            ts.DescID,
-					FamilyName:        ts.FamilyName,
-					StatementTimeName: changefeedbase.StatementTimeName(ts.StatementTimeName),
-				})
 			}
 		}
 	} else {
@@ -75,7 +106,46 @@ func AllTargets(cd jobspb.ChangefeedDetails) (targets changefeedbase.Targets) {
 			})
 		}
 	}
-	return
+	return targets, err
+}
+
+func getTargetsFromDatabaseSpec(
+	ctx context.Context, ts jobspb.ChangefeedTargetSpecification, execCfg *sql.ExecutorConfig,
+) (targets changefeedbase.Targets, err error) {
+	err = sql.DescsTxn(ctx, execCfg, func(ctx context.Context, txn isql.Txn, descs *descs.Collection) error {
+		databaseDescriptor, err := descs.ByIDWithLeased(txn.KV()).Get().Database(ctx, ts.DescID)
+		if err != nil {
+			return err
+		}
+		tables, err := descs.GetAllTablesInDatabase(ctx, txn.KV(), databaseDescriptor)
+		if err != nil {
+			return err
+		}
+		for _, desc := range tables.OrderedDescriptors() {
+			tableDesc, ok := desc.(catalog.TableDescriptor)
+			if !ok {
+				return errors.AssertionFailedf("expected table descriptor, got %T", desc)
+			}
+			// Skip virtual tables
+			if !tableDesc.IsPhysicalTable() {
+				continue
+			}
+			var tableType jobspb.ChangefeedTargetSpecification_TargetType
+			if len(tableDesc.GetFamilies()) == 1 {
+				tableType = jobspb.ChangefeedTargetSpecification_PRIMARY_FAMILY_ONLY
+			} else {
+				tableType = jobspb.ChangefeedTargetSpecification_EACH_FAMILY
+			}
+
+			targets.Add(changefeedbase.Target{
+				Type:              tableType,
+				DescID:            desc.GetID(),
+				StatementTimeName: changefeedbase.StatementTimeName(desc.GetName()),
+			})
+		}
+		return nil
+	})
+	return targets, err
 }
 
 const (

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -82,6 +82,7 @@ func distChangefeedFlow(
 	localState *cachedState,
 	resultsCh chan<- tree.Datums,
 	onTracingEvent func(ctx context.Context, meta *execinfrapb.TracingAggregatorEvents),
+	targets changefeedbase.Targets,
 ) error {
 	opts := changefeedbase.MakeStatementOptions(details.Opts)
 	progress := localState.progress
@@ -135,7 +136,7 @@ func distChangefeedFlow(
 		}
 	}
 	return startDistChangefeed(
-		ctx, execCtx, jobID, schemaTS, details, description, initialHighWater, localState, resultsCh, onTracingEvent)
+		ctx, execCtx, jobID, schemaTS, details, description, initialHighWater, localState, resultsCh, onTracingEvent, targets)
 }
 
 func fetchTableDescriptors(
@@ -235,9 +236,10 @@ func startDistChangefeed(
 	localState *cachedState,
 	resultsCh chan<- tree.Datums,
 	onTracingEvent func(ctx context.Context, meta *execinfrapb.TracingAggregatorEvents),
+	targets changefeedbase.Targets,
 ) error {
 	execCfg := execCtx.ExecCfg()
-	tableDescs, err := fetchTableDescriptors(ctx, execCfg, AllTargets(details), schemaTS)
+	tableDescs, err := fetchTableDescriptors(ctx, execCfg, targets, schemaTS)
 	if err != nil {
 		return err
 	}

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -121,6 +121,8 @@ type changeAggregator struct {
 	sliMetricsID           int64
 	closeTelemetryRecorder func()
 	knobs                  TestingKnobs
+
+	targets changefeedbase.Targets
 }
 
 type timestampLowerBoundOracle interface {
@@ -286,7 +288,7 @@ func (ca *changeAggregator) MustBeStreaming() bool {
 // wrapMetricsRecorderWithTelemetry wraps the supplied metricsRecorder
 // so it periodically emits metrics to telemetry.
 func (ca *changeAggregator) wrapMetricsRecorderWithTelemetry(
-	ctx context.Context, recorder metricsRecorder,
+	ctx context.Context, recorder metricsRecorder, targets changefeedbase.Targets,
 ) (metricsRecorder, error) {
 	details := ca.spec.Feed
 	jobID := ca.spec.JobID
@@ -307,7 +309,7 @@ func (ca *changeAggregator) wrapMetricsRecorderWithTelemetry(
 		description = job.Payload().Description
 	}
 
-	recorderWithTelemetry, err := wrapMetricsRecorderWithTelemetry(ctx, details, description, jobID, ca.FlowCtx.Cfg.Settings, recorder, ca.knobs)
+	recorderWithTelemetry, err := wrapMetricsRecorderWithTelemetry(ctx, details, description, jobID, ca.FlowCtx.Cfg.Settings, recorder, ca.knobs, targets)
 	if err != nil {
 		return ca.sliMetrics, err
 	}
@@ -350,8 +352,15 @@ func (ca *changeAggregator) Start(ctx context.Context) {
 		return
 	}
 
-	feed := makeChangefeedConfigFromJobDetails(ca.spec.Feed)
-
+	feed, err := makeChangefeedConfigFromJobDetails(ctx, ca.spec.Feed, ca.FlowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig))
+	if err != nil {
+		if log.V(2) {
+			log.Infof(ca.Ctx(), "change aggregator moving to draining due to error making changefeed config: %v", err)
+		}
+		ca.MoveToDraining(err)
+		ca.cancel()
+		return
+	}
 	opts := feed.Opts
 
 	timestampOracle := &changeAggregatorLowerBoundOracle{
@@ -378,20 +387,29 @@ func (ca *changeAggregator) Start(ctx context.Context) {
 		return
 	}
 	ca.sliMetricsID = ca.sliMetrics.claimId()
+	ca.targets, err = AllTargets(ctx, ca.spec.Feed, ca.FlowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig))
+	if err != nil {
+		if log.V(2) {
+			log.Infof(ca.Ctx(), "change aggregator moving to draining due to error getting targets: %v", err)
+		}
+		ca.MoveToDraining(err)
+		ca.cancel()
+		return
+	}
 
 	recorder := metricsRecorder(ca.sliMetrics)
-	recorder, err = ca.wrapMetricsRecorderWithTelemetry(ctx, recorder)
+	recorder, err = ca.wrapMetricsRecorderWithTelemetry(ctx, recorder, ca.targets)
+
 	if err != nil {
 		if log.V(2) {
 			log.Infof(ca.Ctx(), "change aggregator moving to draining due to error wrapping metrics controller: %v", err)
 		}
 		ca.MoveToDraining(err)
 		ca.cancel()
-		return
 	}
 
 	ca.sink, err = getEventSink(ctx, ca.FlowCtx.Cfg, ca.spec.Feed, timestampOracle,
-		ca.spec.User(), ca.spec.JobID, recorder)
+		ca.spec.User(), ca.spec.JobID, recorder, ca.targets)
 	if err != nil {
 		err = changefeedbase.MarkRetryableError(err)
 		if log.V(2) {
@@ -535,7 +553,7 @@ func (ca *changeAggregator) makeKVFeedCfg(
 	if schemaChange.Policy == changefeedbase.OptSchemaChangePolicyIgnore || initialScanOnly {
 		sf = schemafeed.DoNothingSchemaFeed
 	} else {
-		sf = schemafeed.New(ctx, cfg, schemaChange.EventClass, AllTargets(ca.spec.Feed),
+		sf = schemafeed.New(ctx, cfg, schemaChange.EventClass, ca.targets,
 			initialHighWater, &ca.metrics.SchemaFeedMetrics, config.Opts.GetCanHandle())
 	}
 
@@ -552,7 +570,7 @@ func (ca *changeAggregator) makeKVFeedCfg(
 		Clock:                cfg.DB.KV().Clock(),
 		Spans:                spans,
 		SpanLevelCheckpoint:  ca.spec.SpanLevelCheckpoint,
-		Targets:              AllTargets(ca.spec.Feed),
+		Targets:              ca.targets,
 		Metrics:              &ca.metrics.KVFeedMetrics,
 		MM:                   memMon,
 		InitialHighWater:     initialHighWater,
@@ -1073,6 +1091,8 @@ type changeFrontier struct {
 
 	usageWg       sync.WaitGroup
 	usageWgCancel context.CancelFunc
+
+	targets changefeedbase.Targets
 }
 
 const (
@@ -1295,8 +1315,13 @@ func newChangeFrontierProcessor(
 	if err != nil {
 		return nil, err
 	}
+	targets, err := AllTargets(ctx, spec.Feed, flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig))
+	if err != nil {
+		return nil, err
+	}
+	cf.targets = targets
 	if cf.encoder, err = getEncoder(
-		ctx, encodingOpts, AllTargets(spec.Feed), spec.Feed.Select != "",
+		ctx, encodingOpts, targets, spec.Feed.Select != "",
 		makeExternalConnectionProvider(ctx, flowCtx.Cfg.DB), sliMetrics,
 		sourceProvider,
 	); err != nil {
@@ -1349,9 +1374,8 @@ func (cf *changeFrontier) Start(ctx context.Context) {
 		return
 	}
 	cf.sliMetrics = sli
-
 	cf.sink, err = getResolvedTimestampSink(ctx, cf.FlowCtx.Cfg, cf.spec.Feed, nilOracle,
-		cf.spec.User(), cf.spec.JobID, sli)
+		cf.spec.User(), cf.spec.JobID, sli, cf.targets)
 	if err != nil {
 		err = changefeedbase.MarkRetryableError(err)
 		if log.V(2) {
@@ -1924,10 +1948,9 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 	if highWater.Less(cf.highWaterAtStart) {
 		highWater = cf.highWaterAtStart
 	}
-
 	if progress.ProtectedTimestampRecord == uuid.Nil {
 		ptr := createProtectedTimestampRecord(
-			ctx, cf.FlowCtx.Codec(), cf.spec.JobID, AllTargets(cf.spec.Feed), highWater,
+			ctx, cf.FlowCtx.Codec(), cf.spec.JobID, cf.targets, highWater,
 		)
 		progress.ProtectedTimestampRecord = ptr.ID.GetUUID()
 		return true, pts.Protect(ctx, ptr)
@@ -1954,7 +1977,7 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 	// If we've identified more tables that need to be protected since this
 	// changefeed was created, it will be missing here. If so, we "migrate" it
 	// to include all the appropriate targets.
-	if targets := AllTargets(cf.spec.Feed); !makeTargetToProtect(targets).Equal(rec.Target) {
+	if !makeTargetToProtect(cf.targets).Equal(rec.Target) {
 		if preservePTSTargets := cf.knobs.PreservePTSTargets != nil && cf.knobs.PreservePTSTargets(); preservePTSTargets {
 			return false, nil
 		}
@@ -1988,7 +2011,7 @@ func (cf *changeFrontier) remakePTSRecord(
 ) error {
 	prevRecordId := progress.ProtectedTimestampRecord
 	ptr := createProtectedTimestampRecord(
-		ctx, cf.FlowCtx.Codec(), cf.spec.JobID, AllTargets(cf.spec.Feed), resolved,
+		ctx, cf.FlowCtx.Codec(), cf.spec.JobID, cf.targets, resolved,
 	)
 	if err := pts.Protect(ctx, ptr); err != nil {
 		return err