changefeedccl: modify changefeed job payload to support db-level changefeeds

Update and populate changefeed job payload to
support db-level changefeeds.

Epic: CRDB-1421
Resolves: #147370

Release note: None

Author: KeithCh

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -475,6 +475,20 @@ func evalCursor(
 	return asOf.Timestamp, nil
 }
 
+func getTargetList(changefeedStmt *annotatedChangefeedStatement) (*tree.BackupTargetList, error) {
+	targetList := tree.BackupTargetList{}
+	if changefeedStmt.Level == tree.ChangefeedLevelTable {
+		for _, t := range changefeedStmt.TableTargets {
+			targetList.Tables.TablePatterns = append(targetList.Tables.TablePatterns, t.TableName)
+		}
+	} else if changefeedStmt.Level == tree.ChangefeedLevelDatabase {
+		targetList.Databases = tree.NameList{tree.Name(changefeedStmt.DatabaseTarget)}
+	} else {
+		return nil, errors.AssertionFailedf("unknown changefeed level: %s", changefeedStmt.Level.String())
+	}
+	return &targetList, nil
+}
+
 func createChangefeedJobRecord(
 	ctx context.Context,
 	p sql.PlanHookState,
@@ -542,18 +556,20 @@ func createChangefeedJobRecord(
 		}
 	}
 
-	tableOnlyTargetList := tree.BackupTargetList{}
-	for _, t := range changefeedStmt.TableTargets {
-		tableOnlyTargetList.Tables.TablePatterns = append(tableOnlyTargetList.Tables.TablePatterns, t.TableName)
+	targetList, err := getTargetList(changefeedStmt)
+	if err != nil {
+		return nil, err
 	}
 
-	// This grabs table descriptors once to get their ids.
-	targetDescs, err := getTableDescriptors(ctx, p, &tableOnlyTargetList, statementTime, initialHighWater)
+	var details jobspb.ChangefeedDetails
+
+	tableNameToDescriptor, targetDatabaseDescs, err := getTargetDescriptors(ctx, p, targetList, statementTime, initialHighWater)
+
 	if err != nil {
 		return nil, err
 	}
 
-	for _, t := range targetDescs {
+	for _, t := range tableNameToDescriptor {
 		if tbl, ok := t.(catalog.TableDescriptor); ok && tbl.ExternalRowData() != nil {
 			if tbl.ExternalRowData().TenantID.IsSet() {
 				return nil, errors.UnimplementedError(errors.IssueLink{}, "changefeeds on a replication target are not supported")
@@ -562,7 +578,8 @@ func createChangefeedJobRecord(
 		}
 	}
 
-	targets, tables, err := getTargetsAndTables(ctx, p, targetDescs, changefeedStmt.TableTargets,
+	// This grabs table descriptors once to get their ids.
+	targets, tables, err := getTargetsAndTables(ctx, p, tableNameToDescriptor, changefeedStmt.TableTargets,
 		changefeedStmt.originalSpecs, opts.ShouldUseFullStatementTimeName())
 
 	if err != nil {
@@ -572,19 +589,44 @@ func createChangefeedJobRecord(
 	sd := p.SessionData().Clone()
 	// Add non-local session data state (localization, etc).
 	sessiondata.MarshalNonLocal(p.SessionData(), &sd.SessionData)
-	details := jobspb.ChangefeedDetails{
-		Tables:               tables,
-		SinkURI:              sinkURI,
-		StatementTime:        statementTime,
-		EndTime:              endTime,
-		TargetSpecifications: targets,
-		SessionData:          &sd.SessionData,
+	if changefeedStmt.Level == tree.ChangefeedLevelTable {
+		for _, t := range tableNameToDescriptor {
+			if tbl, ok := t.(catalog.TableDescriptor); ok && tbl.ExternalRowData() != nil {
+				if tbl.ExternalRowData().TenantID.IsSet() {
+					return nil, errors.UnimplementedError(errors.IssueLink{}, "changefeeds on a replication target are not supported")
+				}
+				return nil, errors.UnimplementedError(errors.IssueLink{}, "changefeeds on external tables are not supported")
+			}
+		}
+
+		details = jobspb.ChangefeedDetails{
+			Tables:               tables,
+			SinkURI:              sinkURI,
+			StatementTime:        statementTime,
+			EndTime:              endTime,
+			TargetSpecifications: targets,
+			SessionData:          &sd.SessionData,
+		}
+	} else if changefeedStmt.Level == tree.ChangefeedLevelDatabase {
+		if len(targetDatabaseDescs) == 0 || len(targetDatabaseDescs) > 1 {
+			return nil, errors.Errorf("only one database target is supported")
+		}
+		targets = getDatabaseTargets(targetDatabaseDescs)
+		details = jobspb.ChangefeedDetails{
+			TargetSpecifications: targets,
+			SinkURI:              sinkURI,
+			StatementTime:        statementTime,
+			EndTime:              endTime,
+			SessionData:          &sd.SessionData,
+		}
+	} else {
+		return nil, errors.AssertionFailedf("unknown changefeed level: %s", changefeedStmt.Level)
 	}
 
 	specs := AllTargets(details)
 	hasSelectPrivOnAllTables := true
 	hasChangefeedPrivOnAllTables := true
-	for _, desc := range targetDescs {
+	for _, desc := range tableNameToDescriptor {
 		if table, isTable := desc.(catalog.TableDescriptor); isTable {
 			if err := changefeedvalidators.ValidateTable(specs, table, tolerances); err != nil {
 				return nil, err
@@ -610,7 +652,7 @@ func createChangefeedJobRecord(
 	if changefeedStmt.Select != nil {
 		// Serialize changefeed expression.
 		normalized, withDiff, err := validateAndNormalizeChangefeedExpression(
-			ctx, p, opts, changefeedStmt.Select, targetDescs, targets, statementTime,
+			ctx, p, opts, changefeedStmt.Select, tableNameToDescriptor, targets, statementTime,
 		)
 		if err != nil {
 			return nil, err
@@ -866,7 +908,7 @@ Few hours to a few days range are appropriate values for this option.`
 		Description: description,
 		Username:    p.User(),
 		DescriptorIDs: func() (sqlDescIDs []descpb.ID) {
-			for _, desc := range targetDescs {
+			for _, desc := range tableNameToDescriptor {
 				sqlDescIDs = append(sqlDescIDs, desc.GetID())
 			}
 			return sqlDescIDs
@@ -899,31 +941,31 @@ func validateSettings(ctx context.Context, needsRangeFeed bool, execCfg *sql.Exe
 	return nil
 }
 
-func getTableDescriptors(
+func getTargetDescriptors(
 	ctx context.Context,
 	p sql.PlanHookState,
 	targets *tree.BackupTargetList,
 	statementTime hlc.Timestamp,
 	initialHighWater hlc.Timestamp,
-) (map[tree.TablePattern]catalog.Descriptor, error) {
-	// For now, disallow targeting a database or wildcard table selection.
-	// Getting it right as tables enter and leave the set over time is
-	// tricky.
-	if len(targets.Databases) > 0 {
-		return nil, errors.Errorf(`CHANGEFEED cannot target %s`,
-			tree.AsString(targets))
+) (
+	tableNameToDescriptor map[tree.TablePattern]catalog.Descriptor,
+	databaseDescs []catalog.DatabaseDescriptor,
+	err error,
+) {
+	if len(targets.Databases) > 0 && len(targets.Tables.TablePatterns) > 0 {
+		return nil, nil, errors.Errorf(`CHANGEFEED cannot target both databases and tables`)
 	}
 	for _, t := range targets.Tables.TablePatterns {
 		p, err := t.NormalizeTablePattern()
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		if _, ok := p.(*tree.TableName); !ok {
-			return nil, errors.Errorf(`CHANGEFEED cannot target %s`, tree.AsString(t))
+			return nil, nil, errors.Errorf(`CHANGEFEED cannot target %s`, tree.AsString(t))
 		}
 	}
 
-	_, _, _, targetDescs, err := backupresolver.ResolveTargetsToDescriptors(ctx, p, statementTime, targets)
+	_, _, targetDatabaseDescs, targetTableDescs, err := backupresolver.ResolveTargetsToDescriptors(ctx, p, statementTime, targets)
 	if err != nil {
 		var m *backupresolver.MissingTableErr
 		if errors.As(err, &m) {
@@ -937,7 +979,7 @@ func getTableDescriptors(
 				"do the targets exist at the specified cursor time %s?", initialHighWater)
 		}
 	}
-	return targetDescs, err
+	return targetTableDescs, targetDatabaseDescs, err
 }
 
 func getTargetsAndTables(
@@ -1013,6 +1055,21 @@ func getTargetsAndTables(
 	return targets, tables, nil
 }
 
+func getDatabaseTargets(
+	targetDatabaseDescs []catalog.DatabaseDescriptor,
+) []jobspb.ChangefeedTargetSpecification {
+	targets := make([]jobspb.ChangefeedTargetSpecification, len(targetDatabaseDescs))
+
+	for i, desc := range targetDatabaseDescs {
+		targets[i] = jobspb.ChangefeedTargetSpecification{
+			DescID:            desc.GetID(),
+			Type:              jobspb.ChangefeedTargetSpecification_DATABASE,
+			StatementTimeName: desc.GetName(),
+		}
+	}
+	return targets
+}
+
 func validateSink(
 	ctx context.Context,
 	p sql.PlanHookState,

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11820,3 +11820,17 @@ func TestCloudstorageParallelCompression(t *testing.T) {
 		}
 	})
 }
+
+func TestDatabaseLevelChangefeed(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 foo`)
+		sqlDB.Exec(t, `CREATE TABLE foo.bar(id int primary key, s string)`)
+		sqlDB.Exec(t, `INSERT INTO foo.bar(id, s) VALUES (0, 'hello'), (1, null)`)
+		expectErrCreatingFeed(t, f, `CREATE CHANGEFEED for DATABASE foo`, "database-level changefeed is not implemented")
+	}
+	cdcTest(t, testFn)
+}

pkg/jobs/jobspb/jobs.proto

@@ -1098,8 +1098,8 @@ message ChangefeedTargetSpecification {
     // Column family family_name of table table_id.
     COLUMN_FAMILY = 2;
 
-    // Add TargetTypes for database, secondary index, etc. when implemented
-
+    DATABASE = 3;
+     // Add TargetTypes for schema, secondary index, etc. when implemented
   }
 
   TargetType type = 1;

pkg/sql/sem/tree/changefeed.go

@@ -21,6 +21,10 @@ const (
 	ChangefeedLevelDatabase
 )
 
+func (l ChangefeedLevel) String() string {
+	return []string{"TABLE", "DATABASE"}[l]
+}
+
 // CreateChangefeed represents a CREATE CHANGEFEED statement.
 type CreateChangefeed struct {
 	TableTargets   ChangefeedTableTargets