readyset-mysql: Infer MySQL implicit defaults

MySQL allows users to omit columns from an INSERT statement, if the
column is defined as NOT NULL and has no default value it will
error out. However, if the user disable STRICT_TRANS_TABLES, MySQL
will infer the default value for the column.

This is fine for snapshot and replication with full row image, as we
will receive the inferred default value. However, this is not the case
for minimal row-based replication, where we receive only the columns
that were specified in the INSERT statement and are responsible for
inferring the default based on the column type, which we were
defaulting to None.

If a table has not PK or UK, DELETE/UPDATE operations will require
a full match on all columns to find the row to operate on. This will
cause replication to fail, as we and MySQL will have different values.

This commit also adjusts the mysql-time crate to not display a
negative value for zero time.

Closes: REA-5555
Closes: #1482

Release-Note-Core: Fixed a MySQL issue where INSERT statements with
  omitted columns could cause replication to fail under certain
  conditions.
Change-Id: Id00f4d6dc9b67368fad5453489f1768c4095ce7f
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9140
Reviewed-by: Michael Zink <michael.z@readyset.io>
Tested-by: Buildkite CI

Author: altmannmarcelo

mysql-time/src/lib.rs

@@ -301,7 +301,11 @@ impl Default for MySqlTime {
 
 impl fmt::Display for MySqlTime {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let sign = if self.is_positive() { "" } else { "-" };
+        let sign = if self.is_positive() || self.nanos == 0 {
+            ""
+        } else {
+            "-"
+        };
         let h = self.hour();
         let m = self.minutes();
         let s = self.seconds();

readyset-data/src/lib.rs

@@ -877,6 +877,77 @@ impl DfValue {
             _ => self,
         }
     }
+
+    /// Returns an implicit default value for a column that is not null and has no default.
+    /// MySQL has [implicit defaults], and in case you disable sql_mode STRICT_TRANS_TABLES
+    /// it will let you either omit the columns that has no default and is defined as NOT NULL,
+    /// or allow you to set it to DEFAULT.
+    ///
+    /// [implicit defaults]: https://dev.mysql.com/doc/refman/8.4/en/data-type-defaults.html
+    ///
+    /// # Parameters
+    ///
+    /// * `collation`: The collation of the column.
+    /// * `dftype`: The type of the column.
+    /// * `dialect`: The dialect of the database.
+    ///
+    /// # Returns
+    ///
+    /// The implicit default value for the column.
+    pub fn implicit_default(collation: Collation, dftype: DfType, dialect: Dialect) -> Self {
+        match dialect {
+            Dialect::DEFAULT_MYSQL => {
+                // Yay, MySQL has implicit defaults if you omit a column that is not null and has no default
+                if matches!(dftype, DfType::Blob) {
+                    // TODO: Fix this to return ByteArray(Vec::new().into())
+                    return DfValue::from_str_and_collation("", collation);
+                } else if dftype.is_any_text() {
+                    return DfValue::from_str_and_collation("", collation);
+                } else if dftype.is_any_int() || dftype.is_bool() {
+                    return DfValue::Int(0);
+                } else if dftype.is_any_float() {
+                    return DfValue::Float(0.0);
+                } else if matches!(dftype, DfType::Numeric { .. }) {
+                    return DfValue::Numeric(Arc::new(Decimal::new(0, 2)));
+                } else if matches!(dftype, DfType::Bit { .. }) {
+                    return DfValue::from_str_and_collation("\0", Collation::Utf8);
+                } else if dftype.is_any_temporal() {
+                    match dftype {
+                        DfType::Timestamp { subsecond_digits }
+                        | DfType::DateTime { subsecond_digits } => {
+                            let mut dt = TimestampTz::zero();
+                            dt.set_subsecond_digits(subsecond_digits as u8);
+                            return DfValue::TimestampTz(dt);
+                        }
+                        DfType::Time { .. } => {
+                            return DfValue::Time(MySqlTime::default());
+                        }
+                        DfType::Date { .. } => {
+                            let mut dt = TimestampTz::zero();
+                            dt.set_date_only();
+                            return DfValue::TimestampTz(dt);
+                        }
+                        _ => {}
+                    }
+                } else if matches!(dftype, DfType::Enum { .. }) {
+                    if let DfType::Enum { variants, .. } = dftype {
+                        // Safety, I don't think this can happen
+                        if variants.is_empty() {
+                            return DfValue::from_str_and_collation("", collation);
+                        }
+                        return DfValue::from_str_and_collation(
+                            variants[0].as_str(),
+                            Collation::Utf8,
+                        );
+                    }
+                } else if matches!(dftype, DfType::Json { .. }) {
+                    return DfValue::from_str_and_collation("null", Collation::Utf8);
+                }
+                DfValue::None
+            }
+            Dialect::DEFAULT_POSTGRESQL => DfValue::None,
+        }
+    }
 }
 
 impl PartialEq for DfValue {

readyset-mysql/tests/integration.rs

@@ -2522,72 +2522,77 @@ async fn test_case_expr_then_expr() {
     shutdown_tx.shutdown().await;
 }
 
-async fn populate_all_data_types(direct_mysql: &mut mysql_async::Conn) {
+async fn populate_all_data_types(
+    direct_mysql: &mut mysql_async::Conn,
+    table_name: &str,
+    add_sample_data: bool,
+) {
     // Drop existing table if any
     direct_mysql
-        .query_drop("DROP TABLE IF EXISTS all_data_types CASCADE;")
+        .query_drop(format!("DROP TABLE IF EXISTS {} CASCADE;", table_name))
         .await
         .unwrap();
     // Create table
     direct_mysql
         .query_drop(
-            "CREATE TABLE all_data_types (
+            format!("CREATE TABLE {} (
         -- Numeric Data Types (Signed and Unsigned Integers)
-        col_tinyint TINYINT,                  -- Signed (-128 to 127)
-        col_tinyint_unsigned TINYINT UNSIGNED, -- Unsigned (0 to 255)
-
-        col_smallint SMALLINT,                 -- Signed (-32,768 to 32,767)
-        col_smallint_unsigned SMALLINT UNSIGNED, -- Unsigned (0 to 65,535)
-
-        col_mediumint MEDIUMINT,               -- Signed (-8,388,608 to 8,388,607)
-        col_mediumint_unsigned MEDIUMINT UNSIGNED, -- Unsigned (0 to 16,777,215)
-
-        col_int INT,                           -- Signed (-2,147,483,648 to 2,147,483,647)
-        col_int_unsigned INT UNSIGNED,         -- Unsigned (0 to 4,294,967,295)
-
-        col_bigint BIGINT,                     -- Signed (-2^63 to 2^63-1)
-        col_bigint_unsigned BIGINT UNSIGNED,   -- Unsigned (0 to 2^64-1)
-
-        col_decimal DECIMAL(12,2),             -- Fixed-point exact decimal (e.g., 99999999.99)
-        col_numeric NUMERIC(10,2),             -- Synonym for DECIMAL
-
-        col_float FLOAT,                       -- 32-bit floating point (approximate value)
-        col_double DOUBLE,                     -- 64-bit floating point (approximate value)
-        col_real REAL,                         -- Synonym for DOUBLE
-
-        col_bit BIT(8),                        -- Bit field (e.g., 8-bit binary)
-        col_boolean BOOLEAN,                   -- Alias for TINYINT(1) (0 = false, 1 = true)
-
+        col_tinyint TINYINT NOT NULL,                  -- Signed (-128 to 127)
+        col_tinyint_unsigned TINYINT UNSIGNED NOT NULL, -- Unsigned (0 to 255)
+        
+        col_smallint SMALLINT NOT NULL,                 -- Signed (-32,768 to 32,767)
+        col_smallint_unsigned SMALLINT UNSIGNED NOT NULL, -- Unsigned (0 to 65,535)
+        
+        col_mediumint MEDIUMINT NOT NULL,               -- Signed (-8,388,608 to 8,388,607)
+        col_mediumint_unsigned MEDIUMINT UNSIGNED NOT NULL, -- Unsigned (0 to 16,777,215)
+        
+        col_int INT NOT NULL,                           -- Signed (-2,147,483,648 to 2,147,483,647)
+        col_int_unsigned INT UNSIGNED NOT NULL,         -- Unsigned (0 to 4,294,967,295)
+        
+        col_bigint BIGINT NOT NULL,                     -- Signed (-2^63 to 2^63-1)
+        col_bigint_unsigned BIGINT UNSIGNED NOT NULL,   -- Unsigned (0 to 2^64-1)
+
+        col_decimal DECIMAL(12,2) NOT NULL,             -- Fixed-point exact decimal (e.g., 99999999.99)
+        col_numeric NUMERIC(10,2) NOT NULL,             -- Synonym for DECIMAL
+        
+        col_float FLOAT NOT NULL,                       -- 32-bit floating point (approximate value)
+        col_double DOUBLE NOT NULL,                     -- 64-bit floating point (approximate value)
+        col_real REAL NOT NULL,                         -- Synonym for DOUBLE
+        
+        col_bit BIT(8) NOT NULL,                        -- Bit field (e.g., 8-bit binary)
+        col_boolean BOOLEAN NOT NULL,                   -- Alias for TINYINT(1) (0 = false, 1 = true)
+        
         -- Date and Time Data Types
-        col_date DATE,
-        col_datetime DATETIME(2),
-        col_timestamp TIMESTAMP,
-        col_time TIME,
+        col_date DATE NOT NULL,
+        col_datetime DATETIME(2) NOT NULL,
+        col_timestamp TIMESTAMP NOT NULL,
+        col_time TIME NOT NULL,
 
         -- String Data Types
-        col_char CHAR(10),
-        col_varchar VARCHAR(255),
-
-        col_text TEXT,
-        col_tinytext TINYTEXT,
-        col_mediumtext MEDIUMTEXT,
-        col_longtext LONGTEXT,
-
-        col_blob BLOB,
-        col_tinyblob TINYBLOB,
-        col_mediumblob MEDIUMBLOB,
-        col_longblob LONGBLOB,
-
-        col_enum ENUM('small', 'medium', 'large'),
-        col_json JSON
-    );",
+        col_char CHAR(10) NOT NULL,
+        col_varchar VARCHAR(255) NOT NULL,
+
+        col_text TEXT NOT NULL,
+        col_tinytext TINYTEXT NOT NULL,
+        col_mediumtext MEDIUMTEXT NOT NULL,
+        col_longtext LONGTEXT NOT NULL,
+
+        col_blob BLOB NOT NULL,
+        col_tinyblob TINYBLOB NOT NULL,
+        col_mediumblob MEDIUMBLOB NOT NULL,
+        col_longblob LONGBLOB NOT NULL,
+
+        col_enum ENUM('small', 'medium', 'large') NOT NULL,
+        col_json JSON NOT NULL
+    );", table_name)
         )
         .await
         .unwrap();
-    // Insert test data
-    direct_mysql
-        .query_drop(
-            "INSERT INTO all_data_types (
+    if add_sample_data {
+        // Insert test data
+        direct_mysql
+            .query_drop(format!(
+                "INSERT INTO {} (
 col_tinyint, col_tinyint_unsigned,
 col_smallint, col_smallint_unsigned,
 col_mediumint, col_mediumint_unsigned,
@@ -2643,23 +2648,29 @@ CURRENT_TIMESTAMP,
 'long blob contents here',
 
 'medium',
-'{\"name\": \"Alice\", \"roles\": [\"admin\", \"editor\"], \"active\": true}'
+'{{\"name\": \"Alice\", \"roles\": [\"admin\", \"editor\"], \"active\": true}}'
 );",
-        )
-        .await
-        .unwrap();
+                table_name
+            ))
+            .await
+            .unwrap();
+    }
 }
 
 async fn test_column_definition_verify(
     direct_mysql: &mut mysql_async::Conn,
     rs_conn: &mut mysql_async::Conn,
+    table_name: &str,
 ) {
     // Verify results
     let direct_rows: Vec<Row> = direct_mysql
-        .query("SELECT * FROM all_data_types")
+        .query(format!("SELECT * FROM {}", table_name))
+        .await
+        .unwrap();
+    let rs_rows: Vec<Row> = rs_conn
+        .query(format!("SELECT * FROM {}", table_name))
         .await
         .unwrap();
-    let rs_rows: Vec<Row> = rs_conn.query("SELECT * FROM all_data_types").await.unwrap();
 
     let direct_columns = direct_rows[0].columns();
     let rs_columns = rs_rows[0].columns();
@@ -2711,7 +2722,7 @@ async fn test_column_definition_verify(
 async fn test_column_definition_upstream_readyset_snapshot() {
     readyset_tracing::init_test_logging();
     let mut direct_mysql = mysql_async::Conn::from_url(mysql_url()).await.unwrap();
-    populate_all_data_types(&mut direct_mysql).await;
+    populate_all_data_types(&mut direct_mysql, "all_data_types", true).await;
 
     // Setup ReadySet connection after table creation
     let (rs_opts, _rs_handle, tx) = TestBuilder::default()
@@ -2720,7 +2731,7 @@ async fn test_column_definition_upstream_readyset_snapshot() {
         .await;
     let mut conn = mysql_async::Conn::new(rs_opts).await.unwrap();
 
-    test_column_definition_verify(&mut direct_mysql, &mut conn).await;
+    test_column_definition_verify(&mut direct_mysql, &mut conn, "all_data_types").await;
     tx.shutdown().await;
 }
 
@@ -2737,8 +2748,8 @@ async fn test_column_definition_upstream_readyset_replication() {
         .await;
     let mut conn = mysql_async::Conn::new(rs_opts).await.unwrap();
 
-    populate_all_data_types(&mut direct_mysql).await;
-    test_column_definition_verify(&mut direct_mysql, &mut conn).await;
+    populate_all_data_types(&mut direct_mysql, "all_data_types", true).await;
+    test_column_definition_verify(&mut direct_mysql, &mut conn, "all_data_types").await;
     tx.shutdown().await;
 }
 
@@ -2790,3 +2801,51 @@ async fn text_citext_default_coercion_minimal_row_base_replication() {
     assert_eq!(rs_row[1], direct_row[1]);
     tx.shutdown().await;
 }
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial(mysql)]
+async fn test_default_value_not_null_for_replication() {
+    readyset_tracing::init_test_logging();
+    let (rs_opts, _rs_handle, tx) = TestBuilder::default()
+        .recreate_database(false)
+        .fallback(true)
+        .build::<MySQLAdapter>()
+        .await;
+    sleep().await;
+    let mut direct_mysql = mysql_async::Conn::from_url(mysql_url()).await.unwrap();
+    direct_mysql
+        .query_drop("SET SESSION binlog_row_image = MINIMAL; SET SESSION sql_mode = '';")
+        .await
+        .unwrap();
+    populate_all_data_types(&mut direct_mysql, "all_data_types_not_null", false).await;
+    direct_mysql
+        .query_drop("INSERT INTO all_data_types_not_null () VALUES ();")
+        .await
+        .unwrap();
+
+    let mut conn = mysql_async::Conn::new(rs_opts).await.unwrap();
+    conn.query_drop("CREATE CACHE FROM SELECT * FROM all_data_types_not_null")
+        .await
+        .unwrap();
+    let rs_row: Row = conn
+        .query_first("SELECT * FROM all_data_types_not_null")
+        .await
+        .unwrap()
+        .unwrap();
+    let direct_row: Row = direct_mysql
+        .query_first("SELECT * FROM all_data_types_not_null")
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(rs_row.len(), direct_row.len());
+    let row_length = rs_row.len();
+    for i in 0..row_length {
+        assert_eq!(
+            rs_row[i],
+            direct_row[i],
+            "Value mismatch for column: {}",
+            String::from_utf8_lossy(rs_row.columns()[i].name_ref())
+        );
+    }
+    tx.shutdown().await;
+}

readyset-server/src/controller/mir_to_flow.rs

@@ -368,8 +368,8 @@ fn make_base_node(
         .map(|cs| DfColumn::from_spec(cs.clone(), mig.dialect, |ty| custom_types.get(&ty).cloned()))
         .collect::<Result<Vec<_>, _>>()?;
 
-    // note that this defaults to a "None" (= NULL) default value for columns that do not have one
-    // specified; we don't currently handle a "NOT NULL" SQL constraint for defaults
+    // note that this defaults to a "None" (= NULL) default value for columns that don't have a default value
+    // or we cannot infer a default value from the SQL type
     let default_values = column_specs
         .iter()
         .map(|cs| {
@@ -387,6 +387,21 @@ fn make_base_node(
                                 internal_err!("Failed to convert SQL type to DfType: {}", e)
                             })?;
                     return df.coerce_to(&dftype_to, &df.infer_dataflow_type());
+                } else if let ColumnConstraint::NotNull { .. } = *c {
+                    let collation = cs.get_collation().map(|collation| {
+                        Collation::from_mysql_collation(collation).unwrap_or(Collation::Utf8)
+                    });
+                    let dftype_to =
+                        DfType::from_sql_type(&cs.sql_type, mig.dialect, |_| None, collation)
+                            .map_err(|e| {
+                                internal_err!("Failed to convert SQL type to DfType: {}", e)
+                            })?;
+
+                    return Ok(DfValue::implicit_default(
+                        collation.unwrap_or(Collation::Utf8),
+                        dftype_to,
+                        mig.dialect,
+                    ));
                 }
             }
             Ok(DfValue::None)