fix MaterializeInc/database-issues#11315

Author: patrickwwbutler

Cargo.lock

@@ -525,6 +525,7 @@ url = "2.5.8"
 urlencoding = "2.1.3"
 utoipa = "5.4.0"
 uuid = "1.19.0"
+version-compare = "0.2.1"
 walkdir = "2.5.0"
 which = "8"
 yansi = "1.0.1"

Cargo.toml

@@ -82,6 +82,7 @@ tracing.workspace = true
 tracing-subscriber.workspace = true
 uncased.workspace = true
 uuid = { workspace = true, features = ["serde", "v4"] }
+version-compare.workspace = true
 
 [dev-dependencies]
 datadriven.workspace = true

src/sql/Cargo.toml

@@ -68,6 +68,7 @@ use protobuf_native::compiler::{SourceTreeDescriptorDatabase, VirtualSourceTree}
 use rdkafka::admin::AdminClient;
 use references::{RetrievedSourceReferences, SourceReferenceClient};
 use uuid::Uuid;
+use version_compare;
 
 use crate::ast::{
     AlterSourceAddSubsourceOption, AvroSchema, CreateSourceConnection, CreateSourceStatement,
@@ -1858,9 +1859,22 @@ async fn purify_create_table_from_source(
                 )
                 .await?;
 
+            if version_compare::compare_to(
+                mz_mysql_util::query_sys_var(&mut conn, "version").await?,
+                "8.0.1",
+                version_compare::Cmp::Lt,
+            )
+            .expect("failed to parse version string from mysql")
+            {
+                Err(
+                    MySqlSourcePurificationError::UnsupportedBinlogMetadataSetting {
+                        setting: "Using MySQL version < 8.0.1, which does not support full binlog row metadata".to_string(),
+                    },
+                )?;
+            }
             let binlog_metadata_setting =
                 mz_mysql_util::query_sys_var(&mut conn, "binlog_row_metadata").await?;
-            if binlog_metadata_setting != "FULL" {
+            if binlog_metadata_setting.to_uppercase() != "FULL" {
                 Err(
                     MySqlSourcePurificationError::UnsupportedBinlogMetadataSetting {
                         setting: binlog_metadata_setting,

src/sql/src/pure.rs

@@ -91,6 +91,7 @@ tracing.workspace = true
 thiserror.workspace = true
 uuid = { workspace = true, features = ["serde", "v4"] }
 arrow-ipc.workspace = true
+version-compare.workspace = true
 
 [dev-dependencies]
 async-trait.workspace = true

src/storage/Cargo.toml

@@ -8,6 +8,7 @@
 // by the Apache License, Version 2.0.
 
 use std::collections::{BTreeMap, BTreeSet};
+use version_compare;
 
 use mysql_async::prelude::Queryable;
 use mz_mysql_util::{MySqlError, SchemaRequest, schema_info};
@@ -45,12 +46,21 @@ where
     })
     .collect();
 
-    let full_metadata = conn
-        .query_first::<String, String>("SELECT @@binlog_row_metadata".to_string())
-        .await?
-        .unwrap()
-        .to_uppercase()
-        == "FULL";
+    let full_metadata = version_compare::compare_to(
+        conn.query_first::<String, String>("SELECT VERSION()".to_string())
+            .await?
+            .unwrap()
+            .to_uppercase(),
+        "8.0.1",
+        version_compare::Cmp::Ge,
+    )
+    .expect("failed to parse version string from mysql")
+        && conn
+            .query_first::<String, String>("SELECT @@binlog_row_metadata".to_string())
+            .await?
+            .unwrap()
+            .to_uppercase()
+            == "FULL";
 
     Ok(expected
         .into_iter()