From 6dcfdc6ae0d344836c26a4b6598a22eb07a15008 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Fri, 14 Aug 2026 19:09:05 +0800 Subject: [PATCH] fix(mysql): isolate compatible database-specific logic --- crates/dbx-core/src/connection.rs | 29 +- crates/dbx-core/src/db/doris.rs | 27 + crates/dbx-core/src/db/manticoresearch.rs | 19 + crates/dbx-core/src/db/mod.rs | 3 + crates/dbx-core/src/db/mysql.rs | 767 +----------------- crates/dbx-core/src/db/mysql_compatible.rs | 9 + crates/dbx-core/src/db/oceanbase_mysql.rs | 56 ++ crates/dbx-core/src/db/starrocks.rs | 204 +++++ crates/dbx-core/src/db/tdsql_mysql.rs | 54 ++ crates/dbx-core/src/db/tidb.rs | 24 + crates/dbx-core/src/query.rs | 2 +- crates/dbx-core/src/schema.rs | 103 +-- .../dbx-core/src/schema/providers/native.rs | 40 +- crates/dbx-core/src/sql.rs | 21 +- crates/dbx-core/src/transfer.rs | 13 +- 15 files changed, 484 insertions(+), 887 deletions(-) create mode 100644 crates/dbx-core/src/db/oceanbase_mysql.rs create mode 100644 crates/dbx-core/src/db/tdsql_mysql.rs create mode 100644 crates/dbx-core/src/db/tidb.rs diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index 57d1bf9e68..5b233314ff 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -63,21 +63,8 @@ pub enum MysqlMode { OceanBaseOracle, } -fn is_oceanbase_mysql_config(config: &ConnectionConfig) -> bool { - config.db_type == DatabaseType::Mysql - && config.driver_profile.as_deref().is_some_and(|profile| profile.eq_ignore_ascii_case("oceanbase")) -} - -pub(crate) fn oceanbase_mysql_query_timeout_sql(config: &ConnectionConfig, timeout_secs: u64) -> Option { - if !is_oceanbase_mysql_config(config) || timeout_secs == 0 { - return None; - } - let timeout_us = timeout_secs.saturating_mul(1_000_000); - Some(format!("SET ob_query_timeout = {timeout_us}")) -} - fn oceanbase_mysql_setup_queries(config: &ConnectionConfig) -> Vec { - oceanbase_mysql_query_timeout_sql(config, config.query_timeout_secs).into_iter().collect() + db::oceanbase_mysql::query_timeout_sql(config, config.query_timeout_secs).into_iter().collect() } fn mysql_pool_setup_queries(config: &ConnectionConfig, url: &str) -> Vec { @@ -5246,12 +5233,12 @@ mod tests { database_connection_config_with_catalog, gaussdb_identifier_quote_from_query_result, gaussdb_m_jdbc_config_for_endpoint, gaussdb_uses_m_jdbc_driver, kafka_single_loopback_bootstrap_endpoint, metadata_connection_config, mysql_metadata_fallback_url, mysql_pool_setup_queries, - oceanbase_mysql_query_timeout_sql, oceanbase_mysql_setup_queries, prestosql_jdbc_config_for_endpoint, - redacted_connection_url_for_endpoint, redis_sentinel_transport_id, redis_sentinel_transport_prefix, - sqlserver_legacy_agent_config, sqlserver_legacy_driver_error, sqlserver_uses_legacy_driver, - task_client_session_id, upsert_connection_url_param, uses_bare_mysql_pool, uses_tcp_probe, - validate_connection_url_params, validate_h2_database_path, AppState, MysqlMode, PoolKind, - GAUSSDB_M_JDBC_DRIVER_CLASS, GAUSSDB_M_JDBC_DRIVER_PROFILE, PRESTOSQL_JDBC_DRIVER_CLASS, + oceanbase_mysql_setup_queries, prestosql_jdbc_config_for_endpoint, redacted_connection_url_for_endpoint, + redis_sentinel_transport_id, redis_sentinel_transport_prefix, sqlserver_legacy_agent_config, + sqlserver_legacy_driver_error, sqlserver_uses_legacy_driver, task_client_session_id, + upsert_connection_url_param, uses_bare_mysql_pool, uses_tcp_probe, validate_connection_url_params, + validate_h2_database_path, AppState, MysqlMode, PoolKind, GAUSSDB_M_JDBC_DRIVER_CLASS, + GAUSSDB_M_JDBC_DRIVER_PROFILE, PRESTOSQL_JDBC_DRIVER_CLASS, }; use crate::agent_connection::{ agent_connect_params, mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver, @@ -5857,7 +5844,7 @@ mod tests { config.driver_profile = Some("oceanbase".to_string()); assert_eq!( - oceanbase_mysql_query_timeout_sql(&config, 300_000), + db::oceanbase_mysql::query_timeout_sql(&config, 300_000), Some("SET ob_query_timeout = 300000000000".to_string()) ); } diff --git a/crates/dbx-core/src/db/doris.rs b/crates/dbx-core/src/db/doris.rs index 627e8416d8..0be6dc1ba1 100644 --- a/crates/dbx-core/src/db/doris.rs +++ b/crates/dbx-core/src/db/doris.rs @@ -1,5 +1,32 @@ +use crate::models::connection::{ConnectionConfig, DatabaseType}; + pub use super::mysql_compatible::{ get_columns_show_from as get_catalog_columns, list_catalog_indexes, list_catalogs, list_databases_show_from as list_catalog_databases, list_indexes_with_ddl_fallback as list_indexes, list_tables_show_from as list_catalog_tables, show_create_table_ddl_from as get_catalog_table_ddl, }; + +pub fn is_config(config: &ConnectionConfig) -> bool { + is_profile(&config.db_type, config.driver_profile.as_deref()) +} + +pub fn is_profile(db_type: &DatabaseType, driver_profile: Option<&str>) -> bool { + *db_type == DatabaseType::Doris + || (*db_type == DatabaseType::Mysql + && driver_profile + .is_some_and(|profile| matches!(profile.to_ascii_lowercase().as_str(), "doris" | "selectdb"))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn profile_matches_doris_and_selectdb_only() { + assert!(is_profile(&DatabaseType::Doris, None)); + assert!(is_profile(&DatabaseType::Mysql, Some("DORIS"))); + assert!(is_profile(&DatabaseType::Mysql, Some("selectdb"))); + assert!(!is_profile(&DatabaseType::Mysql, Some("starrocks"))); + assert!(!is_profile(&DatabaseType::Postgres, Some("doris"))); + } +} diff --git a/crates/dbx-core/src/db/manticoresearch.rs b/crates/dbx-core/src/db/manticoresearch.rs index bc53bb4498..f48c0cfba0 100644 --- a/crates/dbx-core/src/db/manticoresearch.rs +++ b/crates/dbx-core/src/db/manticoresearch.rs @@ -1,10 +1,21 @@ use mysql_async::prelude::*; use std::collections::HashMap; +use crate::models::connection::{ConnectionConfig, DatabaseType}; use crate::types::{ColumnInfo, IndexInfo, ObjectInfo}; use super::mysql::{self, MySqlPool}; +pub fn is_config(config: &ConnectionConfig) -> bool { + is_profile(&config.db_type, config.driver_profile.as_deref()) +} + +pub fn is_profile(db_type: &DatabaseType, driver_profile: Option<&str>) -> bool { + *db_type == DatabaseType::ManticoreSearch + || (*db_type == DatabaseType::Mysql + && driver_profile.is_some_and(|profile| profile.eq_ignore_ascii_case("manticoresearch"))) +} + fn quote_identifier(value: &str) -> String { format!("`{}`", value.replace('`', "``")) } @@ -205,6 +216,14 @@ fn index_info_from_row(row: &mysql_async::Row) -> IndexInfo { mod tests { use super::*; + #[test] + fn profile_matches_native_and_mysql_profile_connections() { + assert!(is_profile(&DatabaseType::ManticoreSearch, None)); + assert!(is_profile(&DatabaseType::Mysql, Some("MANTICORESEARCH"))); + assert!(!is_profile(&DatabaseType::Mysql, None)); + assert!(!is_profile(&DatabaseType::Postgres, Some("manticoresearch"))); + } + #[test] fn show_metadata_uses_unqualified_table_names_when_database_is_empty() { assert_eq!(show_columns_sql("", "idx", true), "SHOW FULL COLUMNS FROM `idx`"); diff --git a/crates/dbx-core/src/db/mod.rs b/crates/dbx-core/src/db/mod.rs index 601ca3f6e4..a500b55ad4 100644 --- a/crates/dbx-core/src/db/mod.rs +++ b/crates/dbx-core/src/db/mod.rs @@ -24,6 +24,7 @@ pub mod mongo_driver; pub mod mysql; pub mod mysql_compatible; pub mod ob_oracle; +pub mod oceanbase_mysql; pub mod postgres; pub mod proxy_tunnel; pub mod questdb; @@ -35,6 +36,8 @@ pub mod ssh_host_key; pub mod ssh_prompt; pub mod ssh_tunnel; pub mod starrocks; +pub mod tdsql_mysql; +pub mod tidb; pub mod transport_layer_tunnel; pub mod turso_driver; pub mod vector_driver; diff --git a/crates/dbx-core/src/db/mysql.rs b/crates/dbx-core/src/db/mysql.rs index b377311a71..b4a6b46928 100644 --- a/crates/dbx-core/src/db/mysql.rs +++ b/crates/dbx-core/src/db/mysql.rs @@ -41,14 +41,12 @@ pub(crate) fn mysql_catalog_dialect( db_type: DatabaseType, driver_profile: Option<&str>, ) -> Option { - match db_type { - DatabaseType::Doris => Some(MySqlCatalogDialect::Doris), - DatabaseType::StarRocks => Some(MySqlCatalogDialect::StarRocks), - _ => match driver_profile.map(str::to_ascii_lowercase).as_deref() { - Some("doris" | "selectdb") => Some(MySqlCatalogDialect::Doris), - Some("starrocks") => Some(MySqlCatalogDialect::StarRocks), - _ => None, - }, + if super::doris::is_profile(&db_type, driver_profile) { + Some(MySqlCatalogDialect::Doris) + } else if super::starrocks::is_profile(&db_type, driver_profile) { + Some(MySqlCatalogDialect::StarRocks) + } else { + None } } @@ -74,14 +72,11 @@ const MYSQL_RESULT_CELL_PREVIEW_MAX_BYTES: usize = 8 * 1024; impl MySqlQueryDialect { pub fn for_connection(db_type: DatabaseType, driver_profile: Option<&str>) -> Self { - let profile = driver_profile.map(str::to_ascii_lowercase); Self { - supports_admin_show_results: matches!( - db_type, - DatabaseType::Doris | DatabaseType::StarRocks | DatabaseType::ManticoreSearch - ) || profile.as_deref().is_some_and(|profile| { - matches!(profile, "doris" | "selectdb" | "starrocks" | "manticoresearch" | "tidb") - }), + supports_admin_show_results: super::doris::is_profile(&db_type, driver_profile) + || super::starrocks::is_profile(&db_type, driver_profile) + || super::manticoresearch::is_profile(&db_type, driver_profile) + || super::tidb::is_profile(&db_type, driver_profile), } } } @@ -91,7 +86,7 @@ pub enum MySqlQueryStreamItem { Row(Vec), } -fn quote_value(s: &str) -> String { +pub(super) fn quote_value(s: &str) -> String { format!("'{}'", s.replace('\\', "\\\\").replace('\'', "\\'")) } @@ -2655,7 +2650,7 @@ pub async fn get_table_comment(pool: &MySqlPool, database: &str, table: &str) -> } #[derive(Clone, Debug, Default)] -struct TableStatusMeta { +pub(super) struct TableStatusMeta { comment: Option, created_at: Option, updated_at: Option, @@ -2889,7 +2884,7 @@ fn mysql_fallback_like_patterns(filter: &str) -> Vec { patterns } -async fn list_tables_show_with_status( +pub(super) async fn list_tables_show_with_status( pool: &MySqlPool, database: &str, ) -> Result<(Vec, HashMap), String> { @@ -2943,116 +2938,6 @@ pub async fn list_tables_show(pool: &MySqlPool, database: &str) -> Result String { - format!( - "SELECT TABLE_NAME FROM information_schema.materialized_views WHERE TABLE_SCHEMA = {}", - quote_value(database) - ) -} - -/// Fallback DDL source for StarRocks materialized views when `SHOW CREATE -/// MATERIALIZED VIEW` fails (e.g. on versions predating starrocks/starrocks#73396, -/// merged 2026-05-19, which reject the statement for sync MVs with "Table not -/// found" because sync MVs are not registered as separate Tables). -/// -/// `information_schema.materialized_views` is documented as the authoritative -/// list of all materialized views, with a column distinguishing SYNC from -/// ASYNC. See -/// https://docs.starrocks.io/docs/sql-reference/information_schema/materialized_views/. -/// -/// Made `pub(super)` so the dispatch site in `schema::mysql_object_source` can -/// rely on it without rewriting the escape convention. -pub(crate) fn mysql_materialized_view_definition_sql(database: &str, name: &str) -> String { - format!( - "SELECT MATERIALIZED_VIEW_DEFINITION \ - FROM information_schema.materialized_views \ - WHERE TABLE_SCHEMA = {} AND TABLE_NAME = {} \ - LIMIT 1", - quote_value(database), - quote_value(name) - ) -} - -async fn list_starrocks_materialized_view_names(pool: &MySqlPool, database: &str) -> Result, String> { - let sql = starrocks_materialized_views_sql(database); - let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?; - let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; - let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; - Ok(rows - .iter() - .filter_map(|row| { - let name = get_str_by_name(row, "TABLE_NAME").trim().to_string(); - (!name.is_empty()).then_some(name) - }) - .collect()) -} - -fn merge_starrocks_materialized_views( - tables: &mut Vec, - materialized_view_names: Result, String>, - database: &str, -) { - let materialized_view_names = match materialized_view_names { - Ok(names) => names, - Err(err) => { - // Older StarRocks versions and restricted accounts may not expose this - // information_schema view; keep the base SHOW TABLES result usable. - log::warn!("Skipping materialized view classification for StarRocks database `{database}`: {err}"); - return; - } - }; - - // Snapshot the names already returned by SHOW FULL TABLES so the second pass can - // append MVs that are absent from SHOW FULL TABLES without duplicating rows. - let known_names: HashSet = tables.iter().map(|table| table.name.clone()).collect(); - - // Step 1 — reclassify: rows whose name appears in `information_schema.materialized_views` - // are MVs even when SHOW FULL TABLES labeled them as VIEW (sync MVs) or BASE TABLE - // (async MVs). See https://docs.starrocks.io/docs/sql-reference/information_schema/materialized_views/ - // for the authoritative distinction between the two MV kinds. - for table in tables.iter_mut() { - if materialized_view_names.contains(&table.name) { - table.table_type = "MATERIALIZED_VIEW".to_string(); - } - } - - // Step 2 — union: on StarRocks versions predating starrocks/starrocks#73396 (merged - // 2026-05-19), sync MVs "are not registered as separate Tables" so SHOW FULL TABLES - // omits them entirely. Append those rows from the system view so they appear in the - // sidebar and the DDL source path has something to resolve. Sort names so that - // the resulting table order is deterministic across runs. - let mut materialized_view_names_sorted: Vec<&String> = materialized_view_names.iter().collect(); - materialized_view_names_sorted.sort(); - for name in materialized_view_names_sorted { - if !known_names.contains(name.as_str()) { - tables.push(TableInfo { - name: name.clone(), - table_type: "MATERIALIZED_VIEW".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }); - } - } -} - -async fn list_starrocks_tables_with_status( - pool: &MySqlPool, - database: &str, -) -> Result<(Vec, HashMap), String> { - let (tables, materialized_view_names) = tokio::join!( - list_tables_show_with_status(pool, database), - list_starrocks_materialized_view_names(pool, database) - ); - let (mut tables, status) = tables?; - merge_starrocks_materialized_views(&mut tables, materialized_view_names, database); - Ok((tables, status)) -} - -pub async fn list_starrocks_tables(pool: &MySqlPool, database: &str) -> Result, String> { - list_starrocks_tables_with_status(pool, database).await.map(|(tables, _)| tables) -} - fn requested_object_type(object_types: Option<&[String]>, object_type: &str) -> bool { object_types.is_none_or(|types| { types.is_empty() || types.iter().any(|candidate| candidate.eq_ignore_ascii_case(object_type)) @@ -3447,21 +3332,7 @@ fn filter_table_objects_fallback( .collect() } -pub async fn list_starrocks_table_objects(pool: &MySqlPool, database: &str) -> Result, String> { - let (tables, routines) = - tokio::join!(list_starrocks_tables_with_status(pool, database), list_routine_objects(pool, database)); - let (tables, status) = tables?; - let mut objects = table_infos_to_objects(tables, &status, database); - - match routines { - Ok(routines) => objects.extend(routines), - Err(err) => log::warn!("Skipping routines for database `{}` in object browser: {}", database, err), - } - - Ok(objects) -} - -fn table_infos_to_objects( +pub(super) fn table_infos_to_objects( tables: Vec, status: &HashMap, database: &str, @@ -3497,7 +3368,7 @@ fn table_infos_to_objects( .collect() } -async fn list_routine_objects(pool: &MySqlPool, database: &str) -> Result, String> { +pub(super) async fn list_routine_objects(pool: &MySqlPool, database: &str) -> Result, String> { let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?; let rows = query_routine_rows(&mut conn, database, None, None, None).await?; Ok(rows.iter().map(|row| row_to_object(row, database)).collect()) @@ -5302,438 +5173,6 @@ pub async fn show_create_table_ddl(pool: &MySqlPool, database: &str, table: &str .ok_or_else(|| "Failed to read DDL".to_string()) } -// --------------------------------------------------------------------------- -// Doris / StarRocks multi-catalog support. -// -// These engines expose external catalogs (iceberg, hive, jdbc, ...) alongside -// the native `internal` catalog via `SHOW CATALOGS`. The functions below address -// objects in a specific catalog using 3-part qualified names -// (`..`), which the engines accept directly without -// needing to `SWITCH` the session catalog. -// --------------------------------------------------------------------------- - -/// Build a 2-part qualified identifier `` ``.`` ``. -fn doris_catalog_database_ref(catalog: &str, database: &str) -> String { - format!("{}.{}", quote_identifier(catalog), quote_identifier(database)) -} - -/// Build a 3-part qualified identifier `` ``.``.`
` ``. -fn doris_catalog_table_ref(catalog: &str, database: &str, table: &str) -> String { - format!("{}.{}.{}", quote_identifier(catalog), quote_identifier(database), quote_identifier(table)) -} - -/// `SHOW CATALOGS` → list of catalogs visible to the current user. -/// -/// Column layouts differ between engines: Doris exposes `CatalogName` (with -/// `CatalogId`/`IsCurrent`/`CreateTime`/`LastUpdateTime`), while StarRocks -/// exposes `Catalog` (only `Type`/`Comment`, no `IsCurrent`). The name is read -/// from either column; missing trailing columns degrade gracefully to -/// empty/None. The built-in catalog is named `internal` in Doris and -/// `default_catalog` in StarRocks (both with `Type=internal`); detection is -/// type-based (see `CatalogInfo::is_internal`), not name-based. -pub async fn list_doris_catalogs(pool: &MySqlPool) -> Result, String> { - let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?; - let result = conn.query_iter("SHOW CATALOGS").await.map_err(|e| e.to_string())?; - let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; - let catalogs: Vec = rows - .iter() - .filter_map(|row| { - // Doris column is `CatalogName`; StarRocks column is `Catalog`. - let name = first_nonempty_str_by_name(row, &["CatalogName", "Catalog"]).trim().to_string(); - if name.is_empty() { - return None; - } - let catalog_type = get_str_by_name(row, "Type").trim().to_string(); - let is_current = { - let value = get_str_by_name(row, "IsCurrent").trim().to_ascii_lowercase(); - !value.is_empty() && value != "no" && value != "false" && value != "0" - }; - let comment = get_opt_str(row, "Comment").map(|s| s.trim().to_string()).filter(|s| !s.is_empty()); - Some(crate::db::CatalogInfo { name, catalog_type, is_current, comment }) - }) - .collect(); - Ok(normalize_doris_catalogs(catalogs)) -} - -/// Sort with the built-in catalog first, then the rest alphabetically by name. -/// The built-in catalog is identified by `CatalogInfo::is_internal` (type-based) -/// rather than by name, so StarRocks `default_catalog` sorts first just like -/// Doris `internal`. No synthetic catalog is injected: `SHOW CATALOGS` always -/// lists the built-in catalog on both engines, and a single-catalog result is -/// handled by the flat-sidebar fallback in the caller. -fn normalize_doris_catalogs(mut catalogs: Vec) -> Vec { - catalogs.sort_by(|a, b| match (a.is_internal(), b.is_internal()) { - (true, false) => std::cmp::Ordering::Less, - (false, true) => std::cmp::Ordering::Greater, - _ => a.name.cmp(&b.name), - }); - catalogs -} - -/// `SHOW DATABASES FROM ` → databases in the given catalog. -pub async fn list_databases_show_from(pool: &MySqlPool, catalog: &str) -> Result, String> { - let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?; - let sql = format!("SHOW DATABASES FROM {}", quote_identifier(catalog)); - let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; - let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; - Ok(database_infos_from_names(rows.iter().map(|row| get_str(row, 0)), false)) -} - -/// `SHOW TABLES FROM .` → tables in an external catalog. -/// -/// External catalogs do not support `SHOW TABLE STATUS`, so comments/status are -/// not fetched (the caller only needs names + types for browsing). -pub async fn list_tables_show_from(pool: &MySqlPool, catalog: &str, database: &str) -> Result, String> { - let sql = format!("SHOW TABLES FROM {}", doris_catalog_database_ref(catalog, database)); - let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?; - let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; - let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; - let mut tables: Vec = rows - .iter() - .filter_map(|row| { - let name = get_str(row, 0).trim().to_string(); - if name.is_empty() { - return None; - } - // SHOW FULL TABLES exposes a type column; plain SHOW TABLES does not. - let table_type = get_str(row, 1); - Some(TableInfo { - name, - table_type: if table_type.trim().is_empty() { "TABLE".to_string() } else { table_type }, - comment: None, - parent_schema: None, - parent_name: None, - }) - }) - .collect(); - tables.sort_by(|a, b| a.name.cmp(&b.name)); - Ok(tables) -} - -/// `SHOW COLUMNS FROM ..
` → columns of an external -/// catalog table. Falls back to `DESCRIBE` if `SHOW COLUMNS` is rejected. -pub async fn get_columns_show_from( - pool: &MySqlPool, - catalog: &str, - database: &str, - table: &str, -) -> Result, String> { - let qualified = doris_catalog_table_ref(catalog, database, table); - let full_sql = format!("SHOW FULL COLUMNS FROM {qualified}"); - let plain_sql = format!("SHOW COLUMNS FROM {qualified}"); - let describe_sql = format!("DESCRIBE {qualified}"); - let mut conn = get_conn_with_health_check(pool).await?; - let rows: Vec = match conn.query_iter(&full_sql).await { - Ok(result) => result.collect_and_drop().await.map_err(|e| e.to_string())?, - Err(_) => match conn.query_iter(&plain_sql).await { - Ok(result) => result.collect_and_drop().await.map_err(|e| e.to_string())?, - Err(_) => { - let result = conn.query_iter(&describe_sql).await.map_err(|e| e.to_string())?; - result.collect_and_drop().await.map_err(|e| e.to_string())? - } - }, - }; - Ok(rows - .iter() - .filter_map(|row| { - let name = get_str_by_name(row, "Field").trim().to_string(); - if name.is_empty() { - return None; - } - let key = get_str_by_name(row, "Key"); - let collation = get_opt_str(row, "Collation").filter(|s| !s.is_empty()); - Some(ColumnInfo { - name, - data_type: get_str_by_name(row, "Type"), - is_nullable: get_str_by_name(row, "Null").eq_ignore_ascii_case("YES"), - column_default: get_opt_str(row, "Default"), - is_primary_key: key.eq_ignore_ascii_case("PRI"), - is_unique: key.eq_ignore_ascii_case("UNI"), - extra: get_opt_str(row, "Extra"), - comment: get_opt_str(row, "Comment") - .map(|s| fix_potential_double_encoding(&s)) - .filter(|s| !s.is_empty()), - numeric_precision: None, - numeric_scale: None, - character_maximum_length: None, - enum_values: None, - character_set: collation - .as_deref() - .and_then(|c| c.split_once('_').map(|(charset, _)| charset.to_string())) - .filter(|s| !s.is_empty()), - collation, - }) - }) - .collect()) -} - -/// `SHOW CREATE TABLE ..
` → DDL for an external -/// catalog table. -pub async fn show_create_table_ddl_from( - pool: &MySqlPool, - catalog: &str, - database: &str, - table: &str, -) -> Result { - let sql = format!("SHOW CREATE TABLE {}", doris_catalog_table_ref(catalog, database, table)); - let mut conn = get_conn_with_health_check(pool).await?; - let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; - let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; - let row = rows.first().ok_or("DDL not found")?; - row.get_opt::(1) - .and_then(|result| result.ok()) - .or_else(|| { - row.get_opt::, usize>(1) - .and_then(|result| result.ok()) - .map(|b| String::from_utf8_lossy(&b).to_string()) - }) - .ok_or_else(|| "Failed to read DDL".to_string()) -} - -/// Best-effort index listing for an external catalog table. External catalogs -/// generally do not expose MySQL-style index metadata via `information_schema` -/// (that view is scoped to the internal catalog), so indexes are derived from -/// `SHOW CREATE TABLE` parsing. Returns empty on failure (graceful degradation -/// — indexes are informational for external tables). -pub async fn list_doris_catalog_indexes( - pool: &MySqlPool, - catalog: &str, - database: &str, - table: &str, -) -> Result, String> { - let ddl = show_create_table_ddl_from(pool, catalog, database, table).await?; - Ok(doris_indexes_from_create_table_ddl(&ddl)) -} - -fn doris_indexes_from_create_table_ddl(ddl: &str) -> Vec { - let mut indexes = Vec::new(); - for raw_line in ddl.lines() { - let line = trim_ddl_definition_line(raw_line); - if line.is_empty() { - continue; - } - let upper = line.to_ascii_uppercase(); - if upper.starts_with("PRIMARY KEY") { - if let Some(index) = doris_table_key_index("PRIMARY", line, true, true, "PRIMARY KEY") { - indexes.push(index); - } - } else if upper.starts_with("UNIQUE KEY") { - if let Some(index) = doris_table_key_index("UNIQUE KEY", line, true, false, "UNIQUE KEY") { - indexes.push(index); - } - } else if upper.starts_with("INDEX ") { - if let Some(index) = doris_secondary_index(line) { - indexes.push(index); - } - } - } - indexes -} - -fn trim_ddl_definition_line(line: &str) -> &str { - let mut trimmed = line.trim(); - if let Some(rest) = trimmed.strip_prefix(',') { - trimmed = rest.trim_start(); - } - while let Some(rest) = trimmed.strip_suffix(',') { - trimmed = rest.trim_end(); - } - trimmed -} - -fn doris_table_key_index( - name: &str, - line: &str, - is_unique: bool, - is_primary: bool, - index_type: &str, -) -> Option { - let columns = parse_mysql_index_columns(first_parenthesized_content(line)?); - if columns.is_empty() { - return None; - } - Some(IndexInfo { - name: name.to_string(), - columns, - is_unique, - is_primary, - filter: None, - index_type: Some(index_type.to_string()), - included_columns: None, - comment: None, - }) -} - -fn doris_secondary_index(line: &str) -> Option { - let (_, rest) = split_keyword_prefix(line, "INDEX")?; - let (name, after_name) = read_mysql_identifier(rest.trim_start())?; - let columns = parse_mysql_index_columns(first_parenthesized_content(after_name)?); - if columns.is_empty() { - return None; - } - Some(IndexInfo { - name, - columns, - is_unique: false, - is_primary: false, - filter: None, - index_type: mysql_keyword_argument(after_name, "USING").or_else(|| Some("INDEX".to_string())), - included_columns: None, - comment: mysql_quoted_string_argument(after_name, "COMMENT"), - }) -} - -fn split_keyword_prefix<'a>(line: &'a str, keyword: &str) -> Option<(&'a str, &'a str)> { - if line.len() < keyword.len() || !line[..keyword.len()].eq_ignore_ascii_case(keyword) { - return None; - } - let rest = &line[keyword.len()..]; - if !rest.is_empty() && is_mysql_identifier_byte(rest.as_bytes()[0]) { - return None; - } - Some((&line[..keyword.len()], rest)) -} - -fn read_mysql_identifier(input: &str) -> Option<(String, &str)> { - let input = input.trim_start(); - if input.is_empty() { - return None; - } - let bytes = input.as_bytes(); - if bytes[0] == b'`' { - let mut i = 1; - let mut value = String::new(); - while i < bytes.len() { - if bytes[i] == b'`' { - if i + 1 < bytes.len() && bytes[i + 1] == b'`' { - value.push('`'); - i += 2; - continue; - } - return Some((value, &input[i + 1..])); - } - let ch = input[i..].chars().next()?; - value.push(ch); - i += ch.len_utf8(); - } - return None; - } - - let end = input.find(|ch: char| ch.is_whitespace() || matches!(ch, '(' | ')' | ',')).unwrap_or(input.len()); - if end == 0 { - return None; - } - Some((input[..end].to_string(), &input[end..])) -} - -fn first_parenthesized_content(input: &str) -> Option<&str> { - let bytes = input.as_bytes(); - let mut depth = 0usize; - let mut start = None; - let mut i = 0usize; - while i < bytes.len() { - match bytes[i] { - b'\'' | b'"' | b'`' => { - i = skip_mysql_quoted(input, i, bytes[i]); - continue; - } - b'(' => { - if depth == 0 { - start = Some(i + 1); - } - depth += 1; - } - b')' if depth > 0 => { - depth -= 1; - if depth == 0 { - return start.map(|start| &input[start..i]); - } - } - _ => {} - } - i += 1; - } - None -} - -fn split_top_level_csv(input: &str) -> Vec<&str> { - let bytes = input.as_bytes(); - let mut parts = Vec::new(); - let mut depth = 0usize; - let mut start = 0usize; - let mut i = 0usize; - while i < bytes.len() { - match bytes[i] { - b'\'' | b'"' | b'`' => { - i = skip_mysql_quoted(input, i, bytes[i]); - continue; - } - b'(' => depth += 1, - b')' if depth > 0 => depth -= 1, - b',' if depth == 0 => { - parts.push(input[start..i].trim()); - start = i + 1; - } - _ => {} - } - i += 1; - } - parts.push(input[start..].trim()); - parts -} - -fn parse_mysql_index_columns(input: &str) -> Vec { - split_top_level_csv(input) - .into_iter() - .filter_map(|part| read_mysql_identifier(part).map(|(column, _)| column)) - .filter(|column| !column.is_empty()) - .collect() -} - -fn mysql_keyword_argument(input: &str, keyword: &str) -> Option { - let bytes = input.as_bytes(); - let mut i = 0usize; - while i < bytes.len() { - match bytes[i] { - b'\'' | b'"' | b'`' => { - i = skip_mysql_quoted(input, i, bytes[i]); - continue; - } - _ if mysql_keyword_at(input, i, keyword) => { - return read_mysql_identifier(&input[i + keyword.len()..]).map(|(value, _)| value); - } - _ => i += 1, - } - } - None -} - -fn mysql_quoted_string_argument(input: &str, keyword: &str) -> Option { - let bytes = input.as_bytes(); - let mut i = 0usize; - while i < bytes.len() { - match bytes[i] { - b'\'' | b'"' | b'`' => { - i = skip_mysql_quoted(input, i, bytes[i]); - continue; - } - _ if mysql_keyword_at(input, i, keyword) => { - let rest = input[i + keyword.len()..].trim_start(); - if rest.as_bytes().first().copied() != Some(b'\'') { - return None; - } - let end = skip_mysql_quoted(rest, 0, b'\''); - if end <= 1 || end > rest.len() { - return None; - } - return Some(rest[1..end - 1].replace("\\'", "'").replace("''", "'")); - } - _ => i += 1, - } - } - None -} - pub async fn list_foreign_keys(pool: &MySqlPool, database: &str, table: &str) -> Result, String> { let column_sql = format!( "SELECT CONSTRAINT_NAME, COLUMN_NAME, REFERENCED_TABLE_SCHEMA, \ @@ -6540,182 +5979,6 @@ mod tests { assert_eq!(filtered.iter().map(|object| object.name.as_str()).collect::>(), vec!["c_table"]); } - #[test] - fn starrocks_materialized_views_are_classified_without_duplicating_tables() { - let mut tables = vec![ - TableInfo { - name: "orders".to_string(), - table_type: "BASE TABLE".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }, - TableInfo { - name: "orders_view".to_string(), - table_type: "VIEW".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }, - TableInfo { - name: "orders_mv".to_string(), - table_type: "VIEW".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }, - ]; - let materialized_views = HashSet::from(["orders_mv".to_string(), "orders_mv".to_string()]); - - merge_starrocks_materialized_views(&mut tables, Ok(materialized_views), "analytics"); - - assert_eq!(tables.len(), 3); - assert_eq!( - tables.iter().map(|table| (table.name.as_str(), table.table_type.as_str())).collect::>(), - vec![("orders", "BASE TABLE"), ("orders_view", "VIEW"), ("orders_mv", "MATERIALIZED_VIEW")] - ); - } - - #[test] - fn starrocks_async_materialized_views_reported_as_base_table_are_reclassified() { - // Async materialized views (StarRocks >= 2.5) appear as `BASE TABLE` in - // `SHOW FULL TABLES`. Classification must trust the - // `information_schema.materialized_views` source. - let mut tables = vec![ - TableInfo { - name: "orders".to_string(), - table_type: "BASE TABLE".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }, - TableInfo { - name: "orders_async_mv".to_string(), - table_type: "BASE TABLE".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }, - ]; - let materialized_views = HashSet::from(["orders_async_mv".to_string()]); - - merge_starrocks_materialized_views(&mut tables, Ok(materialized_views), "analytics"); - - assert_eq!( - tables.iter().map(|table| (table.name.as_str(), table.table_type.as_str())).collect::>(), - vec![("orders", "BASE TABLE"), ("orders_async_mv", "MATERIALIZED_VIEW")] - ); - } - - #[test] - fn starrocks_materialized_view_lookup_failure_keeps_base_types() { - let mut tables = vec![TableInfo { - name: "orders_mv".to_string(), - table_type: "VIEW".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }]; - - merge_starrocks_materialized_views(&mut tables, Err("permission denied".to_string()), "analytics"); - - assert_eq!(tables[0].table_type, "VIEW"); - } - - #[test] - fn starrocks_sync_mv_absent_from_show_full_tables_is_appended_from_information_schema() { - // StarRocks versions predating starrocks/starrocks#73396 (merged - // 2026-05-19) report sync MVs as "not registered as separate Tables", - // so SHOW FULL TABLES omits them. The merger must union names from - // information_schema.materialized_views so the sidebar and DDL path - // still resolve them. - let mut tables = vec![TableInfo { - name: "orders".to_string(), - table_type: "BASE TABLE".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }]; - let materialized_views = HashSet::from([ - "orders_mv".to_string(), // already present (reclassify path) - "daily_orders_mv".to_string(), // absent from SHOW FULL TABLES (union path) - ]); - - merge_starrocks_materialized_views(&mut tables, Ok(materialized_views), "analytics"); - - assert_eq!(tables.len(), 3); - assert_eq!( - tables.iter().map(|table| (table.name.as_str(), table.table_type.as_str())).collect::>(), - vec![ - ("orders", "BASE TABLE"), - ("daily_orders_mv", "MATERIALIZED_VIEW"), - ("orders_mv", "MATERIALIZED_VIEW"), - ] - ); - } - - #[test] - fn starrocks_materialized_view_query_is_scoped_to_database() { - let sql = starrocks_materialized_views_sql("tenant's analytics"); - - assert_eq!( - sql, - "SELECT TABLE_NAME FROM information_schema.materialized_views WHERE TABLE_SCHEMA = 'tenant\\'s analytics'" - ); - } - - #[test] - fn mysql_materialized_view_definition_fallback_is_scoped_to_db_and_name() { - // StarRocks predating PR 73396 (merged 2026-05-19) rejects - // `SHOW CREATE MATERIALIZED VIEW` for sync MVs with "Table not found" - // because sync MVs are not registered as separate Tables. The fallback - // path queries information_schema.materialized_views directly. The - // regression guards the SQL shape and the value escaping used by that - // fallback so the wire format isn't accidentally regressed. - assert_eq!( - mysql_materialized_view_definition_sql("shop", "daily_sales_mv"), - "SELECT MATERIALIZED_VIEW_DEFINITION FROM information_schema.materialized_views WHERE TABLE_SCHEMA = 'shop' AND TABLE_NAME = 'daily_sales_mv' LIMIT 1" - ); - assert_eq!( - mysql_materialized_view_definition_sql("tenant's analytics", "weird'name"), - "SELECT MATERIALIZED_VIEW_DEFINITION FROM information_schema.materialized_views WHERE TABLE_SCHEMA = 'tenant\\'s analytics' AND TABLE_NAME = 'weird\\'name' LIMIT 1" - ); - } - - #[test] - fn starrocks_object_conversion_preserves_table_view_and_materialized_view_types() { - let tables = vec![ - TableInfo { - name: "orders".to_string(), - table_type: "BASE TABLE".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }, - TableInfo { - name: "orders_view".to_string(), - table_type: "VIEW".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }, - TableInfo { - name: "orders_mv".to_string(), - table_type: "MATERIALIZED_VIEW".to_string(), - comment: None, - parent_schema: None, - parent_name: None, - }, - ]; - - let objects = table_infos_to_objects(tables, &HashMap::new(), "analytics"); - - assert_eq!( - objects.iter().map(|object| (object.name.as_str(), object.object_type.as_str())).collect::>(), - vec![("orders", "TABLE"), ("orders_view", "VIEW"), ("orders_mv", "MATERIALIZED_VIEW")] - ); - } - #[test] fn mysql_table_comment_sql_targets_single_table() { let sql = table_comment_sql("app", "users"); diff --git a/crates/dbx-core/src/db/mysql_compatible.rs b/crates/dbx-core/src/db/mysql_compatible.rs index 27524a4920..9c4aefe9e6 100644 --- a/crates/dbx-core/src/db/mysql_compatible.rs +++ b/crates/dbx-core/src/db/mysql_compatible.rs @@ -1,6 +1,7 @@ use mysql_async::prelude::*; use std::collections::HashSet; +use crate::models::connection::ConnectionConfig; use crate::types::{ColumnInfo, DatabaseInfo, IndexInfo, TableInfo}; use super::mysql::{ @@ -9,6 +10,14 @@ use super::mysql::{ list_indexes, mysql_keyword_at, quote_identifier, show_create_table_ddl, skip_mysql_quoted, MySqlPool, }; +pub fn uses_show_metadata(config: &ConnectionConfig) -> bool { + super::doris::is_config(config) || super::starrocks::is_config(config) || super::manticoresearch::is_config(config) +} + +pub fn supports_external_catalogs(config: &ConnectionConfig) -> bool { + super::doris::is_config(config) || super::starrocks::is_config(config) +} + // Doris and StarRocks reuse the MySQL wire protocol, but catalog addressing and DDL/index // metadata semantics are MySQL-compatible distributed behavior and stay isolated in this module. pub async fn list_indexes_with_ddl_fallback( diff --git a/crates/dbx-core/src/db/oceanbase_mysql.rs b/crates/dbx-core/src/db/oceanbase_mysql.rs new file mode 100644 index 0000000000..aebf229421 --- /dev/null +++ b/crates/dbx-core/src/db/oceanbase_mysql.rs @@ -0,0 +1,56 @@ +use crate::models::connection::{ConnectionConfig, DatabaseType}; + +pub const DRIVER_PROFILE: &str = "oceanbase"; + +pub fn is_config(config: &ConnectionConfig) -> bool { + is_profile(&config.db_type, config.driver_profile.as_deref()) +} + +pub fn is_profile(db_type: &DatabaseType, driver_profile: Option<&str>) -> bool { + *db_type == DatabaseType::Mysql + && driver_profile.is_some_and(|profile| profile.eq_ignore_ascii_case(DRIVER_PROFILE)) +} + +pub fn query_timeout_sql(config: &ConnectionConfig, timeout_secs: u64) -> Option { + if !is_config(config) || timeout_secs == 0 { + return None; + } + Some(format!("SET ob_query_timeout = {}", timeout_secs.saturating_mul(1_000_000))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn profile_is_isolated_from_oracle_mode_and_standard_mysql() { + assert!(is_profile(&DatabaseType::Mysql, Some("OceanBase"))); + assert!(!is_profile(&DatabaseType::Mysql, None)); + assert!(!is_profile(&DatabaseType::OceanbaseOracle, Some("oceanbase"))); + } + + #[test] + fn query_timeout_uses_microseconds_and_saturates() { + let mut config = test_config(); + assert_eq!(query_timeout_sql(&config, 30), Some("SET ob_query_timeout = 30000000".to_string())); + assert_eq!(query_timeout_sql(&config, u64::MAX), Some(format!("SET ob_query_timeout = {}", u64::MAX))); + + config.driver_profile = None; + assert_eq!(query_timeout_sql(&config, 30), None); + } + + fn test_config() -> ConnectionConfig { + serde_json::from_value(serde_json::json!({ + "id": "oceanbase", + "name": "OceanBase", + "db_type": "mysql", + "driver_profile": "oceanbase", + "host": "127.0.0.1", + "port": 2883, + "username": "root", + "password": "", + "database": "test" + })) + .unwrap() + } +} diff --git a/crates/dbx-core/src/db/starrocks.rs b/crates/dbx-core/src/db/starrocks.rs index 627e8416d8..fd179c9074 100644 --- a/crates/dbx-core/src/db/starrocks.rs +++ b/crates/dbx-core/src/db/starrocks.rs @@ -1,5 +1,209 @@ +use mysql_async::prelude::*; +use std::collections::HashSet; + +use crate::models::connection::{ConnectionConfig, DatabaseType}; +use crate::types::{ObjectInfo, TableInfo}; + +use super::mysql::{ + get_conn_with_timeout, get_str_by_name, list_routine_objects, list_tables_show_with_status, quote_value, + table_infos_to_objects, MySqlPool, TableStatusMeta, +}; + pub use super::mysql_compatible::{ get_columns_show_from as get_catalog_columns, list_catalog_indexes, list_catalogs, list_databases_show_from as list_catalog_databases, list_indexes_with_ddl_fallback as list_indexes, list_tables_show_from as list_catalog_tables, show_create_table_ddl_from as get_catalog_table_ddl, }; + +pub fn is_config(config: &ConnectionConfig) -> bool { + is_profile(&config.db_type, config.driver_profile.as_deref()) +} + +pub fn is_profile(db_type: &DatabaseType, driver_profile: Option<&str>) -> bool { + *db_type == DatabaseType::StarRocks + || (*db_type == DatabaseType::Mysql + && driver_profile.is_some_and(|profile| profile.eq_ignore_ascii_case("starrocks"))) +} + +fn materialized_views_sql(database: &str) -> String { + format!( + "SELECT TABLE_NAME FROM information_schema.materialized_views WHERE TABLE_SCHEMA = {}", + quote_value(database) + ) +} + +pub(crate) fn materialized_view_definition_sql(database: &str, name: &str) -> String { + format!( + "SELECT MATERIALIZED_VIEW_DEFINITION \ + FROM information_schema.materialized_views \ + WHERE TABLE_SCHEMA = {} AND TABLE_NAME = {} \ + LIMIT 1", + quote_value(database), + quote_value(name) + ) +} + +async fn list_materialized_view_names(pool: &MySqlPool, database: &str) -> Result, String> { + let sql = materialized_views_sql(database); + let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?; + let result = conn.query_iter(&sql).await.map_err(|error| error.to_string())?; + let rows: Vec = result.collect_and_drop().await.map_err(|error| error.to_string())?; + Ok(rows + .iter() + .filter_map(|row| { + let name = get_str_by_name(row, "TABLE_NAME").trim().to_string(); + (!name.is_empty()).then_some(name) + }) + .collect()) +} + +fn merge_materialized_views( + tables: &mut Vec, + materialized_view_names: Result, String>, + database: &str, +) { + let materialized_view_names = match materialized_view_names { + Ok(names) => names, + Err(error) => { + log::warn!("Skipping materialized view classification for StarRocks database `{database}`: {error}"); + return; + } + }; + + let known_names: HashSet = tables.iter().map(|table| table.name.clone()).collect(); + for table in tables.iter_mut() { + if materialized_view_names.contains(&table.name) { + table.table_type = "MATERIALIZED_VIEW".to_string(); + } + } + + let mut sorted_names: Vec<&String> = materialized_view_names.iter().collect(); + sorted_names.sort(); + for name in sorted_names { + if !known_names.contains(name.as_str()) { + tables.push(TableInfo { + name: name.clone(), + table_type: "MATERIALIZED_VIEW".to_string(), + comment: None, + parent_schema: None, + parent_name: None, + }); + } + } +} + +async fn list_tables_with_status( + pool: &MySqlPool, + database: &str, +) -> Result<(Vec, std::collections::HashMap), String> { + let (tables, materialized_view_names) = + tokio::join!(list_tables_show_with_status(pool, database), list_materialized_view_names(pool, database)); + let (mut tables, status) = tables?; + merge_materialized_views(&mut tables, materialized_view_names, database); + Ok((tables, status)) +} + +pub async fn list_tables(pool: &MySqlPool, database: &str) -> Result, String> { + list_tables_with_status(pool, database).await.map(|(tables, _)| tables) +} + +pub async fn list_table_objects(pool: &MySqlPool, database: &str) -> Result, String> { + let (tables, routines) = + tokio::join!(list_tables_with_status(pool, database), list_routine_objects(pool, database)); + let (tables, status) = tables?; + let mut objects = table_infos_to_objects(tables, &status, database); + + match routines { + Ok(routines) => objects.extend(routines), + Err(error) => log::warn!("Skipping routines for database `{database}` in object browser: {error}"), + } + + Ok(objects) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn profile_matches_starrocks_only() { + assert!(is_profile(&DatabaseType::StarRocks, None)); + assert!(is_profile(&DatabaseType::Mysql, Some("STARROCKS"))); + assert!(!is_profile(&DatabaseType::Mysql, Some("doris"))); + assert!(!is_profile(&DatabaseType::Postgres, Some("starrocks"))); + } + + #[test] + fn materialized_views_are_classified_without_duplicates() { + let mut tables = vec![table("orders", "BASE TABLE"), table("orders_view", "VIEW"), table("orders_mv", "VIEW")]; + + merge_materialized_views(&mut tables, Ok(HashSet::from(["orders_mv".to_string()])), "analytics"); + + assert_eq!( + tables.iter().map(|table| (table.name.as_str(), table.table_type.as_str())).collect::>(), + vec![("orders", "BASE TABLE"), ("orders_view", "VIEW"), ("orders_mv", "MATERIALIZED_VIEW")] + ); + } + + #[test] + fn missing_materialized_views_are_appended_deterministically() { + let mut tables = vec![table("orders", "BASE TABLE")]; + let names = HashSet::from(["orders_mv".to_string(), "daily_orders_mv".to_string()]); + + merge_materialized_views(&mut tables, Ok(names), "analytics"); + + assert_eq!( + tables.iter().map(|table| (table.name.as_str(), table.table_type.as_str())).collect::>(), + vec![ + ("orders", "BASE TABLE"), + ("daily_orders_mv", "MATERIALIZED_VIEW"), + ("orders_mv", "MATERIALIZED_VIEW"), + ] + ); + } + + #[test] + fn lookup_failure_preserves_base_table_types() { + let mut tables = vec![table("orders_mv", "VIEW")]; + + merge_materialized_views(&mut tables, Err("permission denied".to_string()), "analytics"); + + assert_eq!(tables[0].table_type, "VIEW"); + } + + #[test] + fn materialized_view_queries_escape_database_and_name() { + assert_eq!( + materialized_views_sql("tenant's analytics"), + "SELECT TABLE_NAME FROM information_schema.materialized_views WHERE TABLE_SCHEMA = 'tenant\\'s analytics'" + ); + assert_eq!( + materialized_view_definition_sql("tenant's analytics", "weird'name"), + "SELECT MATERIALIZED_VIEW_DEFINITION FROM information_schema.materialized_views WHERE TABLE_SCHEMA = 'tenant\\'s analytics' AND TABLE_NAME = 'weird\\'name' LIMIT 1" + ); + } + + #[test] + fn object_conversion_preserves_materialized_view_type() { + let objects = table_infos_to_objects( + vec![table("orders", "BASE TABLE"), table("orders_view", "VIEW"), table("orders_mv", "MATERIALIZED_VIEW")], + &std::collections::HashMap::new(), + "analytics", + ); + + assert_eq!( + objects.iter().map(|object| (object.name.as_str(), object.object_type.as_str())).collect::>(), + vec![("orders", "TABLE"), ("orders_view", "VIEW"), ("orders_mv", "MATERIALIZED_VIEW")] + ); + } + + fn table(name: &str, table_type: &str) -> TableInfo { + TableInfo { + name: name.to_string(), + table_type: table_type.to_string(), + comment: None, + parent_schema: None, + parent_name: None, + } + } +} diff --git a/crates/dbx-core/src/db/tdsql_mysql.rs b/crates/dbx-core/src/db/tdsql_mysql.rs new file mode 100644 index 0000000000..3dab4815db --- /dev/null +++ b/crates/dbx-core/src/db/tdsql_mysql.rs @@ -0,0 +1,54 @@ +use crate::models::connection::{ConnectionConfig, DatabaseType}; + +pub const DRIVER_PROFILE: &str = "tdsql"; + +pub fn is_config(config: &ConnectionConfig) -> bool { + is_profile(&config.db_type, config.driver_profile.as_deref()) +} + +pub fn is_profile(db_type: &DatabaseType, driver_profile: Option<&str>) -> bool { + *db_type == DatabaseType::Mysql + && driver_profile.is_some_and(|profile| profile.eq_ignore_ascii_case(DRIVER_PROFILE)) +} + +pub(crate) fn preserves_proxy_directive_for_database_type(db_type: DatabaseType) -> bool { + db_type == DatabaseType::Mysql +} + +pub(crate) fn proxy_directive_start(statement: &str, executable_start: usize) -> Option { + let prefix = statement.get(..executable_start)?.trim_end(); + prefix.strip_suffix("/*proxy*/").map(|before| before.len()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn profile_is_isolated_from_standard_mysql() { + assert!(is_profile(&DatabaseType::Mysql, Some("TDSQL"))); + assert!(!is_profile(&DatabaseType::Mysql, None)); + assert!(!is_profile(&DatabaseType::Postgres, Some("tdsql"))); + } + + #[test] + fn proxy_directive_is_available_only_to_mysql_parsing() { + assert!(preserves_proxy_directive_for_database_type(DatabaseType::Mysql)); + assert!(!preserves_proxy_directive_for_database_type(DatabaseType::Postgres)); + } + + #[test] + fn proxy_directive_must_be_exact_and_adjacent_to_executable_sql() { + let exact = "/*proxy*/ \n\tSHOW PROXY STATUS"; + let executable_start = exact.find("SHOW").unwrap(); + assert_eq!(proxy_directive_start(exact, executable_start), Some(0)); + + let ordinary = "/* proxy */\nSHOW PROXY STATUS"; + let executable_start = ordinary.find("SHOW").unwrap(); + assert_eq!(proxy_directive_start(ordinary, executable_start), None); + + let separated = "/*proxy*/\n/* audit */\nSHOW PROXY STATUS"; + let executable_start = separated.find("SHOW").unwrap(); + assert_eq!(proxy_directive_start(separated, executable_start), None); + } +} diff --git a/crates/dbx-core/src/db/tidb.rs b/crates/dbx-core/src/db/tidb.rs new file mode 100644 index 0000000000..d1bf969656 --- /dev/null +++ b/crates/dbx-core/src/db/tidb.rs @@ -0,0 +1,24 @@ +use crate::models::connection::{ConnectionConfig, DatabaseType}; + +pub const DRIVER_PROFILE: &str = "tidb"; + +pub fn is_config(config: &ConnectionConfig) -> bool { + is_profile(&config.db_type, config.driver_profile.as_deref()) +} + +pub fn is_profile(db_type: &DatabaseType, driver_profile: Option<&str>) -> bool { + *db_type == DatabaseType::Mysql + && driver_profile.is_some_and(|profile| profile.eq_ignore_ascii_case(DRIVER_PROFILE)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn profile_is_isolated_from_standard_mysql() { + assert!(is_profile(&DatabaseType::Mysql, Some("TiDB"))); + assert!(!is_profile(&DatabaseType::Mysql, None)); + assert!(!is_profile(&DatabaseType::Postgres, Some("tidb"))); + } +} diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index d1593740ad..09f4ffec83 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -1558,7 +1558,7 @@ async fn configured_operation_budget_for_pool_key(state: &AppState, pool_key: &s fn oceanbase_mysql_session_timeout_sql(config: Option<&ConnectionConfig>, timeout_secs: Option) -> Option { let config = config?; let timeout_secs = timeout_secs.unwrap_or(config.query_timeout_secs); - crate::connection::oceanbase_mysql_query_timeout_sql(config, timeout_secs) + crate::db::oceanbase_mysql::query_timeout_sql(config, timeout_secs) } async fn apply_oceanbase_mysql_session_timeout( diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 637dece85c..0c27f2a182 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -365,7 +365,7 @@ pub async fn list_doris_catalogs_core(state: &AppState, connection_id: &str) -> let db_config = connection_config(state, connection_id).await; let connections = state.connections.read().await; if let Some(PoolKind::Mysql(p, _)) = connections.get(&pool_key) { - return if db_config.as_ref().is_some_and(is_starrocks_config) { + return if db_config.as_ref().is_some_and(db::starrocks::is_config) { db::starrocks::list_catalogs(p).await } else { db::doris::list_catalogs(p).await @@ -391,7 +391,7 @@ pub async fn list_doris_catalog_databases_core( let PoolKind::Mysql(p, _) = pool else { return Ok(vec![]); }; - let databases = if db_config.as_ref().is_some_and(is_starrocks_config) { + let databases = if db_config.as_ref().is_some_and(db::starrocks::is_config) { db::starrocks::list_catalog_databases(p, catalog).await } else { db::doris::list_catalog_databases(p, catalog).await @@ -435,7 +435,7 @@ pub async fn list_doris_catalog_tables_core( let PoolKind::Mysql(p, _) = pool else { return Ok(vec![]); }; - let tables = if db_config.as_ref().is_some_and(is_starrocks_config) { + let tables = if db_config.as_ref().is_some_and(db::starrocks::is_config) { db::starrocks::list_catalog_tables(p, catalog, database).await } else { db::doris::list_catalog_tables(p, catalog, database).await @@ -458,7 +458,7 @@ pub async fn get_doris_catalog_columns_core( let PoolKind::Mysql(p, _) = pool else { return Ok(vec![]); }; - let columns = if db_config.as_ref().is_some_and(is_starrocks_config) { + let columns = if db_config.as_ref().is_some_and(db::starrocks::is_config) { db::starrocks::get_catalog_columns(p, catalog, database, table).await } else { db::doris::get_catalog_columns(p, catalog, database, table).await @@ -481,7 +481,7 @@ pub async fn get_doris_catalog_table_ddl_core( let PoolKind::Mysql(p, _) = pool else { return Err("DDL not supported for this connection".to_string()); }; - if db_config.as_ref().is_some_and(is_starrocks_config) { + if db_config.as_ref().is_some_and(db::starrocks::is_config) { db::starrocks::get_catalog_table_ddl(p, catalog, database, table).await } else { db::doris::get_catalog_table_ddl(p, catalog, database, table).await @@ -503,7 +503,7 @@ pub async fn list_doris_catalog_indexes_core( let PoolKind::Mysql(p, _) = pool else { return Ok(vec![]); }; - if db_config.as_ref().is_some_and(is_starrocks_config) { + if db_config.as_ref().is_some_and(db::starrocks::is_config) { db::starrocks::list_catalog_indexes(p, catalog, database, table).await } else { db::doris::list_catalog_indexes(p, catalog, database, table).await @@ -559,7 +559,7 @@ pub async fn resolve_external_doris_catalog( return None; } let config = connection_config(state, connection_id).await?; - if is_doris_family_catalog_capable_config(&config) { + if db::mysql_compatible::supports_external_catalogs(&config) { Some(catalog.to_string()) } else { None @@ -620,7 +620,7 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result { + PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(db::mysql_compatible::uses_show_metadata) => { db::mysql::list_databases_show(p) .await .map(|databases| filter_mysql_system_databases_for_config(databases, db_config.as_ref())) @@ -644,7 +644,10 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result Result, String> { let config = connection_config(state, connection_id).await; - if config.as_ref().is_some_and(|config| db::dolt::is_config(config) || is_doris_family_config(config)) { + if config + .as_ref() + .is_some_and(|config| db::dolt::is_config(config) || db::mysql_compatible::uses_show_metadata(config)) + { return list_databases_once(state, connection_id).await; } let connections = state.connections.read().await; @@ -1080,8 +1083,8 @@ async fn get_table_comment_core_for_session( match pool { PoolKind::Mysql(p, mode) if *mode != MysqlMode::OceanBaseOracle - && !db_config.as_ref().is_some_and(is_doris_family_config) - && !db_config.as_ref().is_some_and(is_manticoresearch_config) => + && !db_config.as_ref().is_some_and(db::mysql_compatible::uses_show_metadata) + && !db_config.as_ref().is_some_and(db::manticoresearch::is_config) => { db::mysql::get_table_comment(p, database, table).await } @@ -2149,12 +2152,12 @@ async fn list_tables_once( let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { - PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(is_starrocks_config) => { - db::mysql::list_starrocks_tables(p, database) + PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(db::starrocks::is_config) => { + db::starrocks::list_tables(p, database) .await .map(|tables| filter_table_infos(tables, filter, limit, offset, object_types, table_name_filter)) } - PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(is_doris_family_config) => { + PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(db::mysql_compatible::uses_show_metadata) => { db::mysql::list_tables_show(p, database) .await .map(|tables| filter_table_infos(tables, filter, limit, offset, object_types, table_name_filter)) @@ -4542,32 +4545,34 @@ for line in sys.stdin: #[test] fn doris_family_catalog_capable_matches_doris_and_starrocks_only() { // Doris and StarRocks expose multi-catalog federation. - assert!(super::is_doris_family_catalog_capable_config(&test_connection_config(DatabaseType::Doris))); - assert!(super::is_doris_family_catalog_capable_config(&test_connection_config(DatabaseType::StarRocks))); + assert!(db::mysql_compatible::supports_external_catalogs(&test_connection_config(DatabaseType::Doris))); + assert!(db::mysql_compatible::supports_external_catalogs(&test_connection_config(DatabaseType::StarRocks))); // Driver profiles for Doris/SelectDB/StarRocks also qualify. let mut doris = test_connection_config(DatabaseType::Mysql); doris.driver_profile = Some("doris".to_string()); - assert!(super::is_doris_family_catalog_capable_config(&doris)); + assert!(db::mysql_compatible::supports_external_catalogs(&doris)); let mut selectdb = test_connection_config(DatabaseType::Mysql); selectdb.driver_profile = Some("selectdb".to_string()); - assert!(super::is_doris_family_catalog_capable_config(&selectdb)); + assert!(db::mysql_compatible::supports_external_catalogs(&selectdb)); let mut starrocks = test_connection_config(DatabaseType::Mysql); starrocks.driver_profile = Some("starrocks".to_string()); - assert!(super::is_doris_family_catalog_capable_config(&starrocks)); + assert!(db::mysql_compatible::supports_external_catalogs(&starrocks)); // ManticoreSearch shares the MySQL code path but has no catalog concept. - assert!(!super::is_doris_family_catalog_capable_config(&test_connection_config(DatabaseType::ManticoreSearch))); + assert!(!db::mysql_compatible::supports_external_catalogs(&test_connection_config( + DatabaseType::ManticoreSearch + ))); let mut manticore = test_connection_config(DatabaseType::Mysql); manticore.driver_profile = Some("manticoresearch".to_string()); - assert!(!super::is_doris_family_catalog_capable_config(&manticore)); + assert!(!db::mysql_compatible::supports_external_catalogs(&manticore)); // Plain MySQL / Postgres are not catalog-capable. - assert!(!super::is_doris_family_catalog_capable_config(&test_connection_config(DatabaseType::Mysql))); - assert!(!super::is_doris_family_catalog_capable_config(&test_connection_config(DatabaseType::Postgres))); + assert!(!db::mysql_compatible::supports_external_catalogs(&test_connection_config(DatabaseType::Mysql))); + assert!(!db::mysql_compatible::supports_external_catalogs(&test_connection_config(DatabaseType::Postgres))); } } @@ -4991,7 +4996,7 @@ async fn list_object_statistics_once( let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { PoolKind::Mysql(p, mode) => { - if *mode == MysqlMode::OceanBaseOracle || db_config.as_ref().is_some_and(is_manticoresearch_config) { + if *mode == MysqlMode::OceanBaseOracle || db_config.as_ref().is_some_and(db::manticoresearch::is_config) { Ok(vec![]) } else { db::mysql::list_object_statistics(p, database).await @@ -5191,11 +5196,11 @@ async fn list_objects_once( // Note: mysql and ob_oracle take different second args (database vs schema) if *mode == MysqlMode::OceanBaseOracle { db::ob_oracle::list_objects(p, schema).await.map(unpaged_object_list) - } else if db_config.as_ref().is_some_and(is_manticoresearch_config) { + } else if db_config.as_ref().is_some_and(db::manticoresearch::is_config) { db::manticoresearch::list_objects(p, database).await.map(unpaged_object_list) - } else if db_config.as_ref().is_some_and(is_starrocks_config) { - db::mysql::list_starrocks_table_objects(p, database).await.map(unpaged_object_list) - } else if db_config.as_ref().is_some_and(is_doris_family_config) { + } else if db_config.as_ref().is_some_and(db::starrocks::is_config) { + db::starrocks::list_table_objects(p, database).await.map(unpaged_object_list) + } else if db_config.as_ref().is_some_and(db::mysql_compatible::uses_show_metadata) { db::mysql::list_table_objects_show(p, database).await.map(unpaged_object_list) } else { db::mysql::list_objects(p, database, object_types, mysql_limit, mysql_offset) @@ -5834,11 +5839,11 @@ async fn get_columns_core_for_session_inner( let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { - PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(is_manticoresearch_config) => { + PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(db::manticoresearch::is_config) => { let metadata_database = mysql_show_metadata_database_for_config(db_config.as_ref(), database); db::manticoresearch::get_columns(p, metadata_database, table).await.map(deduplicate_column_infos) } - PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(is_doris_family_config) => { + PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(db::mysql_compatible::uses_show_metadata) => { let metadata_database = mysql_show_metadata_database_for_config(db_config.as_ref(), database); // Doris/StarRocks previously went straight to `SHOW COLUMNS` for // speed (see perf(doris) commit), but `SHOW COLUMNS` reports the @@ -6035,14 +6040,14 @@ async fn list_indexes_core_for_session( match pool { PoolKind::Mysql(p, mode) => { - if db_config.as_ref().is_some_and(is_manticoresearch_config) { + if db_config.as_ref().is_some_and(db::manticoresearch::is_config) { return db::manticoresearch::list_indexes(p, table).await; } if *mode == MysqlMode::OceanBaseOracle { db::ob_oracle::list_indexes(p, schema, table).await - } else if db_config.as_ref().is_some_and(is_starrocks_config) { + } else if db_config.as_ref().is_some_and(db::starrocks::is_config) { db::starrocks::list_indexes(p, mysql_table_metadata_catalog(database, schema), table).await - } else if db_config.as_ref().is_some_and(is_doris_config) { + } else if db_config.as_ref().is_some_and(db::doris::is_config) { db::doris::list_indexes(p, mysql_table_metadata_catalog(database, schema), table).await } else { db::mysql::list_indexes(p, mysql_table_metadata_catalog(database, schema), table).await @@ -6693,34 +6698,8 @@ fn agent_paging_likely_applied(enabled: bool, limit: Option, returned_len enabled && limit.is_some_and(|limit| returned_len <= limit) } -fn is_doris_family_config(config: &ConnectionConfig) -> bool { - matches!(config.db_type, DatabaseType::Doris | DatabaseType::StarRocks | DatabaseType::ManticoreSearch) - || matches!(config.driver_profile.as_deref(), Some("doris" | "selectdb" | "starrocks" | "manticoresearch")) -} - -fn is_doris_config(config: &ConnectionConfig) -> bool { - config.db_type == DatabaseType::Doris || matches!(config.driver_profile.as_deref(), Some("doris" | "selectdb")) -} - -fn is_starrocks_config(config: &ConnectionConfig) -> bool { - config.db_type == DatabaseType::StarRocks || matches!(config.driver_profile.as_deref(), Some("starrocks")) -} - -/// Doris-family engines that support multi-catalog federation (`SHOW CATALOGS`). -/// Manticore Search is excluded — it shares the MySQL code path but has no -/// catalog concept. -pub fn is_doris_family_catalog_capable_config(config: &ConnectionConfig) -> bool { - matches!(config.db_type, DatabaseType::Doris | DatabaseType::StarRocks) - || matches!(config.driver_profile.as_deref(), Some("doris" | "selectdb" | "starrocks")) -} - -fn is_manticoresearch_config(config: &ConnectionConfig) -> bool { - matches!(config.db_type, DatabaseType::ManticoreSearch) - || matches!(config.driver_profile.as_deref(), Some("manticoresearch")) -} - fn mysql_show_metadata_database_for_config<'a>(config: Option<&ConnectionConfig>, database: &'a str) -> &'a str { - if config.is_some_and(is_manticoresearch_config) { + if config.is_some_and(db::manticoresearch::is_config) { "" } else { database @@ -6731,7 +6710,7 @@ fn filter_mysql_system_databases_for_config( databases: Vec, config: Option<&ConnectionConfig>, ) -> Vec { - if !config.is_some_and(is_manticoresearch_config) { + if !config.is_some_and(db::manticoresearch::is_config) { return databases; } @@ -7308,7 +7287,7 @@ async fn mysql_object_source( // sync MVs. Fall back to the persistent definition exposed by // information_schema.materialized_views. The fallback returns a single // column (MATERIALIZED_VIEW_DEFINITION) so the column index is always 0. - let fallback_sql = db::mysql::mysql_materialized_view_definition_sql(database, name); + let fallback_sql = db::starrocks::materialized_view_definition_sql(database, name); read_mysql_object_source_row(&mut conn, &fallback_sql, 0).await.map_err(|fallback_err| { format!( "SHOW CREATE MATERIALIZED VIEW failed ({primary_err}); \ diff --git a/crates/dbx-core/src/schema/providers/native.rs b/crates/dbx-core/src/schema/providers/native.rs index 0a128869e1..71bf5b55d3 100644 --- a/crates/dbx-core/src/schema/providers/native.rs +++ b/crates/dbx-core/src/schema/providers/native.rs @@ -7,7 +7,7 @@ pub(in crate::schema) async fn list_databases( config: Option<&ConnectionConfig>, ) -> Result, String> { match pool { - PoolKind::Mysql(p, _) if config.is_some_and(is_doris_family_config) => db::mysql::list_databases_show(p) + PoolKind::Mysql(p, _) if config.is_some_and(db::mysql_compatible::uses_show_metadata) => db::mysql::list_databases_show(p) .await .map(|databases| filter_mysql_system_databases_for_config(databases, config)), PoolKind::Mysql(p, mode) if *mode == MysqlMode::OceanBaseOracle => db::ob_oracle::list_databases(p).await, @@ -36,7 +36,7 @@ pub(in crate::schema) async fn list_tables( schema: &str, ) -> Result, String> { match pool { - PoolKind::Mysql(p, _) if config.is_some_and(is_doris_family_config) => { + PoolKind::Mysql(p, _) if config.is_some_and(db::mysql_compatible::uses_show_metadata) => { db::mysql::list_tables_show(p, database).await } PoolKind::Mysql(p, mode) if *mode == MysqlMode::OceanBaseOracle => { @@ -85,10 +85,10 @@ pub(in crate::schema) async fn list_objects( PoolKind::Mysql(p, mode) if *mode == MysqlMode::OceanBaseOracle => { db::ob_oracle::list_objects(p, schema).await.map(Some) } - PoolKind::Mysql(p, _) if config.is_some_and(is_manticoresearch_config) => { + PoolKind::Mysql(p, _) if config.is_some_and(db::manticoresearch::is_config) => { db::manticoresearch::list_objects(p, database).await.map(Some) } - PoolKind::Mysql(p, _) if config.is_some_and(is_doris_family_config) => { + PoolKind::Mysql(p, _) if config.is_some_and(db::mysql_compatible::uses_show_metadata) => { db::mysql::list_table_objects_show(p, database).await.map(Some) } PoolKind::Mysql(p, _) => { @@ -137,11 +137,11 @@ pub(in crate::schema) async fn get_columns( table: &str, ) -> Result, String> { match pool { - PoolKind::Mysql(p, _) if config.is_some_and(is_manticoresearch_config) => { + PoolKind::Mysql(p, _) if config.is_some_and(db::manticoresearch::is_config) => { let metadata_database = mysql_show_metadata_database_for_config(config, database); db::manticoresearch::get_columns(p, metadata_database, table).await } - PoolKind::Mysql(p, _) if config.is_some_and(is_doris_family_config) => { + PoolKind::Mysql(p, _) if config.is_some_and(db::mysql_compatible::uses_show_metadata) => { let metadata_database = mysql_show_metadata_database_for_config(config, database); db::mysql::get_columns(p, metadata_database, table).await } @@ -176,13 +176,13 @@ pub(in crate::schema) async fn list_indexes( PoolKind::Mysql(p, mode) if *mode == MysqlMode::OceanBaseOracle => { db::ob_oracle::list_indexes(p, schema, table).await } - PoolKind::Mysql(p, _) if config.is_some_and(is_starrocks_config) => { + PoolKind::Mysql(p, _) if config.is_some_and(db::starrocks::is_config) => { db::starrocks::list_indexes(p, database, table).await } - PoolKind::Mysql(p, _) if config.is_some_and(is_doris_config) => { + PoolKind::Mysql(p, _) if config.is_some_and(db::doris::is_config) => { db::doris::list_indexes(p, database, table).await } - PoolKind::Mysql(p, _) if config.is_some_and(is_manticoresearch_config) => { + PoolKind::Mysql(p, _) if config.is_some_and(db::manticoresearch::is_config) => { db::mysql_compatible::list_indexes_with_ddl_fallback(p, database, table).await } PoolKind::Mysql(p, _) => db::mysql::list_indexes(p, schema, table).await, @@ -343,26 +343,8 @@ fn is_cloudberry_config(config: &ConnectionConfig) -> bool { matches!(config.driver_profile.as_deref(), Some("cloudberry")) } -fn is_doris_family_config(config: &ConnectionConfig) -> bool { - matches!(config.db_type, DatabaseType::Doris | DatabaseType::StarRocks | DatabaseType::ManticoreSearch) - || matches!(config.driver_profile.as_deref(), Some("doris" | "selectdb" | "starrocks" | "manticoresearch")) -} - -fn is_doris_config(config: &ConnectionConfig) -> bool { - config.db_type == DatabaseType::Doris || matches!(config.driver_profile.as_deref(), Some("doris" | "selectdb")) -} - -fn is_starrocks_config(config: &ConnectionConfig) -> bool { - config.db_type == DatabaseType::StarRocks || matches!(config.driver_profile.as_deref(), Some("starrocks")) -} - -fn is_manticoresearch_config(config: &ConnectionConfig) -> bool { - matches!(config.db_type, DatabaseType::ManticoreSearch) - || matches!(config.driver_profile.as_deref(), Some("manticoresearch")) -} - fn mysql_show_metadata_database_for_config<'a>(config: Option<&ConnectionConfig>, database: &'a str) -> &'a str { - if config.is_some_and(is_manticoresearch_config) { + if config.is_some_and(db::manticoresearch::is_config) { "" } else { database @@ -373,7 +355,7 @@ fn filter_mysql_system_databases_for_config( databases: Vec, config: Option<&ConnectionConfig>, ) -> Vec { - if !config.is_some_and(is_manticoresearch_config) { + if !config.is_some_and(db::manticoresearch::is_config) { return databases; } diff --git a/crates/dbx-core/src/sql.rs b/crates/dbx-core/src/sql.rs index 73298b4ebd..efe517be5d 100644 --- a/crates/dbx-core/src/sql.rs +++ b/crates/dbx-core/src/sql.rs @@ -143,7 +143,12 @@ impl Default for SqlDialectProfile { impl SqlDialectProfile { fn for_database_type(db_type: DatabaseType) -> Self { if matches!(db_type, DatabaseType::Mysql) { - return Self::mysql(); + return Self { + preserves_tdsql_proxy_directive: crate::db::tdsql_mysql::preserves_proxy_directive_for_database_type( + db_type, + ), + ..Self::mysql_compatible() + }; } if matches!(db_type, DatabaseType::Gaussdb) { @@ -178,10 +183,6 @@ impl SqlDialectProfile { } } - fn mysql() -> Self { - Self { preserves_tdsql_proxy_directive: true, ..Self::mysql_compatible() } - } - fn oracle_like() -> Self { Self { supports_oracle_plsql_blocks: true, supports_slash_line_block_delimiter: true, ..Self::default() } } @@ -2530,18 +2531,13 @@ fn executable_sql_bounds(statement: &str, options: SqlParsingOptions) -> Option< } let executable_start = trimmed.len() - executable.len(); let start = if options.profile.preserves_tdsql_proxy_directive { - tdsql_proxy_directive_start(trimmed, executable_start).unwrap_or(executable_start) + crate::db::tdsql_mysql::proxy_directive_start(trimmed, executable_start).unwrap_or(executable_start) } else { executable_start }; Some((start, trimmed_end)) } -fn tdsql_proxy_directive_start(statement: &str, executable_start: usize) -> Option { - let prefix = statement.get(..executable_start)?.trim_end(); - prefix.strip_suffix("/*proxy*/").map(|before| before.len()) -} - fn has_executable_sql_with_options(statement: &str, options: SqlParsingOptions) -> bool { let chars = statement.chars().collect::>(); let mut in_line_comment = false; @@ -2700,7 +2696,7 @@ mod tests { #[test] fn dash_dash_requires_space_only_for_mysql_compatible_profiles() { - let mysql_profile = SqlDialectProfile::mysql(); + let mysql_profile = SqlDialectProfile::for_database_type(DatabaseType::Mysql); assert!(dash_dash_starts_line_comment(mysql_profile, Some(' '))); assert!(dash_dash_starts_line_comment(mysql_profile, Some('\n'))); assert!(dash_dash_starts_line_comment(mysql_profile, Some('\u{1}'))); @@ -3391,7 +3387,6 @@ SELECT 2;"; assert!(!default.preserves_tdsql_proxy_directive); let mysql = SqlDialectProfile::for_database_type(DatabaseType::Mysql); - assert_eq!(mysql, SqlDialectProfile::mysql()); assert!(mysql.supports_hash_line_comments); assert!(mysql.supports_mysql_routine_blocks); assert!(mysql.preserves_tdsql_proxy_directive); diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index d6c65d6e41..15edff43c3 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -265,7 +265,7 @@ pub fn resolve_external_transfer_catalog_for_config<'a>( config: &crate::models::connection::ConnectionConfig, ) -> Option<&'a str> { let catalog = normalize_external_catalog_name(catalog)?; - if crate::schema::is_doris_family_catalog_capable_config(config) { + if db::mysql_compatible::supports_external_catalogs(config) { Some(catalog) } else { None @@ -3320,11 +3320,6 @@ fn max_transfer_write_rows(db_type: &DatabaseType, mode: &TransferMode) -> usize } } -fn is_oceanbase_mysql_profile(db_type: &DatabaseType, driver_profile: Option<&str>) -> bool { - matches!(db_type, DatabaseType::Mysql) - && driver_profile.is_some_and(|profile| profile.eq_ignore_ascii_case("oceanbase")) -} - fn contains_oceanbase_mysql_table_options(sql: &str) -> bool { let (sql_without_literals_or_comments, _) = protect_sql_literals(sql, true); OCEANBASE_MYSQL_TABLE_OPTION_RE.is_match(&sql_without_literals_or_comments) @@ -3383,8 +3378,8 @@ fn can_reuse_source_table_ddl( target_driver_profile: Option<&str>, preserves_target_table_name: bool, ) -> bool { - if is_oceanbase_mysql_profile(source_db_type, source_driver_profile) - && !is_oceanbase_mysql_profile(target_db_type, target_driver_profile) + if db::oceanbase_mysql::is_profile(source_db_type, source_driver_profile) + && !db::oceanbase_mysql::is_profile(target_db_type, target_driver_profile) { return false; } @@ -6593,7 +6588,7 @@ where } }; if contains_oceanbase_mysql_table_options(&source_ddl) - && !is_oceanbase_mysql_profile(target_db_type, target_driver_profile.as_deref()) + && !db::oceanbase_mysql::is_profile(target_db_type, target_driver_profile.as_deref()) { generate_create_table_ddl( &columns,