feat(database): support Doris and StarRocks catalogs

crates/dbx-core/assets/database-drivers.manifest.json

@@ -506,7 +506,7 @@
       "runtimeMode": "native",
       "mcpMode": "direct",
       "singleConnectionPool": false,
-      "metadataConnectionScoped": false,
+      "metadataConnectionScoped": true,
       "skipTcpProbe": false,
       "defaultPort": 9030,
       "supportLevel": "operate",
@@ -535,7 +535,7 @@
       "runtimeMode": "native",
       "mcpMode": "direct",
       "singleConnectionPool": false,
-      "metadataConnectionScoped": false,
+      "metadataConnectionScoped": true,
       "skipTcpProbe": false,
       "defaultPort": 9030,
       "supportLevel": "operate",
@@ -564,7 +564,7 @@
       "runtimeMode": "native",
       "mcpMode": "direct",
       "singleConnectionPool": false,
-      "metadataConnectionScoped": false,
+      "metadataConnectionScoped": true,
       "skipTcpProbe": false,
       "defaultPort": 9306,
       "supportLevel": "operate",

crates/dbx-core/src/connection.rs

@@ -304,6 +304,17 @@ pub fn metadata_connection_config(config: &ConnectionConfig) -> ConnectionConfig
 }
 
 pub fn database_connection_config(config: &ConnectionConfig, database: Option<&str>) -> ConnectionConfig {
+    database_connection_config_with_catalog(config, database, None)
+}
+
+/// Like [`database_connection_config`], but optionally injects a Doris/StarRocks
+/// `catalog=<name>` URL parameter so mysql_async emits `SET catalog` during
+/// connection setup (before any `USE <database>`).
+pub fn database_connection_config_with_catalog(
+    config: &ConnectionConfig,
+    database: Option<&str>,
+    catalog: Option<&str>,
+) -> ConnectionConfig {
     let mut db_config = if database.is_some() { config.clone() } else { metadata_connection_config(config) };
     if let Some(db) = database {
         if !matches!(
@@ -317,9 +328,35 @@ pub fn database_connection_config(config: &ConnectionConfig, database: Option<&s
             db_config.database = Some(db.to_string());
         }
     }
+    if let Some(catalog) = catalog.map(str::trim).filter(|catalog| !catalog.is_empty()) {
+        db_config.url_params = Some(upsert_connection_url_param(db_config.url_params.as_deref(), "catalog", catalog));
+    }
     db_config
 }
 
+/// Insert or replace a single `key=value` entry in a connection URL-params string.
+pub fn upsert_connection_url_param(params: Option<&str>, key: &str, value: &str) -> String {
+    let key = key.trim();
+    let value = value.trim();
+    let key_lower = key.to_ascii_lowercase();
+    let encoded_value = percent_encoding::utf8_percent_encode(value, percent_encoding::NON_ALPHANUMERIC).to_string();
+    let mut parts: Vec<String> = params
+        .unwrap_or("")
+        .trim()
+        .trim_start_matches('?')
+        .split('&')
+        .filter(|part| !part.trim().is_empty())
+        .filter(|part| {
+            part.split_once('=')
+                .map(|(existing_key, _)| existing_key.trim().to_ascii_lowercase() != key_lower)
+                .unwrap_or(true)
+        })
+        .map(str::to_string)
+        .collect();
+    parts.push(format!("{key}={encoded_value}"));
+    parts.join("&")
+}
+
 pub fn prestosql_jdbc_config_for_endpoint(config: &ConnectionConfig, host: &str, port: u16) -> ConnectionConfig {
     let mut jdbc_config = config.clone();
     jdbc_config.connection_string =
@@ -1138,13 +1175,22 @@ impl AppState {
         self.get_or_create_pool_for_session(connection_id, database, None).await
     }
 
+    pub async fn get_or_create_pool_with_catalog(
+        &self,
+        connection_id: &str,
+        database: Option<&str>,
+        catalog: Option<&str>,
+    ) -> Result<String, String> {
+        self.get_or_create_pool_for_session_with_catalog(connection_id, database, catalog, None).await
+    }
+
     pub async fn get_or_create_pool_for_connection_attempt(
         &self,
         connection_id: &str,
         database: Option<&str>,
         attempt: u64,
     ) -> Result<String, String> {
-        self.get_or_create_pool_for_session_inner(connection_id, database, None, Some(attempt)).await
+        self.get_or_create_pool_for_session_inner(connection_id, database, None, None, Some(attempt)).await
     }
 
     pub async fn get_or_create_pool_for_session(
@@ -1153,13 +1199,24 @@ impl AppState {
         database: Option<&str>,
         client_session_id: Option<&str>,
     ) -> Result<String, String> {
-        self.get_or_create_pool_for_session_inner(connection_id, database, client_session_id, None).await
+        self.get_or_create_pool_for_session_with_catalog(connection_id, database, None, client_session_id).await
+    }
+
+    pub async fn get_or_create_pool_for_session_with_catalog(
+        &self,
+        connection_id: &str,
+        database: Option<&str>,
+        catalog: Option<&str>,
+        client_session_id: Option<&str>,
+    ) -> Result<String, String> {
+        self.get_or_create_pool_for_session_inner(connection_id, database, catalog, client_session_id, None).await
     }
 
     async fn get_or_create_pool_for_session_inner(
         &self,
         connection_id: &str,
         database: Option<&str>,
+        catalog: Option<&str>,
         client_session_id: Option<&str>,
         connection_attempt: Option<u64>,
     ) -> Result<String, String> {
@@ -1170,8 +1227,9 @@ impl AppState {
         validate_connection_url_params(&config)?;
         let db_type = Some(config.db_type);
         let validate_existing_pool = should_validate_existing_pool_before_reuse(config.db_type);
+        let catalog = catalog.map(str::trim).filter(|value| !value.is_empty());
 
-        let base_pool_key = base_pool_key_for(db_type, connection_id, database, false);
+        let base_pool_key = base_pool_key_for_with_catalog(db_type, connection_id, database, catalog, false);
         let pool_key = session_scoped_pool_key_for(Some(&config), base_pool_key.clone(), client_session_id);
 
         loop {
@@ -1198,7 +1256,7 @@ impl AppState {
             break;
         }
 
-        let db_config = database_connection_config(&config, database);
+        let db_config = database_connection_config_with_catalog(&config, database, catalog);
 
         validate_h2_file_connection(&db_config)?;
         self.ensure_current_connection_attempt(connection_id, connection_attempt).await?;
@@ -2383,13 +2441,24 @@ impl AppState {
         connection_id: &str,
         database: Option<&str>,
         client_session_id: Option<&str>,
+    ) -> Result<String, String> {
+        self.reconnect_pool_for_session_with_catalog(connection_id, database, None, client_session_id).await
+    }
+
+    pub async fn reconnect_pool_for_session_with_catalog(
+        &self,
+        connection_id: &str,
+        database: Option<&str>,
+        catalog: Option<&str>,
+        client_session_id: Option<&str>,
     ) -> Result<String, String> {
         let config = {
             let configs = self.configs.read().await;
             configs.get(connection_id).cloned()
         };
         let db_type = config.as_ref().map(|config| config.db_type);
-        let base_pool_key = base_pool_key_for(db_type, connection_id, database, true);
+        let catalog = catalog.map(str::trim).filter(|value| !value.is_empty());
+        let base_pool_key = base_pool_key_for_with_catalog(db_type, connection_id, database, catalog, true);
         let pool_key = session_scoped_pool_key_for(config.as_ref(), base_pool_key, client_session_id);
         if self.uses_forwarded_transport(connection_id).await {
             self.remove_connection_pools(connection_id).await;
@@ -2403,7 +2472,7 @@ impl AppState {
                 close_pool_kind_with_timeout(pool_key.clone(), pool).await;
             }
         }
-        self.get_or_create_pool_for_session(connection_id, database, client_session_id).await
+        self.get_or_create_pool_for_session_with_catalog(connection_id, database, catalog, client_session_id).await
     }
 
     pub async fn close_client_session_pool(
@@ -3558,6 +3627,16 @@ fn base_pool_key_for(
     connection_id: &str,
     database: Option<&str>,
     include_elasticsearch_single_pool: bool,
+) -> String {
+    base_pool_key_for_with_catalog(db_type, connection_id, database, None, include_elasticsearch_single_pool)
+}
+
+fn base_pool_key_for_with_catalog(
+    db_type: Option<DatabaseType>,
+    connection_id: &str,
+    database: Option<&str>,
+    catalog: Option<&str>,
+    include_elasticsearch_single_pool: bool,
 ) -> String {
     let is_single_connection_pool = db_type.as_ref().is_some_and(|db_type| {
         let is_single = database_capabilities::is_single_connection_pool(db_type)
@@ -3573,13 +3652,17 @@ fn base_pool_key_for(
         is_single && (!database_capabilities::is_agent_type(db_type) || shares_database_pool_with_connection(db_type))
     });
 
-    if is_single_connection_pool {
+    let key = if is_single_connection_pool {
         connection_id.to_string()
     } else {
         match database.filter(|db| !db.trim().is_empty()) {
             Some(db) => format!("{connection_id}:{db}"),
             None => connection_id.to_string(),
         }
+    };
+    match catalog.map(str::trim).filter(|value| !value.is_empty()) {
+        Some(catalog) => format!("{key}:catalog:{catalog}"),
+        None => key,
     }
 }
 
@@ -3784,12 +3867,13 @@ async fn detect_ob_oracle_mode(config: &ConnectionConfig, pool: &db::mysql::MySq
 mod tests {
     use super::{
         agent_connect_timeout, connection_remote_endpoint, connection_url_for_endpoint, database_connection_config,
-        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, uses_bare_mysql_pool, uses_tcp_probe, validate_connection_url_params,
-        validate_h2_database_path, AppState, MysqlMode, PoolKind, PRESTOSQL_JDBC_DRIVER_CLASS,
+        database_connection_config_with_catalog, 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,
+        PRESTOSQL_JDBC_DRIVER_CLASS,
     };
     use crate::agent_connection::{
         agent_connect_params, mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver,
@@ -3864,6 +3948,38 @@ mod tests {
         }
     }
 
+    #[test]
+    fn upsert_connection_url_param_inserts_and_replaces_catalog() {
+        assert_eq!(upsert_connection_url_param(None, "catalog", "paimon"), "catalog=paimon");
+        assert_eq!(
+            upsert_connection_url_param(Some("charset=utf8mb4"), "catalog", "hive_catalog"),
+            "charset=utf8mb4&catalog=hive%5Fcatalog"
+        );
+        assert_eq!(
+            upsert_connection_url_param(Some("catalog=old&charset=utf8mb4"), "catalog", "new_cat"),
+            "charset=utf8mb4&catalog=new%5Fcat"
+        );
+    }
+
+    #[test]
+    fn database_connection_config_with_catalog_keeps_database_for_use_after_set_catalog() {
+        let mut config = mysql_config(None);
+        config.db_type = DatabaseType::StarRocks;
+        let db_config = database_connection_config_with_catalog(&config, Some("ads"), Some("paimon_catalog"));
+        assert_eq!(db_config.database.as_deref(), Some("ads"));
+        assert_eq!(db_config.url_params.as_deref(), Some("catalog=paimon%5Fcatalog"));
+        let url = db_config.connection_url();
+        assert!(url.contains("/ads"), "url should include database for USE setup: {url}");
+        assert!(url.contains("catalog=paimon%5Fcatalog"), "url should include catalog param: {url}");
+    }
+
+    #[test]
+    fn metadata_connection_config_clears_starrocks_default_database() {
+        let mut config = mysql_config(Some("ads"));
+        config.db_type = DatabaseType::StarRocks;
+        assert_eq!(metadata_connection_config(&config).database, None);
+    }
+
     #[test]
     fn task_client_session_ids_are_stable_and_isolated() {
         assert_eq!(task_client_session_id("table-export", "job-1"), "table-export:job-1");

crates/dbx-core/src/data_compare.rs

@@ -526,6 +526,7 @@ pub async fn prepare_data_compare_missing_target(
             &target_database_type,
             &source_database_type,
             None,
+            None,
         )
     ));
     pre_sync_statements.extend(

crates/dbx-core/src/database_capabilities.rs

@@ -37,7 +37,14 @@ pub fn is_single_connection_pool(db_type: &DatabaseType) -> bool {
 }
 
 pub fn is_metadata_connection_scoped(db_type: &DatabaseType) -> bool {
-    matches!(db_type, DatabaseType::Mysql)
+    // MySQL-protocol engines (incl. Doris/StarRocks) must not keep a configured
+    // default database when the caller asks for an unscoped metadata pool.
+    // External-catalog databases do not exist in the default catalog, so leaving
+    // `config.database` in the URL makes handshake/`USE` fail with Unknown database.
+    matches!(
+        db_type,
+        DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks | DatabaseType::ManticoreSearch
+    )
 }
 
 pub fn skips_tcp_probe(db_type: &DatabaseType) -> bool {

crates/dbx-core/src/database_export.rs

@@ -1012,7 +1012,7 @@ fn generate_postgres_sequence_owner_ddl(sequence: &PostgresExportSequence, schem
     Some(format!(
         "ALTER SEQUENCE {} OWNED BY {}.{}",
         postgres_sequence_qualified_name(schema, &sequence.name),
-        crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres),
+        crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres, None),
         quote_identifier(owner_column, &DatabaseType::Postgres)
     ))
 }
@@ -1026,7 +1026,7 @@ fn generate_postgres_sequence_setval_sql(sequence: &PostgresExportSequence, sche
     let sequence_literal = quote_postgres_string_literal(&postgres_sequence_qualified_name(schema, &sequence.name));
     match (sequence.owner_table.as_deref(), sequence.owner_column.as_deref()) {
         (Some(owner_table), Some(owner_column)) => {
-            let owner_table = crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres);
+            let owner_table = crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres, None);
             let owner_column = quote_identifier(owner_column, &DatabaseType::Postgres);
             Some(format!(
                 "SELECT setval({sequence_literal}, GREATEST(COALESCE(MAX({owner_column}), {last_value}), {last_value}), true) FROM {owner_table}"
@@ -1269,7 +1269,7 @@ fn record_export_error(file: &mut std::fs::File, fail_on_error: bool, message: S
 
 fn database_export_select_sql(columns: &[String], table: &str, schema: &str, db_type: &DatabaseType) -> String {
     let columns = columns.iter().map(|column| quote_identifier(column, db_type)).collect::<Vec<_>>().join(", ");
-    let table = crate::transfer::qualified_table(table, schema, db_type);
+    let table = crate::transfer::qualified_table(table, schema, db_type, None);
     format!("SELECT {columns} FROM {table}")
 }
 
@@ -1283,7 +1283,8 @@ fn write_database_export_rows(
     schema: &str,
     db_type: &DatabaseType,
 ) -> Result<(), String> {
-    let mut insert_sql = crate::transfer::generate_insert_typed(columns, column_types, rows, table, schema, db_type);
+    let mut insert_sql =
+        crate::transfer::generate_insert_typed(columns, column_types, rows, table, schema, db_type, None);
     if *db_type == DatabaseType::Dameng && selected_columns_include_identity_extras(columns, column_extras) {
         insert_sql = wrap_dameng_identity_insert_sql(&insert_sql, table, schema);
     }
@@ -1829,7 +1830,7 @@ pub async fn export_database_sql_core(
                     )
                     .await?;
                 } else {
-                    let count_query = crate::transfer::count_sql(table_name, &request.schema, &db_type);
+                    let count_query = crate::transfer::count_sql(table_name, &request.schema, &db_type, None);
                     let total_rows = match crate::transfer::execute_read_on_pool(state, &pool_key, &count_query).await {
                         Ok(result) => {
                             let count = result.rows.first().and_then(|row| row.first()).and_then(|value| match value {
@@ -2137,7 +2138,7 @@ fn filter_export_table_infos(
 }
 
 fn drop_table_if_exists_sql(table_name: &str, schema: &str, db_type: &DatabaseType) -> String {
-    format!("DROP TABLE IF EXISTS {};", crate::transfer::qualified_table(table_name, schema, db_type))
+    format!("DROP TABLE IF EXISTS {};", crate::transfer::qualified_table(table_name, schema, db_type, None))
 }
 
 fn build_database_export_object_source_sql(

crates/dbx-core/src/db/cloudflare_d1/import.rs

@@ -35,7 +35,7 @@ pub(crate) fn build_import_insert_batches(
         .collect::<Vec<_>>();
 
     build_sql_batches(rows.len(), max_rows, "import row", |range| {
-        generate_insert_typed(&columns, &column_types, &rows[range], table, schema, &DatabaseType::CloudflareD1)
+        generate_insert_typed(&columns, &column_types, &rows[range], table, schema, &DatabaseType::CloudflareD1, None)
     })
     .map(|batches| {
         batches.into_iter().map(|batch| ImportSqlBatch { sql: batch.sql, row_count: batch.item_count }).collect()

crates/dbx-core/src/db/cloudflare_d1/transfer_tests.rs

@@ -13,6 +13,7 @@ fn uses_sqlite_upsert_syntax() {
         "main",
         &DatabaseType::CloudflareD1,
         &[String::from("id")],
+        None,
     );
 
     assert!(sql.contains("ON CONFLICT (\"id\") DO UPDATE SET"));

crates/dbx-core/src/schema.rs

@@ -316,7 +316,7 @@ pub async fn list_doris_catalog_databases_core(
             return Ok(vec![]);
         }
     };
-    if catalog == "internal" {
+    if catalog == "internal" || catalog.eq_ignore_ascii_case("default_catalog") {
         return Ok(filter_mysql_system_databases_for_config(databases, db_config.as_ref()));
     }
     Ok(databases)
@@ -460,7 +460,8 @@ pub async fn resolve_external_doris_catalog(
     catalog: Option<&str>,
 ) -> Option<String> {
     let catalog = catalog?.trim();
-    if catalog.is_empty() || catalog == "internal" {
+    if catalog.is_empty() || catalog.eq_ignore_ascii_case("internal") || catalog.eq_ignore_ascii_case("default_catalog")
+    {
         return None;
     }
     let config = connection_config(state, connection_id).await?;

crates/dbx-core/src/sql_dialect/identifiers.rs

@@ -386,9 +386,28 @@ pub(crate) fn quote_transfer_identifier(name: &str, database_type: &DatabaseType
     }
 }
 
-pub(crate) fn qualified_transfer_table(table_name: &str, schema: &str, database_type: &DatabaseType) -> String {
+/// Qualified table name for transfer SQL.
+///
+/// * Without catalog: produces `schema.table` (or just `table` for MySQL family).
+/// * With catalog AND the database type supports external catalogs (Doris/StarRocks):
+///   produces `catalog.schema.table` — the 3-part form those engines require to
+///   address objects in an external (non-internal) catalog.
+pub(crate) fn qualified_transfer_table(
+    table_name: &str,
+    schema: &str,
+    database_type: &DatabaseType,
+    catalog: Option<&str>,
+) -> String {
     let table = quote_transfer_identifier(table_name, database_type);
-    if schema.is_empty() || matches!(database_type, DatabaseType::Mysql | DatabaseType::MongoDb | DatabaseType::Questdb)
+    if let Some(catalog) = catalog {
+        format!(
+            "{}.{}.{}",
+            quote_transfer_identifier(catalog, database_type),
+            quote_transfer_identifier(schema, database_type),
+            table
+        )
+    } else if schema.is_empty()
+        || matches!(database_type, DatabaseType::Mysql | DatabaseType::MongoDb | DatabaseType::Questdb)
     {
         table
     } else {

crates/dbx-core/src/sql_dialect/tests.rs

@@ -8,8 +8,8 @@ fn transfer_identifier_policy_preserves_legacy_output() {
     assert_eq!(quote_transfer_identifier("user`events", &DatabaseType::Doris), "`user``events`");
     assert_eq!(quote_transfer_identifier("user]events", &DatabaseType::SqlServer), "[user]]events]");
     assert_eq!(quote_transfer_identifier("user\"events", &DatabaseType::Postgres), "\"user\"\"events\"");
-    assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Hive), "`warehouse`.`events`");
-    assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Mysql), "`events`");
+    assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Hive, None), "`warehouse`.`events`");
+    assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Mysql, None), "`events`");
 }
 
 #[test]

crates/dbx-core/src/table_export.rs

@@ -229,7 +229,7 @@ fn table_cursor_sql(
     col_names: &[String],
     primary_keys: &[String],
 ) -> String {
-    let full_table = qualified_table(&request.table_name, request.schema.as_deref().unwrap_or(""), db_type);
+    let full_table = qualified_table(&request.table_name, request.schema.as_deref().unwrap_or(""), db_type, None);
     let col_list = col_names.iter().map(|column| quote_identifier(column, db_type)).collect::<Vec<_>>().join(", ");
     let predicate = crate::sql_dialect::normalize_where_input(request.where_input.as_deref());
     let where_clause = if predicate.is_empty() { String::new() } else { format!(" WHERE ({predicate})") };
@@ -1155,6 +1155,7 @@ async fn export_table_data_core_inner(
             request.schema.as_deref().unwrap_or(""),
             &db_type,
             request.where_input.as_deref(),
+            None,
         );
         match execute_table_export_count(state, &pool_key, request, &count_query, cancel_token.clone()).await {
             Ok(result) => result

crates/dbx-core/src/table_import.rs

@@ -3068,7 +3068,8 @@ fn build_import_insert_batch_with_plan(
         return Ok(None);
     }
     let mapped_rows = map_import_rows_with_plan(rows, plan, db_type, kingbase_oracle_mode, date_time_format);
-    let sql = generate_insert_typed(&plan.target_columns, &plan.column_types, &mapped_rows, table, schema, db_type);
+    let sql =
+        generate_insert_typed(&plan.target_columns, &plan.column_types, &mapped_rows, table, schema, db_type, None);
     Ok((!sql.trim().is_empty()).then_some(ImportSqlBatch { sql, row_count: rows.len() }))
 }
 
@@ -3093,6 +3094,7 @@ fn build_import_insert_batches_with_plan(
         table,
         schema,
         db_type,
+        None,
         SqlBatchLimits::for_database(db_type, rows.len()).with_hard_sql_bytes(hard_sql_bytes),
     )?;
     Ok(batches.into_iter().map(|(sql, row_count)| ImportSqlBatch { sql, row_count }).collect())
@@ -3359,7 +3361,7 @@ fn build_import_insert_batches_with_format(
 }
 
 pub fn truncate_sql(table: &str, schema: &str, db_type: &DatabaseType) -> String {
-    let full_table = qualified_table(table, schema, db_type);
+    let full_table = qualified_table(table, schema, db_type, None);
     match db_type {
         DatabaseType::Sqlite | DatabaseType::CloudflareD1 => format!("DELETE FROM {full_table}"),
         _ => format!("TRUNCATE TABLE {full_table}"),
@@ -3643,7 +3645,7 @@ pub fn build_import_create_table_plan(
         return Err("No columns mapped for import".to_string());
     }
 
-    let full_table = qualified_table(table.trim(), schema, db_type);
+    let full_table = qualified_table(table.trim(), schema, db_type, None);
     let column_sql = columns
         .iter()
         .map(|column| format!("{} {}", quote_identifier(&column.name, db_type), column.data_type))
@@ -3875,7 +3877,7 @@ fn build_postgres_copy_text_row(
 }
 
 fn postgres_copy_sql(plan: &CompiledImportPlan, table: &str, schema: &str) -> String {
-    let table = qualified_table(table, schema, &DatabaseType::Postgres);
+    let table = qualified_table(table, schema, &DatabaseType::Postgres, None);
     let columns = plan
         .target_columns
         .iter()
@@ -4630,7 +4632,7 @@ fn compile_sqlserver_bulk_import_plan(
         target_types.push(column.column.data_type.clone());
     }
     Ok(SqlServerBulkImportPlan {
-        target_table: qualified_table(table, schema, &DatabaseType::SqlServer),
+        target_table: qualified_table(table, schema, &DatabaseType::SqlServer, None),
         target_columns: import_plan.target_columns.clone(),
         target_types,
         requires_identity_insert,
@@ -5185,6 +5187,7 @@ where
             &request.database,
             &request.schema,
             &request.table,
+            None,
         )
         .await
         .unwrap_or_default()
@@ -5535,6 +5538,7 @@ where
             &request.database,
             &request.schema,
             &request.table,
+            None,
         )
         .await
         .unwrap_or_default()
@@ -6012,6 +6016,7 @@ where
         &request.database,
         &request.schema,
         &request.table,
+        None,
     )
     .await
     .unwrap_or_default()

crates/dbx-core/tests/database_capabilities.rs

@@ -182,8 +182,9 @@ fn identifies_single_connection_pool_types() {
 #[test]
 fn identifies_metadata_connections_that_drop_database_scope() {
     assert!(is_metadata_connection_scoped(&DatabaseType::Mysql));
-    assert!(!is_metadata_connection_scoped(&DatabaseType::Doris));
-    assert!(!is_metadata_connection_scoped(&DatabaseType::StarRocks));
+    assert!(is_metadata_connection_scoped(&DatabaseType::Doris));
+    assert!(is_metadata_connection_scoped(&DatabaseType::StarRocks));
+    assert!(is_metadata_connection_scoped(&DatabaseType::ManticoreSearch));
     assert!(!is_metadata_connection_scoped(&DatabaseType::Postgres));
     assert!(!is_metadata_connection_scoped(&DatabaseType::Kwdb));
     assert!(!is_metadata_connection_scoped(&DatabaseType::Oracle));

crates/dbx-core/tests/live_postgres_transfer.rs

@@ -211,9 +211,11 @@ async fn live_postgres_transfer_preserves_data_and_schema_objects() {
         source_connection_id: source_connection_id.to_string(),
         source_database: source_database.clone(),
         source_schema: source_schema.clone(),
+        source_catalog: None,
         target_connection_id: target_connection_id.to_string(),
         target_database: target_database.clone(),
         target_schema: target_schema.clone(),
+        target_catalog: None,
         tables: vec!["users".to_string(), "audit_logs".to_string(), "files".to_string()],
         create_table: true,
         mode: TransferMode::Append,
@@ -492,9 +494,11 @@ async fn live_postgres_transfer_skips_create_ddl_for_existing_target_table() {
         source_connection_id: source_connection_id.to_string(),
         source_database: source_database.clone(),
         source_schema: source_schema.clone(),
+        source_catalog: None,
         target_connection_id: target_connection_id.to_string(),
         target_database: target_database.clone(),
         target_schema: target_schema.clone(),
+        target_catalog: None,
         tables: vec!["items".to_string()],
         create_table: true,
         mode: TransferMode::Append,

crates/dbx-core/tests/live_sqlserver_completion.rs

@@ -1354,9 +1354,11 @@ async fn live_sqlserver_transfer_table_skips_rowversion_insert_column() {
         source_connection_id: "live-sqlserver-rowversion".to_string(),
         source_database: database.clone(),
         source_schema: "dbo".to_string(),
+        source_catalog: None,
         target_connection_id: "live-sqlserver-rowversion".to_string(),
         target_database: target_database.clone(),
         target_schema: "dbo".to_string(),
+        target_catalog: None,
         tables: vec![source_table.clone()],
         create_table: true,
         mode: dbx_core::transfer::TransferMode::Append,