Merge #144308 #144318 #144362

144308: roachtest: fix backup-restore compaction overlap bug r=msbutler a=kev-cao

The previous implementation of the roachtest did not remove backup end times from the list of candidate end times for compaction after performing a compaction. This could result in scenarios where if multiple compactions were attempted, it could choose a start time of a backup that had been compacted. Because the compaction job compacts backups from an elided backup chain, those compacted backups would not appear in the list of backups and therefore the job would fail due to not being able to find the starting backup.

Informs: #144216

Release note: None

144318: crosscluster/logical: validate schema on resume r=jeffswenson a=msbutler

Because the user can toggle the use of the kv writer via a cluster setting, we
need to validate that their replicating tables comply with their writer mode
whenever we spin up the LDR distsql flow.

This patch also slightly refactors the input to the validation checker during
planning to read from the writer type cluster setting. This means immediate
mode using the the sql writer can replicate even if the column ids in the
tables don't match.

Epic: none

Release note: none

144362: scripts: add 25.2 to scripts/check-pebble-dep.sh r=RaduBerinde a=RaduBerinde

Epic: none
Release note: None

Co-authored-by: Kevin Cao <[email protected]>
Co-authored-by: Michael Butler <[email protected]>
Co-authored-by: Radu Berinde <[email protected]>

Author: 4 people

pkg/cmd/roachtest/tests/mixed_version_backup.go

@@ -15,6 +15,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"regexp"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -2076,7 +2077,7 @@ func (d *BackupRestoreTestDriver) createBackupCollection(
 		numIncrementals = 2
 	}
 	l.Printf("creating %d incremental backups", numIncrementals)
-	for i := 0; i < numIncrementals; i++ {
+	for i := range numIncrementals {
 		d.randomWait(l, rng)
 		if err := d.testUtils.runJobOnOneOf(ctx, l, incBackupSpec.Execute.Nodes, func() error {
 			var err error
@@ -2117,6 +2118,14 @@ func (d *BackupRestoreTestDriver) createBackupCollection(
 			}); err != nil {
 				return nil, err
 			}
+			// Since a compacted backup was made, then the backup end times of the
+			// backups it compacted should be removed from the slice. This prevents a
+			// scenario where a later compaction attempts to pick a start time from
+			// one of the backups that were compacted. Since compaction looks at the
+			// elided backup chain, these compacted backups are replaced by the
+			// compacted backup and compaction will fail due to being unable to find
+			// the starting backup.
+			backupEndTimes = slices.Delete(backupEndTimes, startIdx+1, endIdx)
 		}
 	}
 

pkg/crosscluster/logical/create_logical_replication_stmt.go

@@ -323,10 +323,11 @@ func createLogicalReplicationStreamPlanHook(
 				ReverseStreamCommand:      reverseStreamCmd,
 				ParentID:                  int64(options.ParentID),
 				Command:                   stmt.String(),
+				SkipSchemaCheck:           options.SkipSchemaCheck(),
 			},
 			Progress: progress,
 		}
-		if err := doLDRPlan(ctx, p.User(), p.ExecCfg(), jr, spec.ExternalCatalog, resolvedDestObjects, options.SkipSchemaCheck()); err != nil {
+		if err := doLDRPlan(ctx, p.User(), p.ExecCfg(), jr, spec.ExternalCatalog, resolvedDestObjects); err != nil {
 			return err
 		}
 		resultsCh <- tree.Datums{tree.NewDInt(tree.DInt(jr.JobID))}
@@ -430,7 +431,6 @@ func doLDRPlan(
 	jr jobs.Record,
 	srcExternalCatalog externalpb.ExternalCatalog,
 	resolvedDestObjects ResolvedDestObjects,
-	skipSchemaCheck bool,
 ) error {
 	details := jr.Details.(jobspb.LogicalReplicationDetails)
 	return execCfg.InternalDB.DescsTxn(ctx, func(ctx context.Context, txn descs.Txn) error {
@@ -475,16 +475,21 @@ func doLDRPlan(
 				return errors.AssertionFailedf("srcTableDescs and dstTableDescs should have the same length")
 			}
 		}
+
+		writer, err := getWriterType(ctx, details.Mode, execCfg.Settings)
+		if err != nil {
+			return err
+		}
+
 		for i := range srcExternalCatalog.Tables {
 			destTableDesc := dstTableDescs[i]
-			mayUseKVWriter := false
 			if details.Mode != jobspb.LogicalReplicationDetails_Validated {
 				if len(destTableDesc.OutboundForeignKeys()) > 0 || len(destTableDesc.InboundForeignKeys()) > 0 {
 					return pgerror.Newf(pgcode.InvalidParameterValue, "foreign keys are only supported with MODE = 'validated'")
 				}
-				mayUseKVWriter = true
 			}
-			err := tabledesc.CheckLogicalReplicationCompatibility(&srcExternalCatalog.Tables[i], destTableDesc.TableDesc(), skipSchemaCheck || details.CreateTable, mayUseKVWriter)
+
+			err := tabledesc.CheckLogicalReplicationCompatibility(&srcExternalCatalog.Tables[i], destTableDesc.TableDesc(), details.SkipSchemaCheck || details.CreateTable, writer == writerTypeLegacyKV)
 			if err != nil {
 				return err
 			}

pkg/crosscluster/logical/logical_replication_dist.go

@@ -36,6 +36,7 @@ func constructLogicalReplicationWriterSpecs(
 	discard jobspb.LogicalReplicationDetails_Discard,
 	mode jobspb.LogicalReplicationDetails_ApplyMode,
 	metricsLabel string,
+	writer writerType,
 ) (map[base.SQLInstanceID][]execinfrapb.LogicalReplicationWriterSpec, error) {
 	spanGroup := roachpb.SpanGroup{}
 	baseSpec := execinfrapb.LogicalReplicationWriterSpec{
@@ -50,6 +51,7 @@ func constructLogicalReplicationWriterSpecs(
 		Mode:                        mode,
 		MetricsLabel:                metricsLabel,
 		TypeDescriptors:             srcTypes,
+		WriterType:                  string(writer),
 	}
 
 	writerSpecs := make(map[base.SQLInstanceID][]execinfrapb.LogicalReplicationWriterSpec, len(destSQLInstances))

pkg/crosscluster/logical/logical_replication_job.go

@@ -510,7 +510,10 @@ func (p *logicalReplicationPlanner) generatePlanImpl(
 	if defaultFnID := payload.DefaultConflictResolution.FunctionId; defaultFnID != 0 {
 		defaultFnOID = catid.FuncIDToOID(catid.DescID(defaultFnID))
 	}
-
+	writer, err := getWriterType(ctx, payload.Mode, execCfg.Settings)
+	if err != nil {
+		return nil, nil, info, err
+	}
 	crossClusterResolver := crosscluster.MakeCrossClusterTypeResolver(plan.SourceTypes)
 	tableMetadataByDestID := make(map[int32]execinfrapb.TableReplicationMetadata)
 	if err := sql.DescsTxn(ctx, execCfg, func(ctx context.Context, txn isql.Txn, descriptors *descs.Collection) error {
@@ -536,6 +539,10 @@ func (p *logicalReplicationPlanner) generatePlanImpl(
 				return errors.Wrapf(err, "failed to look up schema descriptor for table %d", pair.DstDescriptorID)
 			}
 
+			if err := tabledesc.CheckLogicalReplicationCompatibility(&srcTableDesc, dstTableDesc.TableDesc(), payload.SkipSchemaCheck || payload.CreateTable, writer == writerTypeLegacyKV); err != nil {
+				return err
+			}
+
 			var fnOID oid.Oid
 			if pair.DstFunctionID != 0 {
 				fnOID = catid.FuncIDToOID(catid.DescID(pair.DstFunctionID))
@@ -590,6 +597,7 @@ func (p *logicalReplicationPlanner) generatePlanImpl(
 		payload.Discard,
 		payload.Mode,
 		payload.MetricsLabel,
+		writer,
 	)
 	if err != nil {
 		return nil, nil, info, err

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -2428,7 +2428,17 @@ func TestMismatchColIDs(t *testing.T) {
 	defer log.Scope(t).Close(t)
 
 	ctx := context.Background()
-	tc, s, sqlA, sqlB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
+
+	streamingKnobs := &sql.StreamingTestingKnobs{
+		DistSQLRetryPolicy: &retry.Options{
+			InitialBackoff: time.Microsecond,
+			MaxBackoff:     2 * time.Microsecond,
+			MaxRetries:     1,
+		},
+	}
+	args := testClusterBaseClusterArgs
+	args.ServerArgs.Knobs.Streaming = streamingKnobs
+	tc, s, sqlA, sqlB := setupLogicalTestServer(t, ctx, args, 1)
 	defer tc.Stopper().Stop(ctx)
 
 	dbBURL := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("b"))
@@ -2449,17 +2459,29 @@ func TestMismatchColIDs(t *testing.T) {
 	sqlB.Exec(t, "ALTER TABLE foo DROP COLUMN bar")
 	sqlB.Exec(t, "INSERT INTO foo VALUES (4, 'world')")
 
-	// LDR immediate mode creation should fail because of mismatched column IDs.
+	// When using the kv writer, LDR immediate mode creation should fail because of mismatched column IDs.
+	sqlA.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.immediate_mode_writer = 'legacy-kv'")
 	sqlA.ExpectErr(t,
 		"destination table foo column baz has ID 3, but the source table foo has ID 4",
 		"CREATE LOGICAL REPLICATION STREAM FROM TABLE foo ON $1 INTO TABLE foo WITH MODE = 'immediate'", dbBURL.String())
 
-	// LDR validated mode creation should succeed because the SQL writer supports mismatched column IDs.
 	var jobID jobspb.JobID
 	sqlA.QueryRow(t, "CREATE LOGICAL REPLICATION STREAM FROM TABLE foo ON $1 INTO TABLE foo WITH MODE = 'validated'", dbBURL.String()).Scan(&jobID)
-
 	now := s.Clock().Now()
 	WaitUntilReplicatedTime(t, now, sqlA, jobID)
+
+	sqlA.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.immediate_mode_writer = 'sql'")
+	sqlA.QueryRow(t, "CREATE LOGICAL REPLICATION STREAM FROM TABLE foo ON $1 INTO TABLE foo WITH MODE = 'immediate'", dbBURL.String()).Scan(&jobID)
+	now = s.Clock().Now()
+	WaitUntilReplicatedTime(t, now, sqlA, jobID)
+
+	// Ensure the validation works on resumption as well.
+	sqlA.Exec(t, "PAUSE JOB $1", jobID)
+	jobutils.WaitForJobToPause(t, sqlA, jobID)
+
+	sqlA.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.immediate_mode_writer = 'legacy-kv'")
+	sqlA.Exec(t, "RESUME JOB $1", jobID)
+	jobutils.WaitForJobToPause(t, sqlA, jobID)
 }
 
 // TestLogicalReplicationCreationChecks verifies that we check that the table

pkg/crosscluster/logical/logical_replication_writer_processor.go

@@ -711,17 +711,24 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 		b.Close(lrw.Ctx())
 	}
 
+	writer := writerType(lrw.spec.WriterType)
+	if writer == "" && !lrw.FlowCtx.Cfg.Settings.Version.IsActive(ctx, clusterversion.V25_2) {
+		var err error
+		writer, err = getWriterType(
+			ctx, lrw.spec.Mode, lrw.FlowCtx.Cfg.Settings,
+		)
+		if err != nil {
+			return err
+		}
+	}
+
 	flowCtx := lrw.FlowCtx
 	lrw.bh = make([]BatchHandler, poolSize)
 	for i := range lrw.bh {
 		var rp BatchHandler
 		var err error
 		sd := sql.NewInternalSessionData(ctx, flowCtx.Cfg.Settings, "" /* opName */)
 
-		writer, err := getWriterType(ctx, lrw.spec.Mode, flowCtx.Cfg.Settings)
-		if err != nil {
-			return err
-		}
 		switch writer {
 		case writerTypeSQL:
 			rp, err = makeSQLProcessor(
@@ -744,7 +751,7 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 				return err
 			}
 		default:
-			return errors.AssertionFailedf("unknown logical replication writer type: %s", writer)
+			return errors.AssertionFailedf("unknown logical replication writer type: %s", lrw.spec.WriterType)
 		}
 
 		if streamingKnobs, ok := flowCtx.TestingKnobs().StreamingTestingKnobs.(*sql.StreamingTestingKnobs); ok {

pkg/jobs/jobspb/jobs.proto

@@ -292,6 +292,8 @@ message LogicalReplicationDetails {
 
   string command = 16;
 
+  bool skip_schema_check = 17;
+
   // Next ID: 17.
 }
 

pkg/sql/catalog/tabledesc/logical_replication_helpers.go

@@ -218,7 +218,7 @@ func checkSrcDstColsMatch(
 		}
 
 		if requireKvWriterCompatible && srcCol.ID != dstCol.ID {
-			return errors.Newf("destination table %s column %s has ID %d, but the source table %s has ID %d",
+			return errors.Newf("destination table %s column %s has ID %d, but the source table %s has ID %d. To circumvent this check, unset logical_replication.consumer.immediate_mode_writer from 'legacy-kv'",
 				dst.Name, dstCol.Name, dstCol.ID, src.Name, srcCol.ID,
 			)
 		}

pkg/sql/execinfrapb/processors_bulk_io.proto

@@ -536,7 +536,10 @@ message LogicalReplicationWriterSpec {
     optional string metrics_label = 11 [(gogoproto.nullable) = false];
 
     repeated cockroach.sql.sqlbase.TypeDescriptor type_descriptors = 13;
-    // Next ID: 14.
+
+    optional string writer_type = 14 [(gogoproto.nullable) = false];
+
+    // Next ID: 15.
 }
 
 message LogicalReplicationOfflineScanSpec {

scripts/check-pebble-dep.sh

@@ -9,7 +9,7 @@
 set -euo pipefail
 #set -x
 
-RELEASES="23.2 24.1 24.3 25.1 master"
+RELEASES="23.2 24.1 24.3 25.1 25.2 master"
 
 for REL in $RELEASES; do
   if [ "$REL" == "master" ]; then