diff --git a/src/sql/src/pure.rs b/src/sql/src/pure.rs index d48aea78485ec..6f320fba6c9f5 100644 --- a/src/sql/src/pure.rs +++ b/src/sql/src/pure.rs @@ -1229,6 +1229,7 @@ async fn purify_create_source( text_columns, exclude_columns, &FilterConstraints::none(), + false, source_name, initial_gtid_set.clone(), &reference_policy, @@ -1674,6 +1675,7 @@ async fn purify_alter_source_add_subsources( text_columns, exclude_columns, &FilterConstraints::none(), + false, &unresolved_source_name, initial_gtid_set, &SourceReferencePolicy::Required, @@ -1967,7 +1969,7 @@ async fn purify_create_table_from_source( }]) }); - if filter_constraints.excludes_any() + if (filter_constraints.excludes_any() || exclude_all_not_null_constraints) && !matches!( desc.connection, GenericSourceConnection::Postgres(_) | GenericSourceConnection::MySql(_) @@ -2076,6 +2078,7 @@ async fn purify_create_table_from_source( qualified_text_columns, qualified_exclude_columns, &filter_constraints, + exclude_all_not_null_constraints, &unresolved_source_name, initial_gtid_set, &SourceReferencePolicy::Required, diff --git a/src/sql/src/pure/mysql.rs b/src/sql/src/pure/mysql.rs index 5dcc0b076250b..f19f7b42cc6af 100644 --- a/src/sql/src/pure/mysql.rs +++ b/src/sql/src/pure/mysql.rs @@ -343,6 +343,7 @@ pub(super) async fn purify_source_exports( // purified, as `CREATE TABLE .. FROM SOURCE` does, since constraint names // are validated against that single table's keys. Enforced below. filter_constraints: &FilterConstraints, + exclude_all_not_null_constraints: bool, unresolved_source_name: &UnresolvedItemName, initial_gtid_set: String, reference_policy: &SourceReferencePolicy, @@ -471,6 +472,31 @@ pub(super) async fn purify_source_exports( } } } + if exclude_all_not_null_constraints { + // Columns of a retained primary key stay NOT NULL. The planner + // declares them NOT NULL in the table's `RelationDesc` + // regardless of this desc, so a nullable column here would + // drop the decoder's NULL check and let a NULL reach persist + // in a column it treats as non-nullable, which panics. + // The reason we need this to avoid panics for MySQL sources + // and not for Postgres sources is because schemas are compared + // by name/column in MySQL. Postgres on the other hand has an OID + // that would differ and prevent this case. + let pk_cols: BTreeSet<&str> = parsed_table + .keys + .iter() + .filter(|k| k.is_primary) + .flat_map(|k| k.columns.iter().map(String::as_str)) + .collect(); + for c in &mut parsed_table.columns { + if pk_cols.contains(c.name.as_str()) { + continue; + } + if let Some(column_type) = &mut c.column_type { + column_type.nullable = true; + } + } + } Ok(requested_export.change_meta(parsed_table)) }) .collect::, PlanError>>()?; diff --git a/test/mysql-cdc/exclude-constraints.td b/test/mysql-cdc/exclude-constraints.td index 77c477ffef0bd..ab7f2612e3563 100644 --- a/test/mysql-cdc/exclude-constraints.td +++ b/test/mysql-cdc/exclude-constraints.td @@ -49,6 +49,9 @@ INSERT INTO stall_tbl VALUES (1, 'a'); CREATE TABLE excl_combo (id INT PRIMARY KEY, u INT NOT NULL, val VARCHAR(64) NOT NULL, UNIQUE KEY uq_u (u)); INSERT INTO excl_combo VALUES (1, 10, 'a'); +CREATE TABLE excl_not_null (id INT PRIMARY KEY, u INT NOT NULL, val VARCHAR(64) NOT NULL, UNIQUE KEY unq_u (u)); +INSERT INTO excl_not_null VALUES (1, 10, 'a'); + > CREATE SOURCE mz_source FROM MYSQL CONNECTION mysqc; # @@ -215,3 +218,136 @@ contains:not available $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} ALTER SYSTEM SET enable_exclude_constraints_option = true + +# +# EXCLUDE ALL NOT NULL CONSTRAINTS: keys and uniqueness remain, but every column nullable, so +# DROP NOT NULL is a non-event. +# + +> CREATE TABLE all_nullable + FROM SOURCE mz_source (REFERENCE public.excl_not_null) + WITH (EXCLUDE ALL NOT NULL CONSTRAINTS); + +> SELECT create_sql LIKE '%EXCLUDE ALL NOT NULL CONSTRAINTS%' FROM (SHOW CREATE TABLE all_nullable); +true + +> CREATE DEFAULT INDEX ON all_nullable; +> SELECT key FROM (SHOW INDEXES ON all_nullable); +{id} + +# All columns except id should be nullable, but id is a PK and cannot be NULL +> SELECT name, nullable FROM mz_columns WHERE id = (SELECT id FROM mz_tables WHERE name = 'all_nullable'); +id false +u true +val true + +> CREATE TABLE all_nullable_combo + FROM SOURCE mz_source (REFERENCE public.excl_not_null) + WITH (EXCLUDE ALL NOT NULL CONSTRAINTS, EXCLUDE CONSTRAINTS ('PRIMARY')); + +> CREATE DEFAULT INDEX ON all_nullable_combo; +> SELECT key FROM (SHOW INDEXES ON all_nullable_combo); +{id,u,val} + +> SELECT name, nullable FROM mz_columns WHERE id = (SELECT id FROM mz_tables WHERE name = 'all_nullable_combo'); +id true +u true +val true + +$ mysql-execute name=mysql +USE public; +ALTER TABLE excl_not_null MODIFY val VARCHAR(64) NULL; +INSERT INTO excl_not_null VALUES (2, 20, NULL); + +> SELECT * FROM all_nullable; +1 10 a +2 20 + +> SELECT * FROM all_nullable_combo; +1 10 a +2 20 + +$ mysql-execute name=mysql +USE public; +ALTER TABLE excl_not_null DROP PRIMARY KEY; +INSERT INTO excl_not_null VALUES (3, 30, 'c'); + +! SELECT * FROM all_nullable; +contains:incompatible schema change on public.excl_not_null: PRIMARY KEY constraint "PRIMARY" (id) was dropped upstream +hint:WITH (EXCLUDE CONSTRAINTS ('PRIMARY')) before the upstream drop + +> DROP TABLE all_nullable; + +> SELECT * FROM all_nullable_combo; +1 10 a +2 20 +3 30 c + +$ mysql-execute name=mysql +USE public; +ALTER TABLE excl_not_null MODIFY u INT NULL; +INSERT INTO excl_not_null VALUES (4, NULL, 'd'); +INSERT INTO excl_not_null VALUES (5, NULL, 'e'); + +> SELECT * FROM all_nullable_combo; +1 10 a +2 20 +3 30 c +4 d +5 e + +> SELECT u, COUNT(*) FROM all_nullable_combo GROUP BY u; +10 1 +20 1 +30 1 + 2 + +# The unique constraint still needs to be explicitly excluded if you want to drop the constraint. +$ mysql-execute name=mysql +USE public; +ALTER TABLE excl_not_null DROP INDEX unq_u; +INSERT INTO excl_not_null VALUES (6, 60, 'f'); + +! SELECT * FROM all_nullable_combo; +contains:incompatible schema change on public.excl_not_null: UNIQUE constraint "unq_u" (u) was dropped upstream +hint:WITH (EXCLUDE CONSTRAINTS ('unq_u')) before the upstream drop + +> DROP TABLE all_nullable_combo; + +# +# A retained primary key keeps its columns NOT NULL. A NULL that slips past +# schema verification, because the key was dropped and restored while the +# source had no replica, is a decode error that the later UPDATE retracts, not +# a NULL stored in a non-nullable column. +# + +$ mysql-execute name=mysql +USE public; +CREATE TABLE pk_gap (id INT PRIMARY KEY, v INT NOT NULL); +INSERT INTO pk_gap VALUES (1, 10); + +> CREATE CLUSTER pk_gap_cluster SIZE '${arg.default-replica-size}'; +> CREATE SOURCE pk_gap_source IN CLUSTER pk_gap_cluster FROM MYSQL CONNECTION mysqc; +> CREATE TABLE pk_gap FROM SOURCE pk_gap_source (REFERENCE public.pk_gap) + WITH (EXCLUDE ALL NOT NULL CONSTRAINTS); + +> SELECT * FROM pk_gap; +1 10 + +> ALTER CLUSTER pk_gap_cluster SET (REPLICATION FACTOR 0) + +$ mysql-execute name=mysql +USE public; +ALTER TABLE pk_gap DROP PRIMARY KEY, MODIFY id INT NULL; +INSERT INTO pk_gap VALUES (NULL, 20); +UPDATE pk_gap SET id = 2 WHERE id IS NULL; +ALTER TABLE pk_gap MODIFY id INT NOT NULL, ADD PRIMARY KEY (id); + +> ALTER CLUSTER pk_gap_cluster SET (REPLICATION FACTOR 1) + +> SELECT * FROM pk_gap; +1 10 +2 20 + +> DROP SOURCE pk_gap_source CASCADE; +> DROP CLUSTER pk_gap_cluster;