Merge #144049 #144508

144049: sql/crdb_internal: rename alter_statements to fk_statements r=rafiss a=littlestar642

Fixes: #143762

Update name of `alter_statements` column to `fk_statements` when executing below sql

Previous
```
SELECT alter_statements FROM crdb_internal.create_statements;
```

Current
```
SELECT fk_statements FROM crdb_internal.create_statements;
```

144508: sql: support partial index schema changes in LDR tables using the SQL writer r=msbutler a=msbutler

If logical_replication.consumer.immediate_mode_writer is set to legacy-kv, we
disallow schema changes related to partial indexes on table with LDR jobs
running on them.

Note that if a user runs LDR in the undocumented validated mode, which always
uses the sql writer, but has this setting equal 'legacy-kv', partial index
schema changes will still fail. This is a papercuit i'm willing to deal with.

Epic: none

Release note (ops change): partial index schema changes are supported in
replicating tables when logical_replication.consumer.immediate_mode_writer is
not set to legacy-kv

Co-authored-by: avinash jaiswal <[email protected]>
Co-authored-by: Michael Butler <[email protected]>

Author: 3 people

pkg/ccl/logictestccl/testdata/logic_test/crdb_internal_tenant

@@ -190,7 +190,7 @@ function  signature  category  details  schema  oid
 query ITTITTTTTTTTTBBBB colnames
 SELECT * FROM crdb_internal.create_statements WHERE database_name = ''
 ----
-database_id  database_name  schema_name  descriptor_id  descriptor_type  descriptor_name  create_statement  state  create_nofks  rls_statements  alter_statements  validate_statements  create_redactable  has_partitions  is_multi_region  is_virtual  is_temporary
+database_id  database_name  schema_name  descriptor_id  descriptor_type  descriptor_name  create_statement  state  create_nofks  rls_statements  fk_statements  validate_statements  create_redactable  has_partitions  is_multi_region  is_virtual  is_temporary
 
 query ITITTBTB colnames
 SELECT * FROM crdb_internal.table_columns WHERE descriptor_name = ''

pkg/cli/zip_table_registry.go

@@ -307,7 +307,7 @@ var zipInternalTablesPerCluster = DebugZipTableRegistry{
 			"is_virtual",
 			"is_temporary",
 			"crdb_internal.hide_sql_constants(create_statement) as create_statement",
-			"crdb_internal.hide_sql_constants(alter_statements) as alter_statements",
+			"crdb_internal.hide_sql_constants(fk_statements) as fk_statements",
 			"crdb_internal.hide_sql_constants(create_nofks) as create_nofks",
 			"crdb_internal.redact(create_redactable) as create_redactable",
 		},

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"
@@ -2179,7 +2180,6 @@ func TestLogicalReplicationSchemaChanges(t *testing.T) {
 		// adding unsuppored indexes disallowed
 		{"add virtual column index", "CREATE INDEX virtual_col_idx ON tab(virtual_col)", false},
 		{"add hash index", "CREATE INDEX hash_idx ON tab(pk) USING HASH WITH (bucket_count = 4)", false},
-		{"add partial index", "CREATE INDEX partial_idx ON tab(composite_col) WHERE pk > 0", false},
 		{"add unique index", "CREATE UNIQUE INDEX unique_idx ON tab(composite_col)", false},
 
 		// Drop table is blocked
@@ -2420,9 +2420,17 @@ func TestPartialIndexes(t *testing.T) {
 	dbA.CheckQueryResultsRetry(t, "SELECT * FROM foo WHERE pi = 2", [][]string{{"1", "2", "hello"}})
 	dbA.CheckQueryResultsRetry(t, "SELECT * FROM foo WHERE pi = -2", [][]string{{"-1", "-2", "world"}})
 
+	// Check that dbA and dbB can add an additional partial index, enabled when using the sql writer.
+	dbA.Exec(t, "CREATE INDEX idx2 ON a.foo (pi) WHERE pk = 0")
+	dbB.Exec(t, "CREATE INDEX idx2 ON b.foo (pi) WHERE pk = 0")
+
 	// Check for partial indexes when using legacy kv writer.
 	dbA.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.immediate_mode_writer = 'legacy-kv'")
 	dbA.ExpectErr(t, "cannot create logical replication stream: table foo has a partial index idx", "CREATE LOGICAL REPLICATION STREAM FROM TABLE foo ON $1 INTO TABLE foo", dbBURL.String())
+
+	// With the kv writer set, we can no longer create partial indexes on the replicating tables.
+	dbA.ExpectErr(t, " this schema change is disallowed on table foo because it is referenced by one or more logical replication jobs", "CREATE INDEX idx3 ON a.foo (pi) WHERE pk = 0")
+	dbB.ExpectErr(t, " this schema change is disallowed on table foo because it is referenced by one or more logical replication jobs", "CREATE INDEX idx3 ON b.foo (pi) WHERE pk = 0")
 }
 
 // TestLogicalReplicationCreationChecks verifies that we check that the table
@@ -2666,7 +2674,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 +2685,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 +2694,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/alter_index.go

@@ -46,7 +46,7 @@ func (p *planner) AlterIndex(ctx context.Context, n *tree.AlterIndex) (planNode,
 	}
 
 	// Disallow schema changes if this table's schema is locked.
-	if err := checkSchemaChangeIsAllowed(tableDesc, n); err != nil {
+	if err := checkSchemaChangeIsAllowed(tableDesc, n, p.ExecCfg().Settings); err != nil {
 		return nil, err
 	}
 

pkg/sql/alter_index_visible.go

@@ -66,7 +66,7 @@ func (p *planner) AlterIndexVisible(
 	}
 
 	// Disallow schema changes if this table's schema is locked.
-	if err := checkSchemaChangeIsAllowed(tableDesc, n); err != nil {
+	if err := checkSchemaChangeIsAllowed(tableDesc, n, p.ExecCfg().Settings); err != nil {
 		return nil, err
 	}