sql: support EXCLUDE CONSTRAINTS for SQL Server sources

src/sql/src/pure.rs

@@ -1088,6 +1088,8 @@ async fn purify_create_source(
                 external_references,
                 &text_columns,
                 &exclude_columns,
+                &BTreeSet::new(),
+                false,
                 source_name,
                 timeout,
                 &reference_policy,
@@ -1678,6 +1680,8 @@ async fn purify_alter_source_add_subsources(
                 &requested_references,
                 &text_columns,
                 &exclude_columns,
+                &BTreeSet::new(),
+                false,
                 &unresolved_source_name,
                 timeout,
                 &SourceReferencePolicy::Required,
@@ -1920,7 +1924,9 @@ async fn purify_create_table_from_source(
     if (!exclude_constraints.is_empty() || exclude_all_constraints)
         && !matches!(
             desc.connection,
-            GenericSourceConnection::Postgres(_) | GenericSourceConnection::MySql(_)
+            GenericSourceConnection::Postgres(_)
+                | GenericSourceConnection::MySql(_)
+                | GenericSourceConnection::SqlServer(_)
         )
     {
         sql_bail!(
@@ -2054,6 +2060,8 @@ async fn purify_create_table_from_source(
                 &requested_references,
                 &qualified_text_columns,
                 &qualified_exclude_columns,
+                &exclude_constraints,
+                exclude_all_constraints,
                 &unresolved_source_name,
                 timeout,
                 &SourceReferencePolicy::Required,

src/sql/src/pure/error.rs

@@ -485,6 +485,11 @@ pub enum SqlServerSourcePurificationError {
         option_name: String,
         items: Vec<UnresolvedItemName>,
     },
+    #[error("EXCLUDE CONSTRAINTS refers to constraints that do not exist on table {table}")]
+    ConstraintsNotFound {
+        table: String,
+        constraints: Vec<String>,
+    },
     #[error("found multiple primary keys for a table. constraints {constraint_names:?}")]
     MultiplePrimaryKeys { constraint_names: Vec<Arc<str>> },
     #[error("column {schema_name}.{tbl_name}.{col_name} of type {col_type} is not supported")]
@@ -520,6 +525,13 @@ impl SqlServerSourcePurificationError {
                 "the following columns are referenced but not added: {}",
                 itertools::join(items, ", ")
             )),
+            Self::ConstraintsNotFound {
+                table: _,
+                constraints,
+            } => Some(format!(
+                "the following constraints were not found: {}",
+                constraints.join(", ")
+            )),
             Self::UnsupportedColumn { context, .. } => Some(context.clone()),
             _ => None,
         }
@@ -530,6 +542,11 @@ impl SqlServerSourcePurificationError {
             Self::RequiresExternalReferences => {
                 Some("provide a FOR TABLES (..), FOR SCHEMAS (..), or FOR ALL TABLES clause".into())
             }
+            Self::ConstraintsNotFound { .. } => Some(
+                "Constraint names are matched exactly, including case, against the upstream \
+                 PRIMARY KEY and UNIQUE constraint names."
+                    .into(),
+            ),
             Self::UnnecessaryOptionsWithoutReferences(option) => Some(format!(
                 "Remove the {} option, as no tables are being added.",
                 option

src/sql/src/pure/sql_server.rs

@@ -59,6 +59,12 @@ pub(super) async fn purify_source_exports(
     requested_references: &Option<ExternalReferences>,
     text_columns: &[UnresolvedItemName],
     excl_columns: &[UnresolvedItemName],
+    // NOTE: `exclude_constraints` and `exclude_all_constraints` are only ever
+    // non-empty/true for `CREATE TABLE .. FROM SOURCE`, which purifies exactly
+    // one export, so constraint names are validated against that single
+    // table's constraints.
+    exclude_constraints: &BTreeSet<String>,
+    exclude_all_constraints: bool,
     unresolved_source_name: &UnresolvedItemName,
     timeout: Duration,
     reference_policy: &SourceReferencePolicy,
@@ -261,6 +267,36 @@ pub(super) async fn purify_source_exports(
             })?
         }
 
+        let missing_exclude_constraints: Vec<_> = exclude_constraints
+            .iter()
+            .filter(|n| !table.constraints.iter().any(|c| &&c.constraint_name == n))
+            .cloned()
+            .collect();
+        if !missing_exclude_constraints.is_empty() {
+            return Err(SqlServerSourcePurificationError::ConstraintsNotFound {
+                table: format!("{}.{}", table.schema_name, table.name),
+                constraints: missing_exclude_constraints,
+            }
+            .into());
+        }
+        table
+            .constraints
+            .retain(|c| !exclude_constraints.contains(&c.constraint_name));
+        if exclude_all_constraints {
+            // Marking columns as nullable allows dropping (and adding) the
+            // NOT NULL constraint without an outage. NOTE: SQL Server detects
+            // incompatible upstream DDL via a textual scan of cdc.ddl_history
+            // rather than the descriptor, so an upstream ALTER COLUMN
+            // (including NOT NULL changes) still errors the table regardless
+            // of this option.
+            table.constraints.clear();
+            for column in &mut table.columns {
+                if let Some(column_type) = &mut column.column_type {
+                    column_type.nullable = true;
+                }
+            }
+        }
+
         tables.push(requested.change_meta(table));
     }
 

test/sql-server-cdc/27-exclude-constraints.td

@@ -0,0 +1,182 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+#
+# EXCLUDE CONSTRAINTS / EXCLUDE ALL CONSTRAINTS: excluded upstream constraints
+# are not recorded as Materialize keys, so dropping them upstream is a
+# non-event instead of stalling the table. NOTE: SQL Server does not allow
+# dropping a PRIMARY KEY while CDC is enabled, so the drop scenarios use UNIQUE
+# constraints. Upstream ALTER COLUMN still errors the table regardless of these
+# options, so EXCLUDE ALL CONSTRAINTS is checked for key recording and
+# nullability only.
+#
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET enable_exclude_constraints_option = true
+
+$ sql-server-connect name=sql-server
+server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password};Database=test
+
+> CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
+
+> DROP CONNECTION IF EXISTS sql_server_test_27_connection CASCADE
+> CREATE CONNECTION sql_server_test_27_connection TO SQL SERVER (
+    HOST 'sql-server',
+    PORT 1433,
+    DATABASE test,
+    USER '${arg.default-sql-server-user}',
+    PASSWORD = SECRET sql_server_pass
+  );
+
+$ sql-server-execute name=sql-server
+CREATE TABLE t27_uniq (id INT NOT NULL, wallet VARCHAR(64) NOT NULL, CONSTRAINT t27_pk PRIMARY KEY (id), CONSTRAINT t27_uq_wallet UNIQUE (wallet));
+INSERT INTO t27_uniq VALUES (1, 'a'), (2, 'b');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't27_uniq', @role_name = 'SA', @supports_net_changes = 0;
+
+CREATE TABLE t27_all (id INT NOT NULL, email VARCHAR(64) NOT NULL, CONSTRAINT t27_all_pk PRIMARY KEY (id));
+INSERT INTO t27_all VALUES (1, 'a@b.c');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't27_all', @role_name = 'SA', @supports_net_changes = 0;
+
+CREATE TABLE t27_stall (id INT NOT NULL, wallet VARCHAR(64) NOT NULL, CONSTRAINT t27_stall_pk PRIMARY KEY (id), CONSTRAINT t27_stall_uq UNIQUE (wallet));
+INSERT INTO t27_stall VALUES (1, 'a');
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 't27_stall', @role_name = 'SA', @supports_net_changes = 0;
+
+> CREATE SOURCE t27_source FROM SQL SERVER CONNECTION sql_server_test_27_connection;
+
+#
+# Validation errors
+#
+
+! CREATE TABLE t27_uniq FROM SOURCE t27_source (REFERENCE t27_uniq)
+  WITH (EXCLUDE CONSTRAINTS ('nope'));
+contains:EXCLUDE CONSTRAINTS refers to constraints that do not exist on table dbo.t27_uniq
+
+! CREATE TABLE t27_uniq FROM SOURCE t27_source (REFERENCE t27_uniq)
+  WITH (EXCLUDE CONSTRAINTS ('t27_uq_wallet'), EXCLUDE ALL CONSTRAINTS);
+contains:EXCLUDE ALL CONSTRAINTS cannot be combined with EXCLUDE CONSTRAINTS
+
+#
+# Excluding a named UNIQUE constraint: the PRIMARY KEY survives, the excluded
+# constraint is not recorded as a key, and its later upstream drop is a
+# non-event.
+#
+
+> CREATE TABLE t27_uniq FROM SOURCE t27_source (REFERENCE t27_uniq)
+  WITH (EXCLUDE CONSTRAINTS ('t27_uq_wallet'));
+
+> SELECT * FROM t27_uniq;
+1 a
+2 b
+
+> CREATE DEFAULT INDEX ON t27_uniq;
+> SELECT key FROM (SHOW INDEXES ON t27_uniq);
+{id}
+
+> SELECT create_sql LIKE '%EXCLUDE CONSTRAINTS%t27_uq_wallet%' FROM (SHOW CREATE TABLE t27_uniq);
+true
+
+$ sql-server-execute name=sql-server
+ALTER TABLE t27_uniq DROP CONSTRAINT t27_uq_wallet;
+INSERT INTO t27_uniq VALUES (3, 'a');
+
+> SELECT * FROM t27_uniq;
+1 a
+2 b
+3 a
+
+#
+# EXCLUDE ALL CONSTRAINTS: no keys, every column nullable.
+#
+
+> CREATE TABLE t27_all FROM SOURCE t27_source (REFERENCE t27_all)
+  WITH (EXCLUDE ALL CONSTRAINTS);
+
+> CREATE DEFAULT INDEX ON t27_all;
+> SELECT key FROM (SHOW INDEXES ON t27_all);
+{id,email}
+
+> SELECT name, nullable FROM mz_columns WHERE id = (SELECT id FROM mz_tables WHERE name = 't27_all');
+id true
+email true
+
+> SELECT * FROM t27_all;
+1 a@b.c
+
+# Replication still works on both tables.
+$ sql-server-execute name=sql-server
+INSERT INTO t27_uniq VALUES (4, 'c');
+INSERT INTO t27_all VALUES (2, 'd@e.f');
+
+> SELECT * FROM t27_uniq;
+1 a
+2 b
+3 a
+4 c
+
+> SELECT * FROM t27_all;
+1 a@b.c
+2 d@e.f
+
+#
+# An empty exclusion list behaves as if the option were omitted.
+#
+
+> CREATE TABLE t27_uniq_empty FROM SOURCE t27_source (REFERENCE t27_uniq)
+  WITH (EXCLUDE CONSTRAINTS ());
+
+> CREATE DEFAULT INDEX ON t27_uniq_empty;
+> SELECT key FROM (SHOW INDEXES ON t27_uniq_empty);
+{id}
+
+> DROP TABLE t27_uniq_empty;
+
+#
+# A non-excluded UNIQUE constraint dropped upstream still stalls the table,
+# with an error that names the constraint and the recovery workflow.
+#
+
+> CREATE TABLE t27_stall FROM SOURCE t27_source (REFERENCE t27_stall);
+
+> SELECT * FROM t27_stall;
+1 a
+
+$ sql-server-execute name=sql-server
+ALTER TABLE t27_stall DROP CONSTRAINT t27_stall_uq;
+INSERT INTO t27_stall VALUES (2, 'a');
+
+! SELECT * FROM t27_stall;
+contains:incompatible schema change on dbo.t27_stall: UNIQUE constraint "t27_stall_uq" (wallet) was dropped upstream
+hint:WITH (EXCLUDE CONSTRAINTS ('t27_stall_uq')) before the upstream drop
+
+# Recovery: a replacement table snapshots the current upstream schema, which no
+# longer has the constraint, so no exclusion is needed. Then drop the errored
+# table.
+> CREATE TABLE t27_stall_v2 FROM SOURCE t27_source (REFERENCE t27_stall);
+
+> SELECT * FROM t27_stall_v2;
+1 a
+2 a
+
+> DROP TABLE t27_stall;
+
+#
+# Feature flag: the options are rejected when disabled.
+#
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET enable_exclude_constraints_option = false
+
+! CREATE TABLE t27_flagged FROM SOURCE t27_source (REFERENCE t27_uniq)
+  WITH (EXCLUDE ALL CONSTRAINTS);
+contains:not available
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET enable_exclude_constraints_option = true
+
+> DROP SOURCE t27_source CASCADE;

test/testdrive/exclude-constraints.td

@@ -8,9 +8,9 @@
 # by the Apache License, Version 2.0.
 
 # EXCLUDE CONSTRAINTS / EXCLUDE ALL CONSTRAINTS on `CREATE TABLE .. FROM
-# SOURCE` is only supported for Postgres sources; other source types reject
-# the options during purification. The Postgres behavior itself is covered in
-# test/pg-cdc/exclude-constraints.td.
+# SOURCE` is only supported for Postgres, MySQL, and SQL Server sources; other
+# source types reject the options during purification. The per-source behavior
+# is covered in test/pg-cdc/, test/mysql-cdc/, and test/sql-server-cdc/.
 
 $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
 ALTER SYSTEM SET enable_exclude_constraints_option = true