fix(transfer): support target table rebuild on mysql 5.7

crates/dbx-core/src/data/transfer_rebuild.rs

@@ -10,7 +10,7 @@
 
 use sha2::{Digest, Sha256};
 use sqlparser::ast::{visit_relations, ObjectNamePart, Statement, TableFactor, Visit, Visitor};
-use sqlparser::dialect::{DuckDbDialect, SQLiteDialect};
+use sqlparser::dialect::{DuckDbDialect, MySqlDialect, SQLiteDialect};
 use sqlparser::parser::Parser;
 use std::collections::BTreeSet;
 use std::ops::ControlFlow;
@@ -425,9 +425,27 @@ async fn detect_dependent_views(
     let sql = dependent_views_sql(database_type, database, schema, target_tables).ok_or_else(|| {
         format!("Cannot safely inspect dependent views for {} before rebuilding target tables", database_type.as_str())
     })?;
-    let result = read_dependency_metadata(state, pool_key, &sql, 4)
-        .await
-        .map_err(|error| format!("Failed to inspect dependent views before rebuilding target tables: {error}"))?;
+    let result = match read_dependency_metadata(state, pool_key, &sql, 4).await {
+        Ok(result) => result,
+        // `information_schema.VIEW_TABLE_USAGE` arrived with MySQL 8.0. Every MySQL 5.7
+        // target fails this lookup (ER_UNKNOWN_TABLE 1109/42S02), which used to abort the
+        // whole rebuild even when the target held no views at all. Inspect the stored view
+        // definitions instead: still fail-closed, just without the 8.0-only catalog.
+        Err(error)
+            if matches!(database_type, DatabaseType::Mysql | DatabaseType::Goldendb)
+                && mysql_view_usage_catalog_missing(&error) =>
+        {
+            log::info!(
+                "MySQL-compatible target has no information_schema.VIEW_TABLE_USAGE; \
+                 inspecting stored view definitions instead"
+            );
+            return detect_parsed_view_dependencies(state, pool_key, database, schema, target_tables, database_type)
+                .await;
+        }
+        Err(error) => {
+            return Err(format!("Failed to inspect dependent views before rebuilding target tables: {error}"))
+        }
+    };
     result
         .rows
         .iter()
@@ -505,6 +523,31 @@ fn dependent_views_sql(
     }
 }
 
+/// `information_schema.VIEW_TABLE_USAGE` exists from MySQL 8.0 on only. Older
+/// MySQL-compatible servers answer the lookup with ER_UNKNOWN_TABLE: MySQL 5.7 reports
+/// `ERROR 1109 (42S02): Unknown table 'VIEW_TABLE_USAGE' in information_schema`, and a
+/// server with `lower_case_table_names=1` (the Windows default) echoes the name
+/// lowercased. Match the table name so an unrelated `42S02` keeps reporting as an error.
+fn mysql_view_usage_catalog_missing(error: &str) -> bool {
+    let lower = error.to_ascii_lowercase();
+    lower.contains("view_table_usage")
+        && (lower.contains("1109")
+            || lower.contains("1146")
+            || lower.contains("42s02")
+            || lower.contains("unknown table"))
+}
+
+/// Stored definitions for the fallback path. Mirrors [`dependent_views_sql`]'s MySQL
+/// branch: only views of the transferred database can be broken by rebuilding one of its
+/// tables, so the schema filter keeps the parse surface identical.
+fn mysql_view_definition_dependencies_sql(database: &str) -> String {
+    format!(
+        "SELECT '', TABLE_SCHEMA, TABLE_NAME, VIEW_DEFINITION FROM information_schema.VIEWS \
+         WHERE {database_match}",
+        database_match = mysql_metadata_name_matches("TABLE_SCHEMA", &[database]),
+    )
+}
+
 async fn detect_parsed_view_dependencies(
     state: &AppState,
     pool_key: &str,
@@ -520,6 +563,13 @@ async fn detect_parsed_view_dependencies(
         let sql =
             "SELECT database_name, schema_name, view_name, sql FROM duckdb_views() WHERE NOT internal".to_string();
         (database, schema, sql)
+    } else if matches!(database_type, DatabaseType::Mysql | DatabaseType::Goldendb) {
+        // `VIEW_DEFINITION` is the resolved SELECT, and a MySQL database *is* the schema
+        // DBX rebuilds in, so the row's `TABLE_SCHEMA` decides which namespace an
+        // unqualified relation belongs to. The 8.0 catalog path keys on the database, so
+        // the fallback keeps the same spelling: dropping the database here would leave
+        // qualified references unmatched whenever the caller passes an empty schema.
+        (database.to_string(), schema.to_string(), mysql_view_definition_dependencies_sql(database))
     } else {
         let schema = resolve_sqlite_dependency_schema(state, pool_key, database, schema, database_type).await?;
         let mut sql = format!(
@@ -547,9 +597,12 @@ async fn detect_parsed_view_dependencies(
         };
         let ddl = dependency_metadata_text(row, 3)
             .map_err(|error| format!("Cannot safely inspect dependent view {qualified}: {error}"))?;
-        let references =
+        let references = if matches!(database_type, DatabaseType::Mysql | DatabaseType::Goldendb) {
+            mysql_view_definition_target_references(ddl, &database, &schema, view_schema, target_tables)
+        } else {
             parsed_view_target_references(ddl, database_type, &database, &schema, view_schema, target_tables)
-                .map_err(|error| format!("Cannot safely inspect dependent view {qualified}: {error}"))?;
+        }
+        .map_err(|error| format!("Cannot safely inspect dependent view {qualified}: {error}"))?;
         for referenced in references {
             blocking.insert(format!("view {qualified} -> {schema}.{referenced}"));
         }
@@ -571,11 +624,41 @@ fn parsed_view_target_references(
     let [Statement::CreateView(view)] = statements.as_slice() else {
         return Err("the catalog did not return a complete CREATE VIEW definition".to_string());
     };
-    if let ControlFlow::Break(error) = view.query.visit(&mut StaticViewRelations) {
+    view_query_target_references(&view.query, database_type, database, schema, view_schema, target_tables)
+}
+
+/// MySQL stores only the resolved `SELECT` in `information_schema.VIEWS.VIEW_DEFINITION`
+/// (the `CREATE VIEW` wrapper and any `WITH CHECK OPTION` live outside that column), so
+/// the fallback parses the statement body instead of a complete DDL.
+fn mysql_view_definition_target_references(
+    definition: &str,
+    database: &str,
+    schema: &str,
+    view_schema: &str,
+    target_tables: &[String],
+) -> Result<BTreeSet<String>, String> {
+    let dialect = MySqlDialect {};
+    let statements = Parser::parse_sql(&dialect, definition).map_err(|error| error.to_string())?;
+    let [Statement::Query(query)] = statements.as_slice() else {
+        return Err("the catalog did not return a complete view definition".to_string());
+    };
+    view_query_target_references(query, DatabaseType::Mysql, database, schema, view_schema, target_tables)
+}
+
+/// Shared relation walk for a parsed view body.
+fn view_query_target_references(
+    query: &sqlparser::ast::Query,
+    database_type: DatabaseType,
+    database: &str,
+    schema: &str,
+    view_schema: &str,
+    target_tables: &[String],
+) -> Result<BTreeSet<String>, String> {
+    if let ControlFlow::Break(error) = query.visit(&mut StaticViewRelations) {
         return Err(error.to_string());
     }
     let mut references = BTreeSet::new();
-    let visited = visit_relations(&view.query, |relation| {
+    let visited = visit_relations(query, |relation| {
         let names = relation
             .0
             .iter()
@@ -603,6 +686,15 @@ fn parsed_view_target_references(
                 }
                 _ => return ControlFlow::Break("a DuckDB view relation has an unsupported qualified name".to_string()),
             }
+        } else if matches!(database_type, DatabaseType::Mysql | DatabaseType::Goldendb) {
+            // MySQL stores relations with their database name. DBX's `schema` and
+            // `database` are the same namespace for MySQL, so either spelling of the
+            // target proves an unqualified or single-qualifier reference points at it.
+            match qualifiers {
+                [] => view_schema.eq_ignore_ascii_case(schema) || view_schema.eq_ignore_ascii_case(database),
+                [qualified] => qualified.eq_ignore_ascii_case(schema) || qualified.eq_ignore_ascii_case(database),
+                _ => return ControlFlow::Break("a MySQL view relation has an unsupported qualified name".to_string()),
+            }
         } else {
             match qualifiers {
                 [] => view_schema.eq_ignore_ascii_case(schema) || view_schema.eq_ignore_ascii_case("temp"),
@@ -1272,6 +1364,107 @@ mod tests {
         assert!(matches!(statements.first(), Some(Statement::Query(_))));
     }
 
+    #[test]
+    fn mysql_view_usage_catalog_absence_is_recognized_without_hiding_other_failures() {
+        for message in [
+            "Server error: `ERROR 1109 (42S02): Unknown table 'VIEW_TABLE_USAGE' in information_schema`",
+            "Server error: `ERROR 1109 (42S02): Unknown table 'view_table_usage' in information_schema`",
+            "Server error: `ERROR 1146 (42S02): Table 'information_schema.view_table_usage' doesn't exist`",
+        ] {
+            assert!(mysql_view_usage_catalog_missing(message), "{message}");
+        }
+        for message in [
+            "Server error: `ERROR 1146 (42S02): Table 'information_schema.VIEWS' doesn't exist`",
+            "Server error: `ERROR 1045 (28000): Access denied for user 'root'@'localhost' (using password: YES)`",
+            "Server error: `ERROR 1142 (42000): SELECT command denied to user 'dbx'@'%' for table 'VIEW_TABLE_USAGE'`",
+            "connection reset by peer",
+        ] {
+            assert!(!mysql_view_usage_catalog_missing(message), "{message}");
+        }
+    }
+
+    #[test]
+    fn mysql_view_definitions_are_parsed_for_target_references() {
+        // Captured from MySQL 5.7.43: `information_schema.VIEWS.VIEW_DEFINITION` names the
+        // resolved database on every relation and expands `SELECT *` into explicit columns.
+        let targets = vec!["orders".to_string(), "customers".to_string()];
+        let qualified = mysql_view_definition_target_references(
+            "select `shop`.`orders`.`id` AS `id`,`shop`.`orders`.`total` AS `total` \
+             from `shop`.`orders` where (`shop`.`orders`.`id` > 0)",
+            "shop",
+            "shop",
+            "shop",
+            &targets,
+        )
+        .unwrap();
+        assert_eq!(qualified.into_iter().collect::<Vec<_>>(), vec!["orders".to_string()]);
+
+        // A bare API call can reach the MySQL fallback with an empty target schema; the
+        // qualified reference must then match on the database spelling alone, exactly as
+        // the 8.0 catalog path filters `TABLE_SCHEMA` by database.
+        let empty_schema =
+            mysql_view_definition_target_references("select `id` from `shop`.`orders`", "shop", "", "shop", &targets)
+                .unwrap();
+        assert_eq!(empty_schema.into_iter().collect::<Vec<_>>(), vec!["orders".to_string()]);
+
+        let unqualified = mysql_view_definition_target_references(
+            "select `id` from `shop`.`orders` join `customers` on 1 = 1",
+            "shop",
+            "shop",
+            "shop",
+            &targets,
+        )
+        .unwrap();
+        assert_eq!(unqualified.into_iter().collect::<Vec<_>>(), vec!["customers".to_string(), "orders".to_string()]);
+
+        let other_database = mysql_view_definition_target_references(
+            "select `id` from `archive`.`orders`",
+            "shop",
+            "shop",
+            "shop",
+            &targets,
+        )
+        .unwrap();
+        assert!(
+            other_database.is_empty(),
+            "a view of the transferred database cannot be broken by another database's table"
+        );
+
+        let unrelated = mysql_view_definition_target_references(
+            "select `id` from `orders_summary`",
+            "shop",
+            "shop",
+            "shop",
+            &targets,
+        )
+        .unwrap();
+        assert!(unrelated.is_empty());
+
+        // MySQL has no three-part relation name, so the reference cannot be resolved and
+        // the check must fail closed instead of reporting "no dependency".
+        assert!(mysql_view_definition_target_references(
+            "select `id` from `shop`.`extra`.`orders`",
+            "shop",
+            "shop",
+            "shop",
+            &targets,
+        )
+        .is_err());
+    }
+
+    #[test]
+    fn mysql_view_definition_fallback_query_targets_the_transferred_database() {
+        let sql = mysql_view_definition_dependencies_sql("shop");
+        assert!(sql.contains("FROM information_schema.VIEWS"), "{sql}");
+        assert!(!sql.contains("VIEW_TABLE_USAGE"), "{sql}");
+        assert!(sql.contains("'shop'"), "{sql}");
+        assert!(!sql.contains("TABLE_NAME IN"), "every view of the database has to be parsed: {sql}");
+        let statements = Parser::parse_sql(&MySqlDialect {}, &sql).unwrap();
+        assert_eq!(statements.len(), 1);
+
+        assert!(mysql_view_definition_dependencies_sql("sh'op").contains("'sh''op'"));
+    }
+
     const LONG_TABLE: &str = "customer_order_line_item_revision_history";
 
     fn name(db: DatabaseType, table: &str) -> String {

crates/dbx-core/tests/live_mysql_transfer.rs

@@ -2179,3 +2179,99 @@ async fn live_mysql_progress_read_survives_total_duration_beyond_timeout() {
     source_cleanup.unwrap();
     target_cleanup.unwrap();
 }
+
+/// MySQL 5.7 has no `information_schema.VIEW_TABLE_USAGE` (the table arrived with 8.0), and
+/// the rebuild pre-pass used to abort the whole transfer on that lookup — even when the
+/// target held no views at all (#10027). On such a server the check has to fall back to
+/// inspecting the stored `information_schema.VIEWS.VIEW_DEFINITION` text, which must keep
+/// refusing to break a dependent view.
+#[tokio::test]
+#[ignore = "requires a MySQL 5.7 target via DBX_LIVE_MYSQL_TRANSFER_LEGACY_* variables"]
+async fn live_mysql_transfer_drop_target_inspects_dependent_views_without_the_usage_catalog() {
+    let suffix = uuid::Uuid::new_v4().simple().to_string();
+    let connection_id = format!("mysql-legacy-dependency-{suffix}");
+    let source_database = format!("dbx_legacy_src_{}", &suffix[..12]);
+    let target_database = format!("dbx_legacy_tgt_{}", &suffix[..12]);
+    let config = live_cross_version_mysql_config(&connection_id, "LEGACY");
+    let setup_pool = mysql::connect(&mysql_url(&config), Duration::from_secs(10)).await.unwrap();
+
+    let setup = format!(
+        "CREATE DATABASE `{source_database}`;\
+         CREATE TABLE `{source_database}`.`orders` (id INT PRIMARY KEY, total INT);\
+         INSERT INTO `{source_database}`.`orders` VALUES (1, 10);\
+         CREATE DATABASE `{target_database}`;\
+         CREATE TABLE `{target_database}`.`orders` (id INT PRIMARY KEY, total INT);\
+         INSERT INTO `{target_database}`.`orders` VALUES (99, 99);\
+         CREATE VIEW `{target_database}`.`order_totals` AS \
+             SELECT `{target_database}`.`orders`.`id` AS `id` FROM `{target_database}`.`orders`"
+    );
+    mysql::execute_query(&setup_pool, &setup, true).await.unwrap();
+    assert!(
+        mysql::execute_query(&setup_pool, "SELECT TABLE_NAME FROM information_schema.VIEW_TABLE_USAGE LIMIT 0", false)
+            .await
+            .is_err(),
+        "this test only exercises the fallback on a server without the 8.0-only view usage catalog"
+    );
+
+    let dir = std::env::temp_dir().join(format!("dbx-mysql-legacy-dependency-{suffix}"));
+    std::fs::create_dir_all(&dir).unwrap();
+    let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+    let state = Arc::new(AppState::new(storage));
+    state.configs.write().await.insert(connection_id.clone(), config);
+    let pool_key = state.get_or_create_pool(&connection_id, Some(&target_database)).await.unwrap();
+
+    let request = TransferRequest {
+        transfer_id: format!("legacy-dependency-{suffix}"),
+        source_connection_id: connection_id.clone(),
+        source_database: source_database.clone(),
+        source_schema: source_database.clone(),
+        source_catalog: None,
+        target_connection_id: connection_id.clone(),
+        target_database: target_database.clone(),
+        target_schema: target_database.clone(),
+        target_catalog: None,
+        tables: vec!["orders".to_string()],
+        create_table: true,
+        drop_target_before_create: true,
+        drop_target_confirmed: true,
+        content: TransferContent::default(),
+        objects: Vec::new(),
+        mode: TransferMode::Append,
+        target_table_name_case: TransferTableNameCase::Preserve,
+        quote_target_column_names: true,
+        ownership_policy: TransferOwnershipPolicy::Preserve,
+        batch_size: 10,
+    };
+
+    let blocked = rename_tables_to_backup(&state, &request, &request.tables, DatabaseType::Mysql, &pool_key, |_| {})
+        .await
+        .expect_err("a dependent view must block the rebuild on MySQL 5.7 as well");
+    assert!(
+        blocked.contains("TRANSFER_DROP_TARGET_EXTERNAL_DEPENDENCIES")
+            && blocked.contains(&format!("view {target_database}.order_totals -> {target_database}.orders")),
+        "the check must name the dependent view instead of failing on the missing catalog: {blocked}"
+    );
+    assert_eq!(
+        query_text(&setup_pool, &format!("SELECT CAST(COUNT(*) AS CHAR) FROM `{target_database}`.`orders`")).await,
+        "1",
+        "the blocked rebuild must leave the target table untouched"
+    );
+
+    // Without the dependent view the same rebuild has to go through: MySQL 5.7 used to fail
+    // here regardless of the target's views, which is the reported bug.
+    mysql::execute_query(&setup_pool, &format!("DROP VIEW `{target_database}`.`order_totals`"), true).await.unwrap();
+    let renamed = rename_tables_to_backup(&state, &request, &request.tables, DatabaseType::Mysql, &pool_key, |_| {})
+        .await
+        .expect("a MySQL 5.7 target without dependent views must be rebuildable");
+    assert_eq!(renamed.len(), 1);
+
+    let cleanup = mysql::execute_query(
+        &setup_pool,
+        &format!("DROP DATABASE `{source_database}`; DROP DATABASE `{target_database}`"),
+        true,
+    )
+    .await;
+    setup_pool.disconnect().await.unwrap();
+    let _ = std::fs::remove_dir_all(dir);
+    cleanup.unwrap();
+}