sql: support EXCLUDE ALL NOT NULL CONSTRAINTS for MySQL sources

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,

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::<Result<Vec<_>, PlanError>>()?;

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 <null>
+
+> SELECT * FROM all_nullable_combo;
+1 10 a
+2 20 <null>
+
+$ 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 <null>
+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 <null>
+3 30 c
+4 <null> d
+5 <null> e
+
+> SELECT u, COUNT(*) FROM all_nullable_combo GROUP BY u;
+10 1
+20 1
+30 1
+<null> 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;