readyset-data: Binary Column inferred default value

This change adds support for binary columns in the inferred default
value.

Closes: REA-5706
Closes: #1504

Release-Note-Core: Added support for inferring MySQL default values
  for binary columns.
Change-Id: I6f134c9da2d33039d90ccc1bf14eff09bd9f0c3c
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9290
Tested-by: Buildkite CI
Reviewed-by: Michael Zink <michael.z@readyset.io>

Author: altmannmarcelo

readyset-data/src/lib.rs

@@ -1000,6 +1000,15 @@ impl DfValue {
                     }
                 } else if matches!(dftype, DfType::Json) {
                     return DfValue::from_str_and_collation("null", Collation::Utf8);
+                } else if dftype.is_binary() {
+                    match dftype {
+                        DfType::Binary(len) => {
+                            return DfValue::ByteArray(vec![0; len as usize].into());
+                        }
+                        _ => {
+                            return DfValue::ByteArray(vec![0; 0].into());
+                        }
+                    }
                 }
                 DfValue::None
             }

replicators/tests/tests.rs

@@ -4199,3 +4199,57 @@ async fn mysql_minimal_row_based_char_padding() {
 
     shutdown_tx.shutdown().await;
 }
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial(mysql)]
+#[slow]
+async fn mysql_minimal_row_based_binary() {
+    readyset_tracing::init_test_logging();
+    let url = mysql_url();
+    let mut client = DbConnection::connect(&url).await.unwrap();
+
+    client
+        .query("SET binlog_row_image = minimal;")
+        .await
+        .unwrap();
+    client.query("SET sql_mode = '';").await.unwrap();
+    client
+        .query("DROP TABLE IF EXISTS mrbr_binary;")
+        .await
+        .unwrap();
+    client
+        .query(
+            "CREATE TABLE mrbr_binary (id INT, b BINARY(10) NOT NULL, vb VARBINARY(10) NOT NULL);",
+        )
+        .await
+        .unwrap();
+
+    let (mut ctx, shutdown_tx) = TestHandle::start_noria(url.to_string(), None)
+        .await
+        .unwrap();
+
+    ctx.controller_rx
+        .as_mut()
+        .unwrap()
+        .snapshot_completed()
+        .await
+        .unwrap();
+
+    client
+        .query("INSERT INTO mrbr_binary (id) VALUES (0);")
+        .await
+        .unwrap();
+
+    check_results!(
+        ctx,
+        "mrbr_binary",
+        "mrbr_binary_insert",
+        &[&[
+            DfValue::Int(0),
+            DfValue::ByteArray(vec![0; 10].into()),
+            DfValue::ByteArray(vec![0; 0].into())
+        ]]
+    );
+
+    shutdown_tx.shutdown().await;
+}