github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-15 14:52:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(doris): show external catalogs in sidebar

Browse Source
This commit is contained in:
hhzorg authored and GitHub committed 2026-07-09 00:45:29 +08:00
1 parent 47d9257ac2
commit d9bb92d1e9
36 files changed
+1680 -174
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/csv_export.rs
+1
View File
@@ -129,6 +129,7 @@ pub async fn export_table_data_csv_core(state: &AppState, options: TableCsvExpor
offset: Some(offset),
where_input: None,
include_row_id: false,
..Default::default()
});
let result = execute_sql_statement_with_options(
state,
crates/dbx-core/src/data_grid_sql.rs
+190 -5
View File
@@ -26,6 +26,8 @@ const DATA_GRID_COLUMN_DISTINCT_VALUES_MAX_LIMIT: usize = 1000;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DataGridTableMeta {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub catalog: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
pub table_name: String,
@@ -162,6 +164,8 @@ pub struct DataGridColumnDistinctValuesSqlOptions {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub database_type: Option<DatabaseType>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub catalog: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
pub table_name: String,
pub column_name: String,
@@ -183,6 +187,8 @@ pub struct DataGridCountSqlOptions {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub database_type: Option<DatabaseType>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub catalog: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
pub table_name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -259,8 +265,9 @@ pub fn build_data_grid_copy_update_statements(options: DataGridCopyUpdateStateme
return Vec::new();
}
let table = qualified_table_name(
let table = crate::sql_dialect::qualified_table_name_with_catalog(
options.database_type,
options.table_meta.catalog.as_deref(),
options.table_meta.schema.as_deref(),
&options.table_meta.table_name,
);
@@ -342,7 +349,14 @@ pub fn build_data_grid_copy_insert_statement(options: DataGridCopyInsertStatemen
let table = options.table_meta.as_ref().map_or_else(
|| "table_name".to_string(),
|meta| qualified_table_name(options.database_type, meta.schema.as_deref(), &meta.table_name),
|meta| {
crate::sql_dialect::qualified_table_name_with_catalog(
options.database_type,
meta.catalog.as_deref(),
meta.schema.as_deref(),
&meta.table_name,
)
},
);
let columns = insert_columns
.iter()
@@ -495,7 +509,12 @@ pub fn build_data_grid_column_distinct_values_sql(options: DataGridColumnDistinc
}
let limit = data_grid_column_distinct_values_limit(options.limit);
let table = qualified_table_name(options.database_type, options.schema.as_deref(), &options.table_name);
let table = crate::sql_dialect::qualified_table_name_with_catalog(
options.database_type,
options.catalog.as_deref(),
options.schema.as_deref(),
&options.table_name,
);
let column = column_filter_ref(options.database_type, &options.column_name);
let mut predicates = Vec::new();
let predicate = crate::sql_dialect::normalize_where_input(options.where_input.as_deref());
@@ -541,7 +560,12 @@ pub fn build_data_grid_column_distinct_values_sql(options: DataGridColumnDistinc
}
pub fn build_data_grid_count_sql(options: DataGridCountSqlOptions) -> String {
let table = qualified_table_name(options.database_type, options.schema.as_deref(), &options.table_name);
let table = crate::sql_dialect::qualified_table_name_with_catalog(
options.database_type,
options.catalog.as_deref(),
options.schema.as_deref(),
&options.table_name,
);
let predicate = crate::sql_dialect::normalize_where_input(options.where_input.as_deref());
let where_clause = if predicate.is_empty() { String::new() } else { format!(" WHERE ({predicate})") };
format!("SELECT COUNT(*) AS cnt FROM {table}{where_clause}")
@@ -747,8 +771,9 @@ fn build_data_grid_save_statements(options: &DataGridSaveStatementOptions) -> Ve
let save_columns = effective_columns(options);
let column_info = options.table_meta.columns.as_deref().unwrap_or(&[]);
let table = qualified_table_name(
let table = crate::sql_dialect::qualified_table_name_with_catalog(
options.database_type,
options.table_meta.catalog.as_deref(),
options.table_meta.schema.as_deref(),
&options.table_meta.table_name,
);
@@ -2050,6 +2075,7 @@ mod tests {
let statements = build_data_grid_copy_update_statements(DataGridCopyUpdateStatementOptions {
database_type: Some(DatabaseType::Postgres),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("public".to_string()),
table_name: "users".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2070,6 +2096,7 @@ mod tests {
let statement = build_data_grid_copy_insert_statement(DataGridCopyInsertStatementOptions {
database_type: Some(DatabaseType::Mysql),
table_meta: Some(DataGridTableMeta {
catalog: None,
schema: None,
table_name: "users".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2092,6 +2119,7 @@ mod tests {
let statement = build_data_grid_copy_insert_statement(DataGridCopyInsertStatementOptions {
database_type: Some(DatabaseType::Mysql),
table_meta: Some(DataGridTableMeta {
catalog: None,
schema: None,
table_name: "users".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2116,6 +2144,7 @@ mod tests {
let statement = build_data_grid_copy_insert_statement(DataGridCopyInsertStatementOptions {
database_type: Some(DatabaseType::Oracle),
table_meta: Some(DataGridTableMeta {
catalog: None,
schema: Some("APP".to_string()),
table_name: "USERS".to_string(),
primary_keys: vec!["ID".to_string()],
@@ -2137,6 +2166,7 @@ mod tests {
#[test]
fn mysql_copy_statements_preserve_blob_hex_literals() {
let table_meta = DataGridTableMeta {
catalog: None,
schema: None,
table_name: "reports".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2200,6 +2230,7 @@ mod tests {
let statement = build_data_grid_copy_insert_statement(DataGridCopyInsertStatementOptions {
database_type: Some(DatabaseType::Postgres),
table_meta: Some(DataGridTableMeta {
catalog: None,
schema: Some("public".to_string()),
table_name: "articles".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2308,6 +2339,7 @@ mod tests {
assert_eq!(
build_data_grid_column_distinct_values_sql(DataGridColumnDistinctValuesSqlOptions {
database_type: Some(DatabaseType::Postgres),
catalog: None,
schema: Some("public".to_string()),
table_name: "users".to_string(),
column_name: "status".to_string(),
@@ -2322,6 +2354,7 @@ mod tests {
assert_eq!(
build_data_grid_column_distinct_values_sql(DataGridColumnDistinctValuesSqlOptions {
database_type: Some(DatabaseType::SqlServer),
catalog: None,
schema: None,
table_name: "users".to_string(),
column_name: "status".to_string(),
@@ -2336,6 +2369,7 @@ mod tests {
assert_eq!(
build_data_grid_column_distinct_values_sql(DataGridColumnDistinctValuesSqlOptions {
database_type: Some(DatabaseType::SqlServer),
catalog: None,
schema: None,
table_name: "users".to_string(),
column_name: "id".to_string(),
@@ -2350,6 +2384,7 @@ mod tests {
assert_eq!(
build_data_grid_column_distinct_values_sql(DataGridColumnDistinctValuesSqlOptions {
database_type: Some(DatabaseType::Oracle),
catalog: None,
schema: Some("APP".to_string()),
table_name: "EVENTS".to_string(),
column_name: "KIND".to_string(),
@@ -2364,6 +2399,7 @@ mod tests {
assert_eq!(
build_data_grid_column_distinct_values_sql(DataGridColumnDistinctValuesSqlOptions {
database_type: Some(DatabaseType::Firebird),
catalog: None,
schema: None,
table_name: "USERS".to_string(),
column_name: "STATUS".to_string(),
@@ -2375,6 +2411,53 @@ mod tests {
}),
"SELECT \"STATUS\" AS dbx_value FROM \"USERS\" WHERE (DELETED_AT IS NULL) GROUP BY \"STATUS\" ORDER BY dbx_value ROWS 25"
);
// Doris / StarRocks external-catalog tables are addressed with a 3-part name.
assert_eq!(
build_data_grid_column_distinct_values_sql(DataGridColumnDistinctValuesSqlOptions {
database_type: Some(DatabaseType::Doris),
catalog: Some("iceberg_catalog".to_string()),
schema: Some("sales".to_string()),
table_name: "orders".to_string(),
column_name: "status".to_string(),
column_info: Some(column("status", "varchar", true, None)),
where_input: None,
search_value: None,
limit: Some(10),
include_counts: false,
}),
"SELECT `status` AS dbx_value FROM `iceberg_catalog`.`sales`.`orders` GROUP BY `status` ORDER BY dbx_value LIMIT 10"
);
assert_eq!(
build_data_grid_column_distinct_values_sql(DataGridColumnDistinctValuesSqlOptions {
database_type: Some(DatabaseType::StarRocks),
catalog: Some("hive_catalog".to_string()),
schema: None,
table_name: "orders".to_string(),
column_name: "status".to_string(),
column_info: Some(column("status", "varchar", true, None)),
where_input: None,
search_value: None,
limit: Some(10),
include_counts: true,
}),
"SELECT `status` AS dbx_value, COUNT(*) AS dbx_count FROM `hive_catalog`.`orders` GROUP BY `status` ORDER BY dbx_count DESC, dbx_value LIMIT 10"
);
// The built-in `internal` catalog is never prefixed.
assert_eq!(
build_data_grid_column_distinct_values_sql(DataGridColumnDistinctValuesSqlOptions {
database_type: Some(DatabaseType::Doris),
catalog: Some("internal".to_string()),
schema: None,
table_name: "orders".to_string(),
column_name: "status".to_string(),
column_info: Some(column("status", "varchar", true, None)),
where_input: None,
search_value: None,
limit: Some(10),
include_counts: false,
}),
"SELECT `status` AS dbx_value FROM `orders` GROUP BY `status` ORDER BY dbx_value LIMIT 10"
);
}
#[test]
@@ -2382,12 +2465,33 @@ mod tests {
assert_eq!(
build_data_grid_count_sql(DataGridCountSqlOptions {
database_type: Some(DatabaseType::Postgres),
catalog: None,
schema: Some("public".to_string()),
table_name: "users".to_string(),
where_input: Some("WHERE active = true;".to_string()),
}),
"SELECT COUNT(*) AS cnt FROM \"public\".\"users\" WHERE (active = true)"
);
assert_eq!(
build_data_grid_count_sql(DataGridCountSqlOptions {
database_type: Some(DatabaseType::Doris),
catalog: Some("iceberg_catalog".to_string()),
schema: Some("sales".to_string()),
table_name: "orders".to_string(),
where_input: Some("WHERE active = true;".to_string()),
}),
"SELECT COUNT(*) AS cnt FROM `iceberg_catalog`.`sales`.`orders` WHERE (active = true)"
);
assert_eq!(
build_data_grid_count_sql(DataGridCountSqlOptions {
database_type: Some(DatabaseType::StarRocks),
catalog: Some("hive_catalog".to_string()),
schema: None,
table_name: "orders".to_string(),
where_input: None,
}),
"SELECT COUNT(*) AS cnt FROM `hive_catalog`.`orders`"
);
}
#[test]
@@ -2523,6 +2627,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::SqlServer),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("dbo".to_string()),
table_name: "users".to_string(),
primary_keys: vec!["Id".to_string()],
@@ -2545,6 +2650,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Oracle),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("APP".to_string()),
table_name: "EVENTS".to_string(),
primary_keys: vec!["ID".to_string()],
@@ -2590,6 +2696,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::SqlServer),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("dbo".to_string()),
table_name: "flags".to_string(),
primary_keys: vec![],
@@ -2620,6 +2727,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Mysql),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "employees".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2646,6 +2754,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Mysql),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "employees".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2672,6 +2781,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Mysql),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "employees".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2698,6 +2808,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::SqlServer),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("game".to_string()),
table_name: "player states".to_string(),
primary_keys: vec!["role id".to_string()],
@@ -2726,6 +2837,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Databend),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("default".to_string()),
table_name: "people".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2756,6 +2868,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::ClickHouse),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("default".to_string()),
table_name: "people".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2788,6 +2901,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::ClickHouse),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("default".to_string()),
table_name: "people".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2818,6 +2932,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::ClickHouse),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("default".to_string()),
table_name: "events".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2844,6 +2959,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::ClickHouse),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("default".to_string()),
table_name: "events".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2876,6 +2992,7 @@ mod tests {
let statements = build_data_grid_copy_update_statements(DataGridCopyUpdateStatementOptions {
database_type: Some(DatabaseType::ClickHouse),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("default".to_string()),
table_name: "people".to_string(),
primary_keys: vec!["id".to_string()],
@@ -2889,11 +3006,69 @@ mod tests {
assert_eq!(statements, vec!["ALTER TABLE `people` UPDATE `name` = 'Ada' WHERE `id` = 1;"]);
}
#[test]
fn doris_external_catalog_save_and_copy_statements_use_catalog_scope() {
let table_meta = DataGridTableMeta {
catalog: Some("iceberg_catalog".to_string()),
schema: Some("sales".to_string()),
table_name: "orders".to_string(),
primary_keys: vec!["id".to_string()],
columns: Some(vec![column("id", "bigint", false, None), column("status", "varchar", true, None)]),
};
let copy_updates = build_data_grid_copy_update_statements(DataGridCopyUpdateStatementOptions {
database_type: Some(DatabaseType::Doris),
table_meta: table_meta.clone(),
columns: vec!["id".to_string(), "status".to_string()],
source_columns: None,
rows: vec![vec![json!(1), json!("paid")]],
});
assert_eq!(
copy_updates,
vec!["UPDATE `iceberg_catalog`.`sales`.`orders` SET `status` = 'paid' WHERE `id` = 1;"]
);
let copy_insert = build_data_grid_copy_insert_statement(DataGridCopyInsertStatementOptions {
database_type: Some(DatabaseType::Doris),
table_meta: Some(table_meta.clone()),
columns: vec!["id".to_string(), "status".to_string()],
source_columns: None,
rows: vec![vec![json!(2), json!("new")]],
exclude_primary_keys: false,
insert_mode: DataGridCopyInsertMode::Merged,
});
assert_eq!(
copy_insert.as_deref(),
Some("INSERT INTO `iceberg_catalog`.`sales`.`orders` (`id`, `status`) VALUES (2, 'new');")
);
let save = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Doris),
table_meta,
columns: vec!["id".to_string(), "status".to_string()],
source_columns: None,
rows: vec![vec![json!(1), json!("pending")], vec![json!(3), json!("cancelled")]],
dirty_rows: vec![(0, vec![(1, json!("paid"))])],
deleted_rows: vec![1],
new_rows: vec![vec![json!(4), json!("new")]],
});
assert_eq!(
save.statements,
vec![
"UPDATE `iceberg_catalog`.`sales`.`orders` SET `status` = 'paid' WHERE `id` = 1;",
"DELETE FROM `iceberg_catalog`.`sales`.`orders` WHERE `id` = 3;",
"INSERT INTO `iceberg_catalog`.`sales`.`orders` (`id`, `status`) VALUES (4, 'new');",
]
);
assert!(save.validation_error.is_none());
}
#[test]
fn prepares_databend_keyless_save_statements_with_row_predicate() {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Databend),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("default".to_string()),
table_name: "people".to_string(),
primary_keys: vec![],
@@ -2921,6 +3096,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Oscar),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("SYSDBA".to_string()),
table_name: "PEOPLE".to_string(),
primary_keys: vec![],
@@ -2948,6 +3124,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Postgres),
table_meta: DataGridTableMeta {
catalog: None,
schema: Some("public".to_string()),
table_name: "ihli_data".to_string(),
primary_keys: vec!["iso3".to_string(), "year".to_string()],
@@ -2980,6 +3157,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Mysql),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "policies".to_string(),
primary_keys: vec!["id".to_string()],
@@ -3030,6 +3208,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Mysql),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "school".to_string(),
primary_keys: vec!["id".to_string()],
@@ -3054,6 +3233,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Mysql),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "parts".to_string(),
primary_keys: vec![],
@@ -3084,6 +3264,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::ManticoreSearch),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "rt_products".to_string(),
primary_keys: vec![],
@@ -3113,6 +3294,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Postgres),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "education_data".to_string(),
primary_keys: vec!["country_code".to_string(), "year".to_string()],
@@ -3149,6 +3331,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Sqlite),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "OnlineLogs".to_string(),
primary_keys: vec!["OnlineLogId".to_string()],
@@ -3174,6 +3357,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Sqlite),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "OnlineLogs".to_string(),
primary_keys: vec!["OnlineLogId".to_string()],
@@ -3202,6 +3386,7 @@ mod tests {
let result = prepare_data_grid_save(DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Sqlite),
table_meta: DataGridTableMeta {
catalog: None,
schema: None,
table_name: "OnlineLogs".to_string(),
primary_keys: vec!["OnlineLogId".to_string()],
crates/dbx-core/src/db/mysql.rs
+294
View File
@@ -90,6 +90,19 @@ fn get_opt_str(row: &mysql_async::Row, name: &str) -> Option<String> {
.or_else(|| row_get::<Vec<u8>, _>(row, name).map(|b| String::from_utf8_lossy(&b).to_string()))
}
/// First non-empty string value among the named columns (e.g. Doris `CatalogName`
/// vs StarRocks `Catalog`). Returns an empty string when none of the columns
/// are present or all are empty.
fn first_nonempty_str_by_name(row: &mysql_async::Row, names: &[&str]) -> String {
for name in names {
let value = get_str_by_name(row, name);
if !value.is_empty() {
return value;
}
}
String::new()
}
fn get_opt_metadata_string(row: &mysql_async::Row, name: &str) -> Option<String> {
get_opt_str(row, name)
.or_else(|| row_get::<NaiveDateTime, _>(row, name).map(|value| value.to_string()))
@@ -3081,6 +3094,202 @@ 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");
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"),
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,
})
})
.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 merge_index_infos(target: &mut Vec<IndexInfo>, parsed: Vec<IndexInfo>) {
let mut seen_names: HashSet<String> = target.iter().map(|index| index.name.to_ascii_lowercase()).collect();
for index in parsed {
@@ -4401,4 +4610,89 @@ UNIQUE KEY(`tenant_id`, `name``part`)
vec!["USE `db`", "SET NAMES utf8mb4", "SET @@group_concat_max_len = 1048576"]
);
}
fn doris_catalog_info(name: &str, catalog_type: &str, is_current: bool) -> crate::db::CatalogInfo {
crate::db::CatalogInfo {
name: name.to_string(),
catalog_type: catalog_type.to_string(),
is_current,
comment: None,
}
}
#[test]
fn doris_catalog_database_ref_backtick_qualifies_two_parts() {
assert_eq!(doris_catalog_database_ref("iceberg_catalog", "sales"), "`iceberg_catalog`.`sales`");
}
#[test]
fn doris_catalog_table_ref_backtick_qualifies_three_parts() {
assert_eq!(doris_catalog_table_ref("iceberg_catalog", "sales", "orders"), "`iceberg_catalog`.`sales`.`orders`");
}
#[test]
fn doris_catalog_refs_escape_embedded_backticks() {
assert_eq!(doris_catalog_database_ref("a`b", "c`d"), "`a``b`.`c``d`");
assert_eq!(doris_catalog_table_ref("a`b", "c`d", "e`f"), "`a``b`.`c``d`.`e``f`");
}
#[test]
fn normalize_doris_catalogs_does_not_inject_missing_internal() {
// SHOW CATALOGS always lists the built-in catalog on both engines, so a
// missing internal catalog is not synthesized — the caller's flat-sidebar
// fallback handles a single/empty result instead.
let catalogs = vec![
doris_catalog_info("iceberg_catalog", "iceberg", true),
doris_catalog_info("hive_catalog", "hive", false),
];
let normalized = normalize_doris_catalogs(catalogs);
let names: Vec<&str> = normalized.iter().map(|c| c.name.as_str()).collect();
assert_eq!(names, vec!["hive_catalog", "iceberg_catalog"]);
assert!(!normalized.iter().any(|c| c.is_internal()));
}
#[test]
fn normalize_doris_catalogs_keeps_existing_internal_first() {
let catalogs = vec![
doris_catalog_info("iceberg_catalog", "iceberg", false),
doris_catalog_info("internal", "internal", true),
doris_catalog_info("hive_catalog", "hive", false),
];
let normalized = normalize_doris_catalogs(catalogs);
let names: Vec<&str> = normalized.iter().map(|c| c.name.as_str()).collect();
assert_eq!(names, vec!["internal", "hive_catalog", "iceberg_catalog"]);
}
#[test]
fn normalize_doris_catalogs_sorts_starrocks_default_catalog_first() {
// StarRocks names its built-in catalog `default_catalog` (Type=Internal);
// detection is type-based, so it sorts first just like Doris `internal`.
let catalogs = vec![
doris_catalog_info("hive_catalog", "hive", false),
doris_catalog_info("default_catalog", "Internal", true),
];
let normalized = normalize_doris_catalogs(catalogs);
let names: Vec<&str> = normalized.iter().map(|c| c.name.as_str()).collect();
assert_eq!(names, vec!["default_catalog", "hive_catalog"]);
assert!(normalized[0].is_internal());
assert!(!normalized[1].is_internal());
}
#[test]
fn normalize_doris_catalogs_detects_internal_by_type_not_name() {
// A catalog literally named `internal` but with an external type is NOT
// the built-in catalog and must not sort first.
let catalogs =
vec![doris_catalog_info("internal", "iceberg", false), doris_catalog_info("hive_catalog", "hive", false)];
let normalized = normalize_doris_catalogs(catalogs);
let names: Vec<&str> = normalized.iter().map(|c| c.name.as_str()).collect();
assert_eq!(names, vec!["hive_catalog", "internal"]);
assert!(!normalized.iter().any(|c| c.is_internal()));
}
#[test]
fn normalize_doris_catalogs_handles_empty_input() {
let normalized = normalize_doris_catalogs(Vec::new());
assert!(normalized.is_empty());
}
}
crates/dbx-core/src/schema.rs
+229
View File
@@ -522,6 +522,196 @@ pub async fn list_sqlserver_linked_server_tables_core(
Ok(vec![])
}
// ---------------------------------------------------------------------------
// Doris / StarRocks multi-catalog federation.
//
// These engines expose external catalogs (iceberg, hive, jdbc, ...) alongside
// the native `internal` catalog via `SHOW CATALOGS`. The functions below browse
// a specific catalog's databases/tables and read table metadata using 3-part
// qualified names (`<catalog>.<database>.<table>`), which the engines accept
// directly. The native `internal` catalog continues to use the existing
// `list_databases_core` / `list_tables_core` paths.
// ---------------------------------------------------------------------------
/// `SHOW CATALOGS` → catalogs visible to the current user. Returns an empty
/// list when the connection pool is not a MySQL pool (Doris/StarRocks always
/// use the MySQL protocol, so this is a defensive no-op); the caller's
/// flat-sidebar fallback then renders the standard database list.
pub async fn list_doris_catalogs_core(state: &AppState, connection_id: &str) -> Result<Vec<db::CatalogInfo>, String> {
let pool_key = state.get_or_create_pool(connection_id, None).await?;
let connections = state.connections.read().await;
if let Some(PoolKind::Mysql(p, _)) = connections.get(&pool_key) {
return db::mysql::list_doris_catalogs(p).await;
}
Ok(vec![])
}
/// `SHOW DATABASES FROM <catalog>` → databases in the given catalog.
///
/// For `internal`, system databases are filtered (mirroring `list_databases_core`).
/// For external catalogs, permission errors degrade to an empty list (the user
/// asked that inaccessible catalogs simply not be shown).
pub async fn list_doris_catalog_databases_core(
state: &AppState,
connection_id: &str,
catalog: &str,
) -> Result<Vec<db::DatabaseInfo>, String> {
let pool_key = state.get_or_create_pool(connection_id, None).await?;
let db_config = connection_config(state, connection_id).await;
let connections = state.connections.read().await;
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
let PoolKind::Mysql(p, _) = pool else {
return Ok(vec![]);
};
let databases = db::mysql::list_databases_show_from(p, catalog).await;
// External catalogs may reject `SHOW DATABASES FROM <catalog>` when the user
// lacks permission — surface as an empty list rather than an error.
let databases = match databases {
Ok(databases) => databases,
Err(error) => {
log::warn!(
"[schema][doris:list_catalog_databases] connection_id={} catalog={} error={}",
connection_id,
catalog,
error
);
return Ok(vec![]);
}
};
if catalog == "internal" {
return Ok(filter_mysql_system_databases_for_config(databases, db_config.as_ref()));
}
Ok(databases)
}
/// `SHOW TABLES FROM <catalog>.<database>` → tables in an external catalog.
pub async fn list_doris_catalog_tables_core(
state: &AppState,
connection_id: &str,
catalog: &str,
database: &str,
filter: Option<&str>,
limit: Option<usize>,
offset: Option<usize>,
object_types: Option<&[String]>,
) -> Result<Vec<db::TableInfo>, String> {
let pool_key = state.get_or_create_pool(connection_id, None).await?;
let connections = state.connections.read().await;
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
let PoolKind::Mysql(p, _) = pool else {
return Ok(vec![]);
};
db::mysql::list_tables_show_from(p, catalog, database)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types))
}
/// Columns of an external catalog table via `SHOW COLUMNS FROM <catalog>.<db>.<table>`.
pub async fn get_doris_catalog_columns_core(
state: &AppState,
connection_id: &str,
catalog: &str,
database: &str,
table: &str,
) -> Result<Vec<db::ColumnInfo>, String> {
let pool_key = state.get_or_create_pool(connection_id, None).await?;
let connections = state.connections.read().await;
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
let PoolKind::Mysql(p, _) = pool else {
return Ok(vec![]);
};
db::mysql::get_columns_show_from(p, catalog, database, table).await.map(deduplicate_column_infos)
}
/// DDL for an external catalog table via `SHOW CREATE TABLE <catalog>.<db>.<table>`.
pub async fn get_doris_catalog_table_ddl_core(
state: &AppState,
connection_id: &str,
catalog: &str,
database: &str,
table: &str,
) -> Result<String, String> {
let pool_key = state.get_or_create_pool(connection_id, None).await?;
let connections = state.connections.read().await;
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
let PoolKind::Mysql(p, _) = pool else {
return Err("DDL not supported for this connection".to_string());
};
db::mysql::show_create_table_ddl_from(p, catalog, database, table).await
}
/// Best-effort index listing for an external catalog table (derived from DDL).
pub async fn list_doris_catalog_indexes_core(
state: &AppState,
connection_id: &str,
catalog: &str,
database: &str,
table: &str,
) -> Result<Vec<db::IndexInfo>, String> {
let pool_key = state.get_or_create_pool(connection_id, None).await?;
let connections = state.connections.read().await;
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
let PoolKind::Mysql(p, _) = pool else {
return Ok(vec![]);
};
db::mysql::list_doris_catalog_indexes(p, catalog, database, table).await
}
/// Table comment for an external catalog table. Doris does not reliably expose
/// comments for external catalog tables, so this returns `None`.
pub async fn get_doris_catalog_table_comment_core(
_state: &AppState,
_connection_id: &str,
_catalog: &str,
_database: &str,
_table: &str,
) -> Result<Option<String>, String> {
Ok(None)
}
/// Foreign keys are not applicable to external catalog tables.
pub async fn list_doris_catalog_foreign_keys_core(
_state: &AppState,
_connection_id: &str,
_catalog: &str,
_database: &str,
_table: &str,
) -> Result<Vec<db::ForeignKeyInfo>, String> {
Ok(vec![])
}
/// Triggers are not applicable to external catalog tables.
pub async fn list_doris_catalog_triggers_core(
_state: &AppState,
_connection_id: &str,
_catalog: &str,
_database: &str,
_table: &str,
) -> Result<Vec<db::TriggerInfo>, String> {
Ok(vec![])
}
/// Resolve a non-internal catalog for dispatch to the Doris multi-catalog path.
/// Returns `Some(catalog)` only when `catalog` is a non-empty, non-`internal`
/// name and the connection is a Doris-family engine that supports
/// `SHOW CATALOGS`. Otherwise `None` (caller uses the default metadata path).
pub async fn resolve_external_doris_catalog(
state: &AppState,
connection_id: &str,
catalog: Option<&str>,
) -> Option<String> {
let catalog = catalog?.trim();
if catalog.is_empty() || catalog == "internal" {
return None;
}
let config = connection_config(state, connection_id).await?;
if is_doris_family_catalog_capable_config(&config) {
Some(catalog.to_string())
} else {
None
}
}
async fn list_databases_once(state: &AppState, connection_id: &str) -> Result<Vec<db::DatabaseInfo>, String> {
log::info!("[list_databases] connection_id={connection_id}");
let db_config = connection_config(state, connection_id).await;
@@ -2969,6 +3159,37 @@ mod tests {
vec!["ORDERS".to_string(), "ORDERS_VIEW".to_string()]
);
}
#[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)));
// 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));
let mut selectdb = test_connection_config(DatabaseType::Mysql);
selectdb.driver_profile = Some("selectdb".to_string());
assert!(super::is_doris_family_catalog_capable_config(&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));
// ManticoreSearch shares the MySQL code path but has no catalog concept.
assert!(!super::is_doris_family_catalog_capable_config(&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));
// 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)));
}
}
pub async fn list_objects_core(
@@ -4384,6 +4605,14 @@ fn is_doris_family_config(config: &ConnectionConfig) -> bool {
|| matches!(config.driver_profile.as_deref(), Some("doris" | "selectdb" | "starrocks" | "manticoresearch"))
}
/// 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"))
crates/dbx-core/src/sql_dialect.rs
+3 -1
View File
@@ -10,7 +10,9 @@ pub use capabilities::{
firebird_rows_clause, is_schema_aware, pagination_strategy, table_pagination_strategy, uses_fetch_first,
uses_single_row_insert_statements, PaginationContext, TablePaginationStrategy,
};
pub use identifiers::{normalize_where_input, qualified_table_name, quote_table_identifier};
pub use identifiers::{
normalize_where_input, qualified_table_name, qualified_table_name_with_catalog, quote_table_identifier,
};
pub(crate) use identifiers::{parse_sqlserver_linked_schema_ref, qualified_transfer_table, quote_transfer_identifier};
pub use table_select::{build_count_table_sql, build_table_data_select_sql, build_table_select_sql};
pub use types::*;
crates/dbx-core/src/sql_dialect/identifiers.rs
+23
View File
@@ -71,6 +71,29 @@ pub fn qualified_table_name(database_type: Option<DatabaseType>, schema: Option<
quote_table_identifier(database_type, table_name)
}
/// Like `qualified_table_name`, but prefixes a Doris/StarRocks external
/// catalog (`<catalog>.<schema>.<table>`) when `catalog` is present, non-empty,
/// and not the engine's `internal` catalog. The `internal` guard is defensive —
/// built-in-catalog tables never carry a catalog in the first place (the
/// sidebar routes them through the standard path), so this only ever prefixes
/// genuine external catalogs. Other engines ignore `catalog` (they have no
/// 3-part catalog naming).
pub fn qualified_table_name_with_catalog(
database_type: Option<DatabaseType>,
catalog: Option<&str>,
schema: Option<&str>,
table_name: &str,
) -> String {
let table = qualified_table_name(database_type, schema, table_name);
let catalog = catalog.map(str::trim).filter(|catalog| !catalog.is_empty() && *catalog != "internal");
match (catalog, database_type) {
(Some(catalog), Some(DatabaseType::Doris | DatabaseType::StarRocks)) => {
format!("{}.{}", quote_table_identifier(database_type, catalog), table)
}
_ => table,
}
}
pub fn quote_table_identifier(database_type: Option<DatabaseType>, name: &str) -> String {
match database_type {
Some(DatabaseType::Iotdb) => name.to_string(),
crates/dbx-core/src/sql_dialect/table_select.rs
+73 -2
View File
@@ -1,7 +1,9 @@
use crate::models::connection::DatabaseType;
use super::capabilities::{firebird_rows_clause, table_pagination_strategy, uses_fetch_first, TablePaginationStrategy};
use super::identifiers::{normalize_where_input, qualified_table_name, quote_table_identifier};
use super::identifiers::{
normalize_where_input, qualified_table_name, qualified_table_name_with_catalog, quote_table_identifier,
};
use super::types::{
TableDataSelectSqlOptions, TableSelectSqlOptions, DBX_NEO4J_ELEMENT_ID_COLUMN, DBX_ROWID_COLUMN,
DBX_TDENGINE_TBNAME_COLUMN,
@@ -18,7 +20,13 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
return build_neo4j_table_select_sql(&options, limit);
}
let table = qualified_table_name(database_type, options.schema.as_deref(), &options.table_name);
// Doris / StarRocks multi-catalog: prefix the catalog for external-catalog tables.
let table = qualified_table_name_with_catalog(
database_type,
options.catalog.as_deref(),
options.schema.as_deref(),
&options.table_name,
);
let predicate = normalize_where_input(options.where_input.as_deref());
let where_clause = if predicate.is_empty() { String::new() } else { format!(" WHERE ({predicate})") };
let default_order_by = if database_type == Some(DatabaseType::InfluxDb) {
@@ -399,3 +407,66 @@ pub(super) fn build_questdb_table_select_sql(
let upper_bound = offset + limit;
format!("SELECT {columns_sql} FROM {table}{where_clause}{order_by} LIMIT {offset}, {upper_bound}")
}
#[cfg(test)]
mod tests {
use super::*;
fn opts(database_type: DatabaseType, catalog: Option<&str>, table: &str) -> TableDataSelectSqlOptions {
TableDataSelectSqlOptions {
database_type: Some(database_type),
schema: None,
table_name: table.to_string(),
catalog: catalog.map(|c| c.to_string()),
table_type: None,
primary_keys: Vec::new(),
columns: Vec::new(),
fallback_order_columns: Vec::new(),
order_by: None,
limit: Some(10),
offset: None,
where_input: None,
include_row_id: false,
}
}
#[test]
fn doris_external_catalog_prefixes_from_clause() {
let sql = build_table_data_select_sql(opts(DatabaseType::Doris, Some("iceberg_catalog"), "orders"));
assert!(sql.contains("FROM `iceberg_catalog`.`orders`"), "sql was: {sql}");
}
#[test]
fn starrocks_external_catalog_prefixes_from_clause() {
let sql = build_table_data_select_sql(opts(DatabaseType::StarRocks, Some("hive_catalog"), "orders"));
assert!(sql.contains("FROM `hive_catalog`.`orders`"), "sql was: {sql}");
}
#[test]
fn doris_internal_catalog_is_not_prefixed() {
let sql = build_table_data_select_sql(opts(DatabaseType::Doris, Some("internal"), "orders"));
assert!(!sql.contains("internal"), "sql was: {sql}");
assert!(sql.contains("FROM `orders`"), "sql was: {sql}");
}
#[test]
fn doris_empty_catalog_is_not_prefixed() {
let sql = build_table_data_select_sql(opts(DatabaseType::Doris, Some(" "), "orders"));
assert!(sql.contains("FROM `orders`"), "sql was: {sql}");
}
#[test]
fn doris_no_catalog_is_not_prefixed() {
let sql = build_table_data_select_sql(opts(DatabaseType::Doris, None, "orders"));
assert!(sql.contains("FROM `orders`"), "sql was: {sql}");
}
#[test]
fn external_catalog_is_ignored_for_non_doris_engines() {
// Postgres does not support the 3-part catalog naming; the catalog
// must be ignored to avoid emitting an invalid qualified name.
let sql = build_table_data_select_sql(opts(DatabaseType::Postgres, Some("iceberg_catalog"), "orders"));
assert!(!sql.contains("iceberg_catalog"), "sql was: {sql}");
assert!(sql.contains("orders"), "sql was: {sql}");
}
}
crates/dbx-core/src/sql_dialect/tests.rs
+26
View File
@@ -258,6 +258,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: Some("where status = 'active'".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT * FROM `users` WHERE (status = 'active') LIMIT 100;"
);
@@ -275,6 +276,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT * FROM `sys_dic` LIMIT 100;"
);
@@ -292,6 +294,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: Some(100),
where_input: Some("WHERE amount > 10".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT * FROM \"public\".\"orders\" WHERE (amount > 10) LIMIT 50 OFFSET 100;"
);
@@ -309,6 +312,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT * FROM \"DBX_TEST\".\"PRODUCTS\" LIMIT 100;"
);
@@ -326,6 +330,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: Some("`customer_name` = 'Acme'".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT * FROM `sales_report` WHERE (`customer_name` = 'Acme') LIMIT 100;"
);
@@ -343,6 +348,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: Some("WHERE amount > 10".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT * FROM \"DB2INST1\".\"ORDERS\" WHERE (amount > 10) FETCH FIRST 50 ROWS ONLY"
);
@@ -360,6 +366,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: Some(100),
where_input: Some("WHERE amount > 10".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT * FROM \"ORDERS\" WHERE (amount > 10) ORDER BY \"ID\" ASC ROWS 101 TO 150"
);
@@ -377,6 +384,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: Some(100),
where_input: Some("WHERE amount > 10".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT \"ID\", \"AMOUNT\" FROM (SELECT dbx_inner.*, ROWNUM AS \"__dbx_row_num\" FROM (SELECT \"ID\", \"AMOUNT\" FROM \"DBXTEST\".\"ORDERS\" WHERE (amount > 10) ORDER BY \"ID\" ASC) dbx_inner WHERE ROWNUM <= 150) WHERE \"__dbx_row_num\" > 100"
);
@@ -394,6 +402,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT * FROM (SELECT * FROM \"DBXTEST\".\"ORDERS\") WHERE ROWNUM <= 100"
);
@@ -411,6 +420,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: Some("WHERE amount > 10".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT * FROM (SELECT * FROM \"DBXTEST\".\"ORDERS\" WHERE (amount > 10)) WHERE ROWNUM <= 50"
);
@@ -428,6 +438,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: Some(100),
where_input: Some("WHERE amount > 10".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT \"ID\", \"AMOUNT\" FROM (SELECT dbx_t.\"ID\", dbx_t.\"AMOUNT\", ROW_NUMBER() OVER () AS \"__dbx_row_num\" FROM \"DB2INST1\".\"ORDERS\" dbx_t WHERE (amount > 10)) dbx_page WHERE \"__dbx_row_num\" > 100 AND \"__dbx_row_num\" <= 150 ORDER BY \"__dbx_row_num\""
);
@@ -445,6 +456,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT TOP 100 * FROM \"Ens\".\"AlarmResponse\""
);
@@ -462,6 +474,7 @@ fn builds_table_data_where_and_schema_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT * FROM root.test.device2 LIMIT 100;"
);
@@ -483,6 +496,7 @@ fn builds_informix_table_data_with_skip_first_pagination() {
offset: Some(100),
where_input: Some("WHERE active = 1".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT SKIP 100 FIRST 50 * FROM users WHERE (active = 1)"
);
@@ -516,6 +530,7 @@ fn explicit_table_data_order_is_preserved() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT * FROM \"public\".\"country_gdp\" ORDER BY \"iso3\" ASC LIMIT 100;"
);
@@ -536,6 +551,7 @@ fn builds_iris_table_data_sql_with_literal_top_and_quoted_object() {
offset: None,
where_input: Some("WHERE \"Status\" = 'Open'".to_string()),
include_row_id: false,
..Default::default()
});
assert_eq!(
@@ -569,6 +585,7 @@ fn builds_table_data_special_column_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT tbname, `ts` AS `ts`, `current` AS `current`, `voltage` AS `voltage`, `location` AS `location`, `groupid` AS `groupid` FROM `test_db`.`meters` LIMIT 100;"
);
@@ -586,6 +603,7 @@ fn builds_table_data_special_column_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT `ts` AS `ts`, `current` AS `current` FROM `test_db`.`d1001` LIMIT 100;"
);
@@ -603,6 +621,7 @@ fn builds_table_data_special_column_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT `ts` AS `ts`, `current` AS `current` FROM `test_db`.`d1001` LIMIT 100;"
);
@@ -620,6 +639,7 @@ fn builds_table_data_special_column_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT `id` AS `id`, `name` AS `name` FROM `departments` LIMIT 100;"
);
@@ -641,6 +661,7 @@ fn builds_sqlserver_table_data_pages() {
offset: None,
where_input: Some("where id = 1".to_string()),
include_row_id: false,
..Default::default()
}),
"SELECT TOP (25) * FROM [dbo].[accounts] WHERE (id = 1)"
);
@@ -658,6 +679,7 @@ fn builds_sqlserver_table_data_pages() {
offset: Some(100),
where_input: None,
include_row_id: false,
..Default::default()
}),
"WITH [dbx_page] AS (SELECT [order_id], [customer], ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) AS [__dbx_row_num] FROM [sales].[orders]) SELECT [order_id], [customer] FROM [dbx_page] WHERE [__dbx_row_num] > 100 AND [__dbx_row_num] <= 150 ORDER BY [__dbx_row_num]"
);
@@ -679,6 +701,7 @@ fn builds_oracle_and_neo4j_table_data_queries() {
offset: None,
where_input: None,
include_row_id: true,
..Default::default()
}),
"SELECT * FROM (SELECT ROWIDTOCHAR(t.ROWID) AS \"__DBX_ROWID\", t.* FROM \"DBXTEST\".\"DBX_LOAD_TABLE_006\" t) WHERE ROWNUM <= 100"
);
@@ -696,6 +719,7 @@ fn builds_oracle_and_neo4j_table_data_queries() {
offset: None,
where_input: None,
include_row_id: true,
..Default::default()
}),
"SELECT \"__DBX_ROWID\", \"ID\", \"NAME\" FROM (SELECT ROWIDTOCHAR(t.ROWID) AS \"__DBX_ROWID\", t.* FROM \"DBXTEST\".\"DBX_LOAD_TABLE_006\" t) WHERE ROWNUM <= 100"
);
@@ -713,6 +737,7 @@ fn builds_oracle_and_neo4j_table_data_queries() {
offset: None,
where_input: None,
include_row_id: true,
..Default::default()
}),
"SELECT \"ID\", \"NAME\" FROM (SELECT \"ID\", \"NAME\" FROM \"DBXTEST\".\"DBX_JOIN_VIEW\") WHERE ROWNUM <= 100"
);
@@ -730,6 +755,7 @@ fn builds_oracle_and_neo4j_table_data_queries() {
offset: None,
where_input: None,
include_row_id: false,
..Default::default()
}),
"MATCH (n:`Employee`) RETURN elementId(n) AS `__DBX_ELEMENT_ID`, n.`id` AS `id`, n.`first name` AS `first name`, n.`role` AS `role` LIMIT 100;"
);
crates/dbx-core/src/sql_dialect/types.rs
+6 -1
View File
@@ -16,7 +16,7 @@ pub struct TableSelectSqlOptions<'a> {
pub limit: usize,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TableDataSelectSqlOptions {
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -24,6 +24,11 @@ pub struct TableDataSelectSqlOptions {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
pub table_name: String,
/// Doris / StarRocks multi-catalog: when set to a non-`internal` catalog,
/// the FROM clause is prefixed with the catalog
/// (`<catalog>.<schema>.<table>`).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub catalog: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub table_type: Option<String>,
#[serde(default)]
crates/dbx-core/src/types.rs
+28
View File
@@ -19,6 +19,34 @@ pub struct LinkedServerInfo {
pub data_source: Option<String>,
}
/// A catalog exposed by a multi-catalog engine (e.g. Doris / StarRocks).
/// `internal` is the engine's native catalog; other entries are external
/// catalogs (iceberg, hive, jdbc, ...) federated through the same connection.
///
/// Note: the built-in catalog is named `internal` in Doris (Type=`internal`)
/// but `default_catalog` in StarRocks (Type=`Internal`). The `catalog_type`
/// column is the cross-engine signal, so `is_internal()` matches it
/// case-insensitively and falls back to the canonical Doris name when the
/// column is absent (very old / proxied deployments).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CatalogInfo {
pub name: String,
pub catalog_type: String,
pub is_current: bool,
pub comment: Option<String>,
}
impl CatalogInfo {
/// Whether this is the engine's built-in (non-federated) catalog.
pub fn is_internal(&self) -> bool {
if !self.catalog_type.trim().is_empty() {
self.catalog_type.eq_ignore_ascii_case("internal")
} else {
self.name == "internal"
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableInfo {
pub name: String,