Merge #154061 #154143

154061: sql/schemachanger: simplify validation for CTAS statements r=fqazi a=fqazi

Previously, the validation we used for CTAS statements would re-run the same query using the same AOST. This worked fine for deterministic queries, where the count between the re-executed query and backfill would match. However, if the query was non-deterministic (e.g., due to non-deterministic functions/routines), this check would fail. To address this, this patch removes the AOST query validation and only uses the row count from the bulk writer

Fixes: #153363

Release note: None

154143: changefeedccl: deflake tests that expect frequent span-level checkpoints r=aerfrei,asg0451 a=andyyang890

**changefeedccl: delete leftover comment**

This patch deletes a leftover comment that should've been deleted in
the commit that had aggregators periodically flush their progress.

Release note: None

---

**changefeedccl: deflake tests that expect frequent span-level checkpoints**

In a recent commit, the interval at which change aggregators would
flush their frontiers to the change frontier was decoupled from
the span-level checkpoint interval (which is configurable via a
cluster setting). This caused some tests that only set the cluster
setting (and not `min_checkpoint_frequency`) to a low value to
sometimes flake. This patch updates all such tests to also configure
the `min_checkpoint_frequency` option.

[master]
Fixes #151279

[release-25.4]
Informs #154066

Release note: None

---

**changefeedccl: deflake initial backfill checkpoint tests**

This patch deflakes the initial backfill checkpoint tests, which filter
out resolved spans to force gaps during a backfill. The tests however
did not ensure that there would be at least one span that was not
filtered out, which would result in no span-level checkpoint being
created, leading the tests to time out.

Release note: None

Co-authored-by: Faizan Qazi <[email protected]>
Co-authored-by: Andy Yang <[email protected]>

Author: 3 people

pkg/ccl/changefeedccl/alter_changefeed_test.go

@@ -1520,19 +1520,27 @@ func TestAlterChangefeedAddTargetsDuringBackfill(t *testing.T) {
 
 		// Emit resolved events for the majority of spans. Be extra paranoid and ensure that
 		// we have at least 1 span for which we don't emit resolvedFoo timestamp (to force checkpointing).
-		haveGaps := false
-		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (bool, error) {
+		// We however also need to ensure there's at least one span that isn't filtered out.
+		var allowedOne, haveGaps bool
+		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (filter bool, _ error) {
 			rndMu.Lock()
 			defer rndMu.Unlock()
+			defer func() {
+				t.Logf("resolved span: %s@%s, filter: %t", r.Span, r.Timestamp, filter)
+			}()
 
 			if r.Span.Equal(fooTableSpan) {
 				return true, nil
 			}
-			if haveGaps {
-				return rndMu.rnd.Intn(10) > 7, nil
+			if !allowedOne {
+				allowedOne = true
+				return false, nil
 			}
-			haveGaps = true
-			return true, nil
+			if !haveGaps {
+				haveGaps = true
+				return true, nil
+			}
+			return rndMu.rnd.Intn(10) > 7, nil
 		}
 
 		// Checkpoint progress frequently, and set the checkpoint size limit.
@@ -1542,7 +1550,8 @@ func TestAlterChangefeedAddTargetsDuringBackfill(t *testing.T) {
 			context.Background(), &s.Server.ClusterSettings().SV, maxCheckpointSize)
 
 		registry := s.Server.JobRegistry().(*jobs.Registry)
-		testFeed := feed(t, f, `CREATE CHANGEFEED FOR foo WITH resolved = '100ms'`)
+		testFeed := feed(t, f, `CREATE CHANGEFEED FOR foo
+WITH resolved = '100ms', min_checkpoint_frequency='1ns'`)
 
 		g := ctxgroup.WithContext(context.Background())
 		g.Go(func() error {
@@ -1581,8 +1590,10 @@ func TestAlterChangefeedAddTargetsDuringBackfill(t *testing.T) {
 
 		// Collect spans we attempt to resolve after when we resume.
 		var resolvedFoo []roachpb.Span
-		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (bool, error) {
-			t.Logf("resolved span: %#v", r)
+		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (filter bool, _ error) {
+			defer func() {
+				t.Logf("resolved span: %s@%s, filter: %t", r.Span, r.Timestamp, filter)
+			}()
 			if !r.Span.Equal(fooTableSpan) {
 				resolvedFoo = append(resolvedFoo, r.Span)
 			}

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -932,12 +932,6 @@ func (ca *changeAggregator) noteResolvedSpan(resolved jobspb.ResolvedSpan) error
 
 	forceFlush := resolved.BoundaryType != jobspb.ResolvedSpan_NONE
 
-	// NB: if we miss flush window, and the flush frequency is fairly high (minutes),
-	// it might be a while before frontier advances again (particularly if
-	// the number of ranges and closed timestamp settings are high).
-	// TODO(yevgeniy): Consider doing something similar to how job checkpointing
-	//  works in the frontier where if we missed the window to checkpoint, we will attempt
-	//  the checkpoint at the next opportune moment.
 	checkpointFrontier := (advanced && forceFlush) || ca.frontierFlushLimiter.canSave(ctx)
 
 	if checkpointFrontier {

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2861,7 +2861,8 @@ func TestChangefeedLaggingSpanCheckpointing(t *testing.T) {
 
 	var jobID jobspb.JobID
 	sqlDB.QueryRow(t,
-		`CREATE CHANGEFEED FOR foo INTO 'null://' WITH resolved='50ms', no_initial_scan, cursor=$1`, tsStr,
+		`CREATE CHANGEFEED FOR foo INTO 'null://'
+WITH resolved='50ms', min_checkpoint_frequency='50ms', no_initial_scan, cursor=$1`, tsStr,
 	).Scan(&jobID)
 
 	// Helper to read job progress
@@ -3003,7 +3004,8 @@ func TestChangefeedSchemaChangeBackfillCheckpoint(t *testing.T) {
 		}
 
 		// Setup changefeed job details, avoid relying on initial scan functionality
-		baseFeed := feed(t, f, `CREATE CHANGEFEED FOR foo WITH resolved='100ms', min_checkpoint_frequency='100ms', no_initial_scan`)
+		baseFeed := feed(t, f, `CREATE CHANGEFEED FOR foo
+WITH resolved='100ms', min_checkpoint_frequency='1ns', no_initial_scan`)
 		jobFeed := baseFeed.(cdctest.EnterpriseTestFeed)
 		jobRegistry := s.Server.JobRegistry().(*jobs.Registry)
 
@@ -9174,22 +9176,21 @@ func TestChangefeedBackfillCheckpoint(t *testing.T) {
 
 		// Emit resolved events for majority of spans.  Be extra paranoid and ensure that
 		// we have at least 1 span for which we don't emit resolved timestamp (to force checkpointing).
-		haveGaps := false
+		// We however also need to ensure there's at least one span that isn't filtered out.
+		var allowedOne, haveGaps bool
 		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (bool, error) {
 			if r.Span.Equal(tableSpan) {
-				// Do not emit resolved events for the entire table span.
-				// We "simulate" large table by splitting single table span into many parts, so
-				// we want to resolve those sub-spans instead of the entire table span.
-				// However, we have to emit something -- otherwise the entire changefeed
-				// machine would not work.
-				r.Span.EndKey = tableSpan.Key.Next()
+				return true, nil
+			}
+			if !allowedOne {
+				allowedOne = true
 				return false, nil
 			}
-			if haveGaps {
-				return rnd.Intn(10) > 7, nil
+			if !haveGaps {
+				haveGaps = true
+				return true, nil
 			}
-			haveGaps = true
-			return true, nil
+			return rnd.Intn(10) > 7, nil
 		}
 
 		// Checkpoint progress frequently, and set the checkpoint size limit.
@@ -9199,7 +9200,8 @@ func TestChangefeedBackfillCheckpoint(t *testing.T) {
 			context.Background(), &s.Server.ClusterSettings().SV, maxCheckpointSize)
 
 		registry := s.Server.JobRegistry().(*jobs.Registry)
-		foo := feed(t, f, `CREATE CHANGEFEED FOR foo WITH resolved='100ms'`)
+		foo := feed(t, f, `CREATE CHANGEFEED FOR foo
+WITH resolved='100ms', min_checkpoint_frequency='1ns'`)
 		// Some test feeds (kafka) are not buffered, so we have to consume messages.
 		var shouldDrain int32 = 1
 		g := ctxgroup.WithContext(context.Background())
@@ -12059,7 +12061,8 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 		require.Equal(t, int64(0), managePTSCount)
 		require.Equal(t, int64(0), managePTSErrorCount)
 
-		createStmt := `CREATE CHANGEFEED FOR foo WITH resolved='10ms', no_initial_scan`
+		createStmt := `CREATE CHANGEFEED FOR foo
+WITH resolved='10ms', min_checkpoint_frequency='10ms', no_initial_scan`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)
 
@@ -12172,7 +12175,8 @@ func TestChangefeedProtectedTimestampUpdateError(t *testing.T) {
 			return errors.New("test error")
 		}
 
-		createStmt := `CREATE CHANGEFEED FOR foo WITH resolved='10ms', no_initial_scan`
+		createStmt := `CREATE CHANGEFEED FOR foo
+WITH resolved='10ms', min_checkpoint_frequency='10ms', no_initial_scan`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)
 

pkg/ccl/changefeedccl/protected_timestamps_test.go

@@ -970,7 +970,8 @@ func TestChangefeedProtectedTimestampUpdateForMultipleTables(t *testing.T) {
 		require.Equal(t, int64(0), managePTSCount)
 		require.Equal(t, int64(0), managePTSErrorCount)
 
-		createStmt := `CREATE CHANGEFEED FOR foo, bar WITH resolved='10ms', initial_scan='no'`
+		createStmt := `CREATE CHANGEFEED FOR foo, bar
+WITH resolved='10ms', min_checkpoint_frequency='100ms', initial_scan='no'`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)
 
@@ -1114,7 +1115,8 @@ func TestChangefeedPerTableProtectedTimestampProgression(t *testing.T) {
 			}
 		}
 
-		createStmt := `CREATE CHANGEFEED FOR table1, table2, table3 WITH resolved='100ms'`
+		createStmt := `CREATE CHANGEFEED FOR table1, table2, table3
+WITH resolved='100ms', min_checkpoint_frequency='100ms'`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)
 

pkg/sql/schema_changer.go

@@ -312,58 +312,11 @@ const schemaChangerBackfillTxnDebugName = "schemaChangerBackfill"
 // validateBackfillQueryIntoTable validates that source query matches the contents of
 // a backfilled table, when executing the query at queryTS.
 func (sc *SchemaChanger) validateBackfillQueryIntoTable(
-	ctx context.Context,
-	table catalog.TableDescriptor,
-	entryCountWrittenToPrimaryIdx int64,
-	queryTS hlc.Timestamp,
-	query string,
-	skipAOSTValidation bool,
+	ctx context.Context, table catalog.TableDescriptor, entryCountWrittenToPrimaryIdx int64,
 ) error {
-	var aostEntryCount int64
 	sd := NewInternalSessionData(ctx, sc.execCfg.Settings, "validateBackfillQueryIntoTable")
 	sd.SessionData = *sc.sessionData
-	// First get the expected row count for the source query at the target timestamp.
-	if err := sc.execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-		if skipAOSTValidation {
-			return nil
-		}
-		parsedQuery, err := parser.ParseOne(query)
-		if err != nil {
-			return err
-		}
-		// If the query has an AOST clause, then we will remove it here.
-		selectTop, ok := parsedQuery.AST.(*tree.Select)
-		if ok {
-			selectStmt := selectTop.Select
-			var parenSel *tree.ParenSelect
-			var ok bool
-			for parenSel, ok = selectStmt.(*tree.ParenSelect); ok; parenSel, ok = selectStmt.(*tree.ParenSelect) {
-				selectStmt = parenSel.Select.Select
-			}
-			sc, ok := selectStmt.(*tree.SelectClause)
-			if ok && sc.From.AsOf.Expr != nil {
-				sc.From.AsOf.Expr = nil
-				query = parsedQuery.AST.String()
-			}
-		}
-		// Inject the query and the time we should scan at.
-		err = txn.KV().SetFixedTimestamp(ctx, queryTS)
-		if err != nil {
-			return err
-		}
-		countQuery := fmt.Sprintf("SELECT count(*) FROM (%s)", query)
-		row, err := txn.QueryRow(ctx, "backfill-query-src-count", txn.KV(), countQuery)
-		if err != nil {
-			return err
-		}
-		aostEntryCount = int64(tree.MustBeDInt(row[0]))
-		return nil
-	}, isql.WithSessionData(sd),
-		isql.WithPriority(admissionpb.BulkNormalPri),
-	); err != nil {
-		return err
-	}
-	// Next run validation on table that was populated using count queries.
+	// Run validation on table that was populated using count queries.
 	// Get rid of the table ID prefix.
 	index := table.GetPrimaryIndex()
 	now := sc.db.KV().Clock().Now()
@@ -390,7 +343,7 @@ func (sc *SchemaChanger) validateBackfillQueryIntoTable(
 	// Testing knob that allows us to manipulate counts to fail
 	// the validation.
 	if sc.testingKnobs.RunDuringQueryBackfillValidation != nil {
-		newTblEntryCount, err = sc.testingKnobs.RunDuringQueryBackfillValidation(aostEntryCount, newTblEntryCount)
+		newTblEntryCount, err = sc.testingKnobs.RunDuringQueryBackfillValidation(entryCountWrittenToPrimaryIdx, newTblEntryCount)
 		if err != nil {
 			return err
 		}
@@ -403,9 +356,6 @@ func (sc *SchemaChanger) validateBackfillQueryIntoTable(
 			)
 		}
 	}
-	if !skipAOSTValidation && newTblEntryCount != aostEntryCount {
-		return errors.AssertionFailedf("backfill query did not populate index %q with expected number of rows (expected: %d, got: %d)", index.GetName(), aostEntryCount, newTblEntryCount)
-	}
 	return nil
 }
 
@@ -435,8 +385,6 @@ func (sc *SchemaChanger) backfillQueryIntoTable(
 			}
 		}
 	}()
-	validationTime := ts
-	var skipAOSTValidation bool
 	bulkSummary := kvpb.BulkOpSummary{}
 
 	err = sc.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
@@ -596,23 +544,7 @@ func (sc *SchemaChanger) backfillQueryIntoTable(
 				return
 			}
 		})
-
-		if planAndRunErr != nil {
-			return planAndRunErr
-		}
-		// Otherwise, the select statement had no fixed timestamp. For
-		// validating counts we will start with the read timestamp.
-		if ts.IsEmpty() {
-			validationTime = txn.KV().ReadTimestamp()
-		}
-		// If a virtual table was used then we can't conduct any kind of validation,
-		// since our AOST query might be reading data not in KV.
-		for _, tbl := range localPlanner.curPlan.mem.Metadata().AllTables() {
-			if tbl.Table.IsVirtualTable() {
-				skipAOSTValidation = true
-			}
-		}
-		return nil
+		return planAndRunErr
 	})
 
 	// BatchTimestampBeforeGCError is retryable for the schema changer, but we
@@ -638,15 +570,14 @@ func (sc *SchemaChanger) backfillQueryIntoTable(
 
 	// Validation will be skipped if the query is not AOST safe.
 	// (i.e. reading non-KV data via CRDB internal)
-	if err := sc.validateBackfillQueryIntoTable(ctx, table, entriesWrittenToPrimaryIdx, validationTime, query, skipAOSTValidation); err != nil {
+	if err := sc.validateBackfillQueryIntoTable(ctx, table, entriesWrittenToPrimaryIdx); err != nil {
 		return err
 	}
 	return nil
 }
 
 // maybe backfill a created table by executing the AS query. Return nil if
 // successfully backfilled.
-//
 // Note that this does not connect to the tracing settings of the
 // surrounding SQL transaction. This should be OK as (at the time of
 // this writing) this code path is only used for standalone CREATE

pkg/sql/schema_changer_test.go

@@ -8033,11 +8033,15 @@ func TestCreateTableAsValidationFailure(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	var hookEnabled atomic.Bool
 	s, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{
 		Knobs: base.TestingKnobs{
 			SQLSchemaChanger: &sql.SchemaChangerTestingKnobs{
 				RunDuringQueryBackfillValidation: func(expectedCount int64, currentCount int64) (newCurrentCount int64, err error) {
-					return 0, nil
+					if hookEnabled.Load() {
+						return 0, nil
+					}
+					return currentCount, nil
 				},
 			},
 		},
@@ -8050,7 +8054,13 @@ func TestCreateTableAsValidationFailure(t *testing.T) {
 	runner.Exec(t, "INSERT INTO t1 VALUES (1)")
 	runner.Exec(t, "INSERT INTO t1 VALUES (2)")
 	runner.Exec(t, "INSERT INTO t1 VALUES (3)")
+	// Validate cases that should work
+	runner.Exec(t, "CREATE TABLE simple_copy AS (SELECT * FROM t1)")
+	// Row count is can change.
+	runner.Exec(t, "CREATE TABLE t_random AS (SELECT * FROM t1 WHERE random() > 0.5)")
+	runner.Exec(t, " CREATE TABLE t_random2 AS (SELECT * FROM generate_series(0, CAST((100 * random()) AS INT)));")
 	// Execute a CTAS and CREATE MATERIALIZED VIEW statements that should fail.
+	hookEnabled.Store(true)
 	runner.ExpectErr(t, "backfill query did not populate index \"t2_pkey\" with expected number of rows", "CREATE TABLE t2 AS (SELECT * FROM t1)")
 	runner.ExpectErr(t, "backfill query did not populate index \"t2_pkey\" with expected number of rows", "CREATE MATERIALIZED VIEW t2 AS (SELECT n FROM t1)")
 }