changefeedccl: change db-level feed default to no initial scan

This change the default initial scan behavior for db-level
feeds to not do an initial scan if no option is specified.
When an initial scan is specified, that is respected.
We do not allow db-level changefeeds with initial_scan='only'.

This includes a rewrite of existing db-level feeds tests which
rely on default initial scans in their testing.

Epic: CRDB-1421
Informs: #156484

Release note: None

Author: aerfrei

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -230,9 +230,18 @@ func changefeedPlanHook(
 		return nil, nil, false, err
 	}
 
-	// Treat all tables inside a database as if "split_column_families" is set.
 	if changefeedStmt.Level == tree.ChangefeedLevelDatabase {
+		// Treat all tables inside a database as if "split_column_families" is set.
 		rawOpts[changefeedbase.OptSplitColumnFamilies] = `yes`
+
+		// The default behavior for a database-level changefeed is
+		// NOT to perform an initial scan, unlike table-level changefeeds.
+		_, initialScanSet := rawOpts[changefeedbase.OptInitialScan]
+		_, initialScanOnlySet := rawOpts[changefeedbase.OptInitialScanOnly]
+		_, noInitialScanSet := rawOpts[changefeedbase.OptNoInitialScan]
+		if !initialScanOnlySet && !noInitialScanSet && !initialScanSet {
+			rawOpts[changefeedbase.OptInitialScan] = `no`
+		}
 	}
 	opts := changefeedbase.MakeStatementOptions(rawOpts)
 
@@ -1462,6 +1471,16 @@ func validateDetailsAndOptions(
 	if err := opts.ValidateForCreateChangefeed(details.Select != ""); err != nil {
 		return err
 	}
+	if isDBLevelChangefeed(details) {
+		scanType, err := opts.GetInitialScanType()
+		if err != nil {
+			return err
+		}
+		if scanType == changefeedbase.OnlyInitialScan {
+			return errors.Errorf(
+				`cannot specify %s on a database level changefeed`, changefeedbase.OptInitialScanOnly)
+		}
+	}
 	if opts.HasEndTime() {
 		scanType, err := opts.GetInitialScanType()
 		if err != nil {
@@ -2123,6 +2142,14 @@ func getQualifiedTableNameObj(
 	return tbName, nil
 }
 
+func isDBLevelChangefeed(details jobspb.ChangefeedDetails) bool {
+	targetSpecs := details.TargetSpecifications
+	if len(targetSpecs) == 0 {
+		return false
+	}
+	return targetSpecs[0].Type == jobspb.ChangefeedTargetSpecification_DATABASE
+}
+
 // getChangefeedTargetName gets a table name with or without the dots
 func getChangefeedTargetName(
 	ctx context.Context,