diff --git a/crates/dbx-core/src/data_grid_sql.rs b/crates/dbx-core/src/data_grid_sql.rs
index c34fa2f40..3800e7ed5 100644
--- a/crates/dbx-core/src/data_grid_sql.rs
+++ b/crates/dbx-core/src/data_grid_sql.rs
@@ -864,16 +864,20 @@ fn validate_data_grid_save(options: &DataGridSaveStatementOptions) -> Option<Str
         }
     }
 
-    for row in &options.new_rows {
-        for column_index in 0..options.columns.len() {
-            let source_column = effective_column(options, column_index);
-            if is_null_write_to_not_null_column(
-                options.database_type,
-                &not_null_columns,
-                source_column,
-                row.get(column_index).unwrap_or(&Value::Null),
-            ) {
-                return Some(null_write_error(source_column.unwrap_or_default()));
+    // MySQL BEFORE INSERT triggers can populate omitted NOT NULL columns. New-row NULL values are
+    // omitted from the generated INSERT, so let MySQL apply triggers or report missing required fields.
+    if options.database_type != Some(DatabaseType::Mysql) {
+        for row in &options.new_rows {
+            for column_index in 0..options.columns.len() {
+                let source_column = effective_column(options, column_index);
+                if is_null_write_to_not_null_column(
+                    options.database_type,
+                    &not_null_columns,
+                    source_column,
+                    row.get(column_index).unwrap_or(&Value::Null),
+                ) {
+                    return Some(null_write_error(source_column.unwrap_or_default()));
+                }
             }
         }
     }
@@ -1073,6 +1077,10 @@ fn build_data_grid_save_statements(options: &DataGridSaveStatementOptions) -> Ve
             .filter(|(_, value)| !value.is_null())
             .collect();
         if insert_pairs.is_empty() {
+            if options.database_type == Some(DatabaseType::Mysql) {
+                statements
+                    .push(data_grid_statement(options.database_type, format!("INSERT INTO {table} () VALUES ()")));
+            }
             continue;
         }
         let columns = insert_pairs
@@ -1117,8 +1125,13 @@ fn build_data_grid_rollback_statements(options: &DataGridSaveStatementOptions) -
     let mut statements = Vec::new();
 
     for row in &options.new_rows {
-        let where_clause = build_save_row_where(options.database_type, &save_columns, row, column_info);
-        if !where_clause.is_empty() {
+        let where_clause = if options.database_type == Some(DatabaseType::Mysql) {
+            build_mysql_insert_rollback_where(options, &save_columns, row, column_info)
+        } else {
+            let where_clause = build_save_row_where(options.database_type, &save_columns, row, column_info);
+            (!where_clause.is_empty()).then_some(where_clause)
+        };
+        if let Some(where_clause) = where_clause {
             statements
                 .push(data_grid_statement(options.database_type, format!("DELETE FROM {table} WHERE {where_clause}")));
         }
@@ -1229,6 +1242,34 @@ fn build_data_grid_rollback_statements(options: &DataGridSaveStatementOptions) -
     statements
 }
 
+fn build_mysql_insert_rollback_where(
+    options: &DataGridSaveStatementOptions,
+    columns: &[Option<String>],
+    row: &[Value],
+    column_info: &[DataGridColumnInfo],
+) -> Option<String> {
+    if options.table_meta.primary_keys.is_empty() {
+        return None;
+    }
+
+    for primary_key in &options.table_meta.primary_keys {
+        let index = columns.iter().position(|column| column.as_deref() == Some(primary_key.as_str()))?;
+        let value = row.get(index).unwrap_or(&Value::Null);
+        let info = column_info_for(column_info, primary_key);
+        if value.is_null()
+            || empty_string_saves_as_null(value, info)
+            || info.is_some_and(is_auto_generated_column)
+            || info.is_some_and(|column| is_non_identity_generated_column(Some(column)))
+        {
+            // Generated or trigger-populated keys are unknown until after INSERT.
+            // Do not emit a rollback predicate that cannot match the inserted row.
+            return None;
+        }
+    }
+
+    Some(build_primary_key_where(options.database_type, &options.table_meta.primary_keys, columns, row, column_info))
+}
+
 pub(crate) fn effective_columns(options: &DataGridSaveStatementOptions) -> Vec<Option<String>> {
     match &options.source_columns {
         Some(source_columns) if source_columns.len() == options.columns.len() => source_columns.clone(),
@@ -4668,6 +4709,119 @@ mod tests {
         assert_eq!(result.statements, vec!["INSERT INTO `app`.`users` (`name`) VALUES ('Ada');"]);
     }
 
+    #[test]
+    fn prepare_data_grid_save_omits_mysql_not_null_column_for_before_insert_trigger() {
+        let result = prepare_data_grid_save(DataGridSaveStatementOptions {
+            database_type: Some(DatabaseType::Mysql),
+            table_meta: DataGridTableMeta {
+                catalog: None,
+                database: None,
+                schema: Some("app".to_string()),
+                table_name: "events".to_string(),
+                primary_keys: vec!["id".to_string()],
+                columns: Some(vec![
+                    pk_column("id", "BIGINT", false, Some("auto_increment")),
+                    column("trigger_value", "VARCHAR(64)", false, None),
+                    column("payload", "VARCHAR(64)", false, None),
+                ]),
+            },
+            columns: vec!["id".to_string(), "trigger_value".to_string(), "payload".to_string()],
+            source_columns: None,
+            rows: vec![],
+            dirty_rows: vec![],
+            deleted_rows: vec![],
+            new_rows: vec![vec![Value::Null, Value::Null, json!("created")]],
+        });
+
+        assert_eq!(result.validation_error, None);
+        assert_eq!(result.statements, vec!["INSERT INTO `app`.`events` (`payload`) VALUES ('created');"]);
+        assert!(result.rollback_statements.is_empty());
+    }
+
+    #[test]
+    fn prepare_data_grid_save_uses_mysql_default_row_insert_for_trigger_only_rows() {
+        let result = prepare_data_grid_save(DataGridSaveStatementOptions {
+            database_type: Some(DatabaseType::Mysql),
+            table_meta: DataGridTableMeta {
+                catalog: None,
+                database: None,
+                schema: Some("app".to_string()),
+                table_name: "trigger_only".to_string(),
+                primary_keys: vec!["id".to_string()],
+                columns: Some(vec![
+                    pk_column("id", "BIGINT", false, Some("auto_increment")),
+                    column("required_value", "VARCHAR(64)", false, None),
+                ]),
+            },
+            columns: vec!["id".to_string(), "required_value".to_string()],
+            source_columns: None,
+            rows: vec![],
+            dirty_rows: vec![],
+            deleted_rows: vec![],
+            new_rows: vec![vec![Value::Null, Value::Null]],
+        });
+
+        assert_eq!(result.validation_error, None);
+        assert_eq!(result.statements, vec!["INSERT INTO `app`.`trigger_only` () VALUES ();"]);
+        assert!(result.rollback_statements.is_empty());
+    }
+
+    #[test]
+    fn prepare_data_grid_save_uses_known_mysql_primary_key_for_trigger_rollback() {
+        let result = prepare_data_grid_save(DataGridSaveStatementOptions {
+            database_type: Some(DatabaseType::Mysql),
+            table_meta: DataGridTableMeta {
+                catalog: None,
+                database: None,
+                schema: Some("app".to_string()),
+                table_name: "events".to_string(),
+                primary_keys: vec!["id".to_string()],
+                columns: Some(vec![
+                    pk_column("id", "BIGINT", false, None),
+                    column("trigger_value", "VARCHAR(64)", false, None),
+                    column("payload", "VARCHAR(64)", false, None),
+                ]),
+            },
+            columns: vec!["id".to_string(), "trigger_value".to_string(), "payload".to_string()],
+            source_columns: None,
+            rows: vec![],
+            dirty_rows: vec![],
+            deleted_rows: vec![],
+            new_rows: vec![vec![json!(7), Value::Null, json!("created")]],
+        });
+
+        assert_eq!(result.validation_error, None);
+        assert_eq!(result.statements, vec!["INSERT INTO `app`.`events` (`id`, `payload`) VALUES (7, 'created');"]);
+        assert_eq!(result.rollback_statements, vec!["DELETE FROM `app`.`events` WHERE `id` = 7;"]);
+    }
+
+    #[test]
+    fn prepare_data_grid_save_still_rejects_mysql_null_update() {
+        let result = prepare_data_grid_save(DataGridSaveStatementOptions {
+            database_type: Some(DatabaseType::Mysql),
+            table_meta: DataGridTableMeta {
+                catalog: None,
+                database: None,
+                schema: Some("app".to_string()),
+                table_name: "events".to_string(),
+                primary_keys: vec!["id".to_string()],
+                columns: Some(vec![
+                    pk_column("id", "BIGINT", false, Some("auto_increment")),
+                    column("trigger_value", "VARCHAR(64)", false, None),
+                ]),
+            },
+            columns: vec!["id".to_string(), "trigger_value".to_string()],
+            source_columns: None,
+            rows: vec![vec![json!(1), json!("existing")]],
+            dirty_rows: vec![(0, vec![(1, Value::Null)])],
+            deleted_rows: vec![],
+            new_rows: vec![],
+        });
+
+        assert_eq!(result.validation_error, Some(r#"Column "trigger_value" does not allow NULL."#.to_string()));
+        assert!(result.statements.is_empty());
+    }
+
     #[test]
     fn prepare_data_grid_save_omits_empty_kingbase_sqlserver_identity_value() {
         let result = prepare_data_grid_save(DataGridSaveStatementOptions {