fix(kingbase): quote MySQL-compatible identifiers

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

@@ -21,12 +21,21 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
     }
 
     // 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 table = if database_type == Some(DatabaseType::Kingbase) {
+        table_data_qualified_table_name(
+            database_type,
+            options.schema.as_deref(),
+            &options.table_name,
+            options.identifier_quote.as_deref(),
+        )
+    } else {
+        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) {
@@ -143,6 +152,40 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
     }
 }
 
+pub(crate) fn table_data_qualified_table_name(
+    database_type: Option<DatabaseType>,
+    schema: Option<&str>,
+    table_name: &str,
+    identifier_quote: Option<&str>,
+) -> String {
+    if database_type != Some(DatabaseType::Kingbase) {
+        return qualified_table_name(database_type, schema, table_name);
+    }
+    let table = quote_table_data_identifier(database_type, table_name, identifier_quote);
+    schema
+        .map(str::trim)
+        .filter(|schema| !schema.is_empty())
+        .map(|schema| format!("{}.{}", quote_table_data_identifier(database_type, schema, identifier_quote), table))
+        .unwrap_or(table)
+}
+
+fn quote_table_data_identifier(
+    database_type: Option<DatabaseType>,
+    name: &str,
+    identifier_quote: Option<&str>,
+) -> String {
+    if database_type != Some(DatabaseType::Kingbase) {
+        return quote_table_identifier(database_type, name);
+    }
+    let Some(quote) = identifier_quote else {
+        return quote_table_identifier(database_type, name);
+    };
+    if quote.is_empty() {
+        return name.to_string();
+    }
+    format!("{quote}{}{quote}", name.replace(quote, &format!("{quote}{quote}")))
+}
+
 fn is_view_table_type(table_type: Option<&str>) -> bool {
     table_type.is_some_and(|value| value.to_ascii_uppercase().contains("VIEW"))
 }
@@ -418,6 +461,7 @@ mod tests {
     fn opts(database_type: DatabaseType, catalog: Option<&str>, table: &str) -> TableDataSelectSqlOptions {
         TableDataSelectSqlOptions {
             database_type: Some(database_type),
+            identifier_quote: None,
             schema: None,
             table_name: table.to_string(),
             catalog: catalog.map(|c| c.to_string()),

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

@@ -21,6 +21,11 @@ fn quotes_identifiers_by_database_type() {
     assert_eq!(quote_table_identifier(Some(DatabaseType::StarRocks), "user`name"), "`user``name`");
     assert_eq!(quote_table_identifier(Some(DatabaseType::SqlServer), "user]name"), "[user]]name]");
     assert_eq!(quote_table_identifier(Some(DatabaseType::Postgres), "user\"name"), "\"user\"\"name\"");
+    assert_eq!(quote_table_identifier(Some(DatabaseType::Kingbase), "cqbq_ls"), "\"cqbq_ls\"");
+    assert_eq!(quote_table_identifier(Some(DatabaseType::Kingbase), "actionlogs"), "\"actionlogs\"");
+    assert_eq!(quote_table_identifier(Some(DatabaseType::Kingbase), "order"), "\"order\"");
+    assert_eq!(quote_table_identifier(Some(DatabaseType::Kingbase), "MixedCase"), "\"MixedCase\"");
+    assert_eq!(quote_table_identifier(Some(DatabaseType::Kingbase), "order detail"), "\"order detail\"");
     assert_eq!(quote_table_identifier(Some(DatabaseType::Informix), "users_1"), "users_1");
     assert_eq!(quote_table_identifier(Some(DatabaseType::Jdbc), "users_1"), "users_1");
     assert_eq!(quote_table_identifier(Some(DatabaseType::Jdbc), "user name"), "user name");
@@ -31,6 +36,10 @@ fn quotes_identifiers_by_database_type() {
 fn qualifies_schema_only_for_schema_aware_databases() {
     assert_eq!(qualified_table_name(Some(DatabaseType::Postgres), Some("public"), "users"), "\"public\".\"users\"");
     assert_eq!(qualified_table_name(Some(DatabaseType::Kwdb), Some("public"), "users"), "\"public\".\"users\"");
+    assert_eq!(
+        qualified_table_name(Some(DatabaseType::Kingbase), Some("cqbq_ls"), "actionlogs"),
+        "\"cqbq_ls\".\"actionlogs\""
+    );
     assert_eq!(qualified_table_name(Some(DatabaseType::Mysql), Some("public"), "users"), "`public`.`users`");
     assert_eq!(qualified_table_name(Some(DatabaseType::Goldendb), Some("public"), "users"), "`public`.`users`");
     assert_eq!(qualified_table_name(Some(DatabaseType::StarRocks), Some("warehouse"), "users"), "`warehouse`.`users`");
@@ -244,6 +253,10 @@ fn builds_select_sql_with_limit_syntax_for_database_type() {
 
 #[test]
 fn builds_table_data_where_and_schema_queries() {
+    assert_eq!(
+        build_count_table_sql(Some(DatabaseType::Kingbase), Some("cqbq_ls"), "actionlogs"),
+        "SELECT COUNT(*) AS row_count FROM \"cqbq_ls\".\"actionlogs\""
+    );
     assert_eq!(
         build_table_data_select_sql(TableDataSelectSqlOptions {
             database_type: Some(DatabaseType::Mysql),
@@ -298,6 +311,36 @@ fn builds_table_data_where_and_schema_queries() {
         }),
         "SELECT * FROM \"public\".\"orders\" WHERE (amount > 10) LIMIT 50 OFFSET 100;"
     );
+    assert_eq!(
+        build_table_data_select_sql(TableDataSelectSqlOptions {
+            database_type: Some(DatabaseType::Kingbase),
+            identifier_quote: Some("`".to_string()),
+            schema: Some("cqbq_ls".to_string()),
+            table_name: "actionlogs".to_string(),
+            table_type: None,
+            primary_keys: Vec::new(),
+            columns: Vec::new(),
+            fallback_order_columns: Vec::new(),
+            order_by: None,
+            limit: Some(100),
+            offset: None,
+            where_input: None,
+            include_row_id: false,
+            ..Default::default()
+        }),
+        "SELECT * FROM `cqbq_ls`.`actionlogs` LIMIT 100;"
+    );
+    assert_eq!(
+        build_table_data_select_sql(TableDataSelectSqlOptions {
+            database_type: Some(DatabaseType::Kingbase),
+            identifier_quote: Some("\"".to_string()),
+            schema: Some("App Schema".to_string()),
+            table_name: "ANALYZE".to_string(),
+            limit: Some(100),
+            ..Default::default()
+        }),
+        "SELECT * FROM \"App Schema\".\"ANALYZE\" LIMIT 100;"
+    );
     assert_eq!(
         build_table_data_select_sql(TableDataSelectSqlOptions {
             database_type: Some(DatabaseType::Xugu),

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

@@ -22,6 +22,8 @@ pub struct TableDataSelectSqlOptions {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub database_type: Option<DatabaseType>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub identifier_quote: Option<String>,
+    #[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,