diff --git a/src/sql/src/pure.rs b/src/sql/src/pure.rs index bd5e69bd85ee1..1f7f0848b70dc 100644 --- a/src/sql/src/pure.rs +++ b/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, diff --git a/src/sql/src/pure/error.rs b/src/sql/src/pure/error.rs index ec6182bca247b..0fde2866f2c0b 100644 --- a/src/sql/src/pure/error.rs +++ b/src/sql/src/pure/error.rs @@ -485,6 +485,11 @@ pub enum SqlServerSourcePurificationError { option_name: String, items: Vec, }, + #[error("EXCLUDE CONSTRAINTS refers to constraints that do not exist on table {table}")] + ConstraintsNotFound { + table: String, + constraints: Vec, + }, #[error("found multiple primary keys for a table. constraints {constraint_names:?}")] MultiplePrimaryKeys { constraint_names: Vec> }, #[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 diff --git a/src/sql/src/pure/sql_server.rs b/src/sql/src/pure/sql_server.rs index 1ad165a94e056..8e336ec9d2bcb 100644 --- a/src/sql/src/pure/sql_server.rs +++ b/src/sql/src/pure/sql_server.rs @@ -59,6 +59,12 @@ pub(super) async fn purify_source_exports( requested_references: &Option, 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, + 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)); } diff --git a/test/sql-server-cdc/27-exclude-constraints.td b/test/sql-server-cdc/27-exclude-constraints.td new file mode 100644 index 0000000000000..45c70457171ee --- /dev/null +++ b/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; diff --git a/test/testdrive/exclude-constraints.td b/test/testdrive/exclude-constraints.td index e12aa5d26e312..b1d1e4994f39f 100644 --- a/test/testdrive/exclude-constraints.td +++ b/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