sqlclustersettings: move LDR writer mode to sqlclustersettings package

Currently, a subset of schema changes are allowed on all LDR jobs, but some of
these allowed schema changes need to be disallowed when
logical_replication.consumer.immediate_mode_writer is set to `legacy-kv'. To
that end, we need to move this cluster setting the sqlclustersettings package
so that the schema change planner can cleanly check this setting.

Epic: none

Release note: none

Author: msbutler

pkg/crosscluster/logical/BUILD.bazel

@@ -86,6 +86,7 @@ go_library(
         "//pkg/sql/sem/tree/treecmp",
         "//pkg/sql/sessiondata",
         "//pkg/sql/sessiondatapb",
+        "//pkg/sql/sqlclustersettings",
         "//pkg/sql/stats",
         "//pkg/sql/syntheticprivilege",
         "//pkg/sql/types",
@@ -178,6 +179,7 @@ go_test(
         "//pkg/sql/sem/idxtype",
         "//pkg/sql/sem/tree",
         "//pkg/sql/sessiondata",
+        "//pkg/sql/sqlclustersettings",
         "//pkg/sql/sqltestutils",
         "//pkg/sql/stats",
         "//pkg/testutils",

pkg/crosscluster/logical/create_logical_replication_stmt.go

@@ -40,6 +40,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlclustersettings"
 	"github.com/cockroachdb/cockroach/pkg/sql/syntheticprivilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/buildutil"
@@ -489,7 +490,7 @@ func doLDRPlan(
 				}
 			}
 
-			err := tabledesc.CheckLogicalReplicationCompatibility(&srcExternalCatalog.Tables[i], destTableDesc.TableDesc(), details.SkipSchemaCheck || details.CreateTable, writer == writerTypeLegacyKV)
+			err := tabledesc.CheckLogicalReplicationCompatibility(&srcExternalCatalog.Tables[i], destTableDesc.TableDesc(), details.SkipSchemaCheck || details.CreateTable, writer == sqlclustersettings.LDRWriterTypeLegacyKV)
 			if err != nil {
 				return err
 			}

pkg/crosscluster/logical/logical_replication_dist.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlclustersettings"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 )
@@ -36,7 +37,7 @@ func constructLogicalReplicationWriterSpecs(
 	discard jobspb.LogicalReplicationDetails_Discard,
 	mode jobspb.LogicalReplicationDetails_ApplyMode,
 	metricsLabel string,
-	writer writerType,
+	writer sqlclustersettings.LDRWriterType,
 ) (map[base.SQLInstanceID][]execinfrapb.LogicalReplicationWriterSpec, error) {
 	spanGroup := roachpb.SpanGroup{}
 	baseSpec := execinfrapb.LogicalReplicationWriterSpec{

pkg/crosscluster/logical/logical_replication_job.go

@@ -34,6 +34,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlclustersettings"
 	"github.com/cockroachdb/cockroach/pkg/util/bulk"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
@@ -539,7 +540,7 @@ 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 {
+			if err := tabledesc.CheckLogicalReplicationCompatibility(&srcTableDesc, dstTableDesc.TableDesc(), payload.SkipSchemaCheck || payload.CreateTable, writer == sqlclustersettings.LDRWriterTypeLegacyKV); err != nil {
 				return err
 			}
 

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -44,6 +44,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/idxtype"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlclustersettings"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/jobutils"
@@ -2666,7 +2667,7 @@ func TestGetWriterType(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Validated, st)
 		require.NoError(t, err)
-		require.Equal(t, writerTypeSQL, wt)
+		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 	})
 
 	t.Run("immediate-mode-pre-25.2", func(t *testing.T) {
@@ -2677,7 +2678,7 @@ func TestGetWriterType(t *testing.T) {
 		)
 		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
 		require.NoError(t, err)
-		require.Equal(t, writerTypeSQL, wt)
+		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 	})
 
 	t.Run("immediate-mode-post-25.2", func(t *testing.T) {
@@ -2686,14 +2687,14 @@ func TestGetWriterType(t *testing.T) {
 			clusterversion.PreviousRelease.Version(),
 			true, /* initializeVersion */
 		)
-		immediateModeWriter.Override(ctx, &st.SV, string(writerTypeSQL))
+		sqlclustersettings.LDRImmediateModeWriter.Override(ctx, &st.SV, string(sqlclustersettings.LDRWriterTypeSQL))
 		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
 		require.NoError(t, err)
-		require.Equal(t, writerTypeSQL, wt)
+		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 
-		immediateModeWriter.Override(ctx, &st.SV, string(writerTypeLegacyKV))
+		sqlclustersettings.LDRImmediateModeWriter.Override(ctx, &st.SV, string(sqlclustersettings.LDRWriterTypeSQL))
 		wt, err = getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
 		require.NoError(t, err)
-		require.Equal(t, writerTypeLegacyKV, wt)
+		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 	})
 }

pkg/crosscluster/logical/logical_replication_writer_processor.go

@@ -37,13 +37,13 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlclustersettings"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/bulk"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/log/logcrash"
-	"github.com/cockroachdb/cockroach/pkg/util/metamorphic"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
@@ -95,34 +95,6 @@ var maxChunkSize = settings.RegisterIntSetting(
 	settings.NonNegativeInt,
 )
 
-type writerType string
-
-const (
-	// writerTypeSQL uses the SQL layer to write replicated rows.
-	writerTypeSQL writerType = "sql"
-	// writerTypeLegacyKV uses the legacy KV layer to write rows. The KV writer
-	// is deprecated because it does not support the full set of features of the
-	// SQL writer.
-	writerTypeLegacyKV writerType = "legacy-kv"
-
-	// writerTypeCRUD is the shiny new sql writer that uses explicit reads,
-	// inserts, updates, and deletes instead of upserts.
-	writerTypeCRUD writerType = "crud"
-)
-
-var immediateModeWriter = settings.RegisterStringSetting(
-	settings.ApplicationLevel,
-	"logical_replication.consumer.immediate_mode_writer",
-	"the writer to use when in immediate mode",
-	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(writerTypeSQL), string(writerTypeLegacyKV), string(writerTypeCRUD)),
-	settings.WithValidateString(func(sv *settings.Values, val string) error {
-		if val != string(writerTypeSQL) && val != string(writerTypeLegacyKV) && val != string(writerTypeCRUD) {
-			return errors.Newf("immediate mode writer must be either 'sql', 'legacy-kv', or 'crud', got '%s'", val)
-		}
-		return nil
-	}),
-)
-
 // logicalReplicationWriterProcessor consumes a cross-cluster replication stream
 // by decoding kvs in it to logical changes and applying them by executing DMLs.
 type logicalReplicationWriterProcessor struct {
@@ -719,7 +691,7 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 		b.Close(lrw.Ctx())
 	}
 
-	writer := writerType(lrw.spec.WriterType)
+	writer := sqlclustersettings.LDRWriterType(lrw.spec.WriterType)
 	if writer == "" && !lrw.FlowCtx.Cfg.Settings.Version.IsActive(ctx, clusterversion.V25_2) {
 		var err error
 		writer, err = getWriterType(
@@ -738,7 +710,7 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 		sd := sql.NewInternalSessionData(ctx, flowCtx.Cfg.Settings, "" /* opName */)
 
 		switch writer {
-		case writerTypeSQL:
+		case sqlclustersettings.LDRWriterTypeSQL:
 			rp, err = makeSQLProcessor(
 				ctx, flowCtx.Cfg.Settings, lrw.configByTable,
 				jobspb.JobID(lrw.spec.JobID),
@@ -753,12 +725,12 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 			if err != nil {
 				return err
 			}
-		case writerTypeLegacyKV:
+		case sqlclustersettings.LDRWriterTypeLegacyKV:
 			rp, err = newKVRowProcessor(ctx, flowCtx.Cfg, flowCtx.EvalCtx, lrw.spec, lrw.configByTable)
 			if err != nil {
 				return err
 			}
-		case writerTypeCRUD:
+		case sqlclustersettings.LDRWriterTypeCRUD:
 			rp, err = newCrudSqlWriter(ctx, flowCtx.Cfg, flowCtx.EvalCtx, sd, lrw.spec.Discard, lrw.configByTable, jobspb.JobID(lrw.spec.JobID))
 			if err != nil {
 				return err
@@ -780,17 +752,17 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 
 func getWriterType(
 	ctx context.Context, mode jobspb.LogicalReplicationDetails_ApplyMode, settings *cluster.Settings,
-) (writerType, error) {
+) (sqlclustersettings.LDRWriterType, error) {
 	switch mode {
 	case jobspb.LogicalReplicationDetails_Immediate:
 		// Require v25.2 to use the sql writer by default to ensure that the
 		// KV origin timestamp validation is available on all nodes.
 		if settings.Version.IsActive(ctx, clusterversion.V25_2) {
-			return writerType(immediateModeWriter.Get(&settings.SV)), nil
+			return sqlclustersettings.LDRWriterType(sqlclustersettings.LDRImmediateModeWriter.Get(&settings.SV)), nil
 		}
-		return writerTypeSQL, nil
+		return sqlclustersettings.LDRWriterTypeSQL, nil
 	case jobspb.LogicalReplicationDetails_Validated:
-		return writerTypeSQL, nil
+		return sqlclustersettings.LDRWriterTypeSQL, nil
 	default:
 		return "", errors.Newf("unknown logical replication writer type: %s", mode)
 	}

pkg/sql/sqlclustersettings/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "//pkg/keys",
         "//pkg/settings",
         "//pkg/settings/cluster",
+        "//pkg/util/metamorphic",
         "@com_github_cockroachdb_errors//:errors",
     ],
 )

pkg/sql/sqlclustersettings/clustersettings.go

@@ -9,6 +9,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/util/metamorphic"
 	"github.com/cockroachdb/errors"
 )
 
@@ -109,3 +110,30 @@ var CachedSequencesCacheSizeSetting = settings.RegisterIntSetting(
 	256,
 	settings.PositiveInt,
 )
+
+type LDRWriterType string
+
+const (
+	// LDRWriterTypeSQL uses the SQL layer to write replicated rows.
+	LDRWriterTypeSQL LDRWriterType = "sql"
+	// LDRWriterTypeLegacyKV uses the legacy KV layer to write rows. The KV writer
+	// is deprecated because it does not support the full set of features of the
+	// SQL writer.
+	LDRWriterTypeLegacyKV LDRWriterType = "legacy-kv"
+	// writerTypeCRUD is the shiny new sql writer that uses explicit reads,
+	// inserts, updates, and deletes instead of upserts.
+	LDRWriterTypeCRUD LDRWriterType = "crud"
+)
+
+var LDRImmediateModeWriter = settings.RegisterStringSetting(
+	settings.ApplicationLevel,
+	"logical_replication.consumer.immediate_mode_writer",
+	"the writer to use when in immediate mode",
+	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeSQL), string(LDRWriterTypeLegacyKV), string(LDRWriterTypeCRUD)),
+	settings.WithValidateString(func(sv *settings.Values, val string) error {
+		if val != string(LDRWriterTypeSQL) && val != string(LDRWriterTypeLegacyKV) && val != string(LDRWriterTypeCRUD) {
+			return errors.Newf("immediate mode writer must be either 'sql', 'legacy-kv', or 'crud', got '%s'", val)
+		}
+		return nil
+	}),
+)