github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-21 08:13:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mysql): isolate compatible database-specific logic

Browse Source
This commit is contained in:
t8y2
2026-08-14 19:09:05 +08:00
parent 1ff24aaeb4
commit 6dcfdc6ae0
15 changed files with 484 additions and 887 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/dbx-core/src/connection.rs
+8 -21
View File
@@ -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<String> {
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<String> {
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<String> {
@@ -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())
);
}
crates/dbx-core/src/db/doris.rs
+27
View File
@@ -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")));
}
}
crates/dbx-core/src/db/manticoresearch.rs
+19
View File
@@ -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`");
crates/dbx-core/src/db/mod.rs
+3
View File
@@ -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;
crates/dbx-core/src/db/mysql.rs
+15 -752
View File
@@ -41,14 +41,12 @@ pub(crate) fn mysql_catalog_dialect(
db_type: DatabaseType,
driver_profile: Option<&str>,
) -> Option<MySqlCatalogDialect> {
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<serde_json::Value>),
}
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<String>,
created_at: Option<String>,
updated_at: Option<String>,
@@ -2889,7 +2884,7 @@ fn mysql_fallback_like_patterns(filter: &str) -> Vec<String> {
patterns
}
async fn list_tables_show_with_status(
pub(super) async fn list_tables_show_with_status(
pool: &MySqlPool,
database: &str,
) -> Result<(Vec<TableInfo>, HashMap<String, TableStatusMeta>), String> {
@@ -2943,116 +2938,6 @@ pub async fn list_tables_show(pool: &MySqlPool, database: &str) -> Result<Vec<Ta
list_tables_show_with_status(pool, database).await.map(|(tables, _)| tables)
}
fn starrocks_materialized_views_sql(database: &str) -> 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<HashSet<String>, 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<mysql_async::Row> = 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<TableInfo>,
materialized_view_names: Result<HashSet<String>, 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<String> = 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<TableInfo>, HashMap<String, TableStatusMeta>), 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<Vec<TableInfo>, 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<Vec<ObjectInfo>, 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<TableInfo>,
status: &HashMap<String, TableStatusMeta>,
database: &str,
@@ -3497,7 +3368,7 @@ fn table_infos_to_objects(
.collect()
}
async fn list_routine_objects(pool: &MySqlPool, database: &str) -> Result<Vec<ObjectInfo>, String> {
pub(super) async fn list_routine_objects(pool: &MySqlPool, database: &str) -> Result<Vec<ObjectInfo>, 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
// (`<catalog>.<database>.<table>`), which the engines accept directly without
// needing to `SWITCH` the session catalog.
// ---------------------------------------------------------------------------
/// Build a 2-part qualified identifier `` `<catalog>`.`<database>` ``.
fn doris_catalog_database_ref(catalog: &str, database: &str) -> String {
format!("{}.{}", quote_identifier(catalog), quote_identifier(database))
}
/// Build a 3-part qualified identifier `` `<catalog>`.`<database>`.`<table>` ``.
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<Vec<crate::db::CatalogInfo>, 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<mysql_async::Row> = result.collect_and_drop().await.map_err(|e| e.to_string())?;
let catalogs: Vec<crate::db::CatalogInfo> = 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<crate::db::CatalogInfo>) -> Vec<crate::db::CatalogInfo> {
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 <catalog>` → databases in the given catalog.
pub async fn list_databases_show_from(pool: &MySqlPool, catalog: &str) -> Result<Vec<DatabaseInfo>, 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<mysql_async::Row> = 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 <catalog>.<database>` → 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<Vec<TableInfo>, 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<mysql_async::Row> = result.collect_and_drop().await.map_err(|e| e.to_string())?;
let mut tables: Vec<TableInfo> = 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 <catalog>.<database>.<table>` → 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<Vec<ColumnInfo>, 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<mysql_async::Row> = 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 <catalog>.<database>.<table>` → DDL for an external
/// catalog table.
pub async fn show_create_table_ddl_from(
pool: &MySqlPool,
catalog: &str,
database: &str,
table: &str,
) -> Result<String, String> {
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<mysql_async::Row> = result.collect_and_drop().await.map_err(|e| e.to_string())?;
let row = rows.first().ok_or("DDL not found")?;
row.get_opt::<String, usize>(1)
.and_then(|result| result.ok())
.or_else(|| {
row.get_opt::<Vec<u8>, 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<Vec<IndexInfo>, 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<IndexInfo> {
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<IndexInfo> {
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<IndexInfo> {
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<String> {
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<String> {
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<String> {
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<Vec<ForeignKeyInfo>, 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<_>>(), 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<_>>(),
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<_>>(),
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<_>>(),
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<_>>(),
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");
crates/dbx-core/src/db/mysql_compatible.rs
+9
View File
@@ -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(
crates/dbx-core/src/db/oceanbase_mysql.rs
+56
View File
@@ -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<String> {
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()
}
}
crates/dbx-core/src/db/starrocks.rs
+204
View File
@@ -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<HashSet<String>, 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<mysql_async::Row> = 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<TableInfo>,
materialized_view_names: Result<HashSet<String>, 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<String> = 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<TableInfo>, std::collections::HashMap<String, TableStatusMeta>), 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<Vec<TableInfo>, String> {
list_tables_with_status(pool, database).await.map(|(tables, _)| tables)
}
pub async fn list_table_objects(pool: &MySqlPool, database: &str) -> Result<Vec<ObjectInfo>, 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<_>>(),
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<_>>(),
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<_>>(),
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,
}
}
}
crates/dbx-core/src/db/tdsql_mysql.rs
+54
View File
@@ -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<usize> {
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);
}
}
crates/dbx-core/src/db/tidb.rs
+24
View File
@@ -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")));
}
}
crates/dbx-core/src/query.rs
+1 -1
View File
@@ -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<u64>) -> Option<String> {
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(
crates/dbx-core/src/schema.rs
+41 -62
View File
@@ -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<Ve
{
db::dolt::list_databases(p).await
}
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_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<Ve
async fn list_database_metadata_once(state: &AppState, connection_id: &str) -> Result<Vec<db::DatabaseInfo>, 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<usize>, 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<db::DatabaseInfo>,
config: Option<&ConnectionConfig>,
) -> Vec<db::DatabaseInfo> {
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}); \
crates/dbx-core/src/schema/providers/native.rs
+11 -29
View File
@@ -7,7 +7,7 @@ pub(in crate::schema) async fn list_databases(
config: Option<&ConnectionConfig>,
) -> Result<Vec<db::DatabaseInfo>, 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<Vec<db::TableInfo>, 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<Vec<db::ColumnInfo>, 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<db::DatabaseInfo>,
config: Option<&ConnectionConfig>,
) -> Vec<db::DatabaseInfo> {
if !config.is_some_and(is_manticoresearch_config) {
if !config.is_some_and(db::manticoresearch::is_config) {
return databases;
}
crates/dbx-core/src/sql.rs
+8 -13
View File
@@ -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<usize> {
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::<Vec<_>>();
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);
crates/dbx-core/src/transfer.rs
+4 -9
View File
@@ -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,