chore(query): Compatible with the MySQL BI Ecosystem (#18909)

* fix(query): jdbc test

* test

* fix

* fix

* next

* test

* fix load

* fix

* try

* bump opensrv-mysql 0.9.0

* fix

* fix ci

Author: TCeason

Cargo.lock

@@ -418,7 +418,7 @@ openraft = { version = "0.10.0", features = [
     "serde",
     "tracing-log",
 ] }
-opensrv-mysql = { git = "https://github.com/databendlabs/opensrv.git", rev = "a1fb4da", features = ["tls"] }
+opensrv-mysql = { git = "https://github.com/databendlabs/opensrv.git", tag = "v0.9.0", features = ["tls"] }
 orc-rust = "0.6.0"
 ordered-float = { version = "5.0.0", default-features = false }
 ordq = "0.2.0"

Cargo.toml

@@ -20,9 +20,11 @@ use databend_common_meta_app::schema::DatabaseInfo;
 use databend_common_meta_app::schema::DatabaseMeta;
 use databend_common_meta_app::tenant::Tenant;
 use databend_common_meta_types::SeqV;
+use databend_common_storages_information_schema::CharacterSetsTable;
 use databend_common_storages_information_schema::ColumnsTable;
 use databend_common_storages_information_schema::KeyColumnUsageTable;
 use databend_common_storages_information_schema::KeywordsTable;
+use databend_common_storages_information_schema::ReferentialConstraintsTable;
 use databend_common_storages_information_schema::SchemataTable;
 use databend_common_storages_information_schema::StatisticsTable;
 use databend_common_storages_information_schema::TablesTable;
@@ -41,12 +43,14 @@ impl InformationSchemaDatabase {
     pub fn create(sys_db_meta: &mut InMemoryMetas, ctl_name: &str) -> Self {
         let table_list: Vec<Arc<dyn Table>> = vec![
             ColumnsTable::create(sys_db_meta.next_table_id(), ctl_name),
+            CharacterSetsTable::create(sys_db_meta.next_table_id(), ctl_name),
             TablesTable::create(sys_db_meta.next_table_id(), ctl_name),
             KeywordsTable::create(sys_db_meta.next_table_id(), ctl_name),
             ViewsTable::create(sys_db_meta.next_table_id(), ctl_name),
             SchemataTable::create(sys_db_meta.next_table_id(), ctl_name),
             StatisticsTable::create(sys_db_meta.next_table_id(), ctl_name),
             KeyColumnUsageTable::create(sys_db_meta.next_table_id(), ctl_name),
+            ReferentialConstraintsTable::create(sys_db_meta.next_table_id(), ctl_name),
         ];
 
         let db = "information_schema";

src/query/service/src/databases/information_schema/information_schema_database.rs

@@ -64,6 +64,77 @@ impl MySQLFederated {
         Some((schema, block))
     }
 
+    // Returns empty result set with MySQL-compatible schema for SHOW KEYS/INDEX commands.
+    // Required for PowerBI + MySQL ODBC 9.5 compatibility.
+    // Reference: https://dev.mysql.com/doc/refman/8.0/en/show-index.html
+    fn show_keys_block() -> Option<(TableSchemaRef, DataBlock)> {
+        use databend_common_expression::types::number::*;
+
+        let schema = TableSchemaRefExt::create(vec![
+            TableField::new("Table", TableDataType::String),
+            TableField::new(
+                "Non_unique",
+                TableDataType::Number(databend_common_expression::types::NumberDataType::Int32),
+            ),
+            TableField::new("Key_name", TableDataType::String),
+            TableField::new(
+                "Seq_in_index",
+                TableDataType::Number(databend_common_expression::types::NumberDataType::Int32),
+            ),
+            TableField::new("Column_name", TableDataType::String),
+            TableField::new(
+                "Collation",
+                TableDataType::Nullable(Box::new(TableDataType::String)),
+            ),
+            TableField::new(
+                "Cardinality",
+                TableDataType::Nullable(Box::new(TableDataType::Number(
+                    databend_common_expression::types::NumberDataType::Int64,
+                ))),
+            ),
+            TableField::new(
+                "Sub_part",
+                TableDataType::Nullable(Box::new(TableDataType::Number(
+                    databend_common_expression::types::NumberDataType::Int64,
+                ))),
+            ),
+            TableField::new(
+                "Packed",
+                TableDataType::Nullable(Box::new(TableDataType::String)),
+            ),
+            TableField::new("Null", TableDataType::String),
+            TableField::new("Index_type", TableDataType::String),
+            TableField::new("Comment", TableDataType::String),
+            TableField::new("Index_comment", TableDataType::String),
+            TableField::new("Visible", TableDataType::String),
+            TableField::new(
+                "Expression",
+                TableDataType::Nullable(Box::new(TableDataType::String)),
+            ),
+        ]);
+
+        // Create empty block with proper column types
+        let block = DataBlock::new_from_columns(vec![
+            StringType::from_data(Vec::<String>::new()), // Table
+            Int32Type::from_data(Vec::<i32>::new()),     // Non_unique
+            StringType::from_data(Vec::<String>::new()), // Key_name
+            Int32Type::from_data(Vec::<i32>::new()),     // Seq_in_index
+            StringType::from_data(Vec::<String>::new()), // Column_name
+            StringType::from_data_with_validity(Vec::<&str>::new(), Vec::<bool>::new()), /* Collation */
+            Int64Type::from_data_with_validity(Vec::<i64>::new(), Vec::<bool>::new()), /* Cardinality */
+            Int64Type::from_data_with_validity(Vec::<i64>::new(), Vec::<bool>::new()), // Sub_part
+            StringType::from_data_with_validity(Vec::<&str>::new(), Vec::<bool>::new()), // Packed
+            StringType::from_data(Vec::<String>::new()),                               // Null
+            StringType::from_data(Vec::<String>::new()), // Index_type
+            StringType::from_data(Vec::<String>::new()), // Comment
+            StringType::from_data(Vec::<String>::new()), // Index_comment
+            StringType::from_data(Vec::<String>::new()), // Visible
+            StringType::from_data_with_validity(Vec::<&str>::new(), Vec::<bool>::new()), /* Expression */
+        ]);
+
+        Some((schema, block))
+    }
+
     // SELECT @@aa, @@bb as cc, @dd...
     // Block is built by the variables.
     fn select_variable_data_block(query: &str) -> Option<(TableSchemaRef, DataBlock)> {
@@ -217,6 +288,9 @@ impl MySQLFederated {
             (Regex::new("(?i)^(LOCK TABLES FOR BACKUP)").unwrap(), None),
             (Regex::new("(?i)^(UNLOCK BINLOG(.*))").unwrap(), None),
             (Regex::new("(?i)^(/\\*!40101 SET(.*) \\*/)$").unwrap(), None),
+            // PowerBI.
+            (Regex::new("(?i)^(SET SQL_AUTO_IS_NULL(.*))").unwrap(), None),
+            (Regex::new("(?i)^(SHOW KEYS FROM(.*))").unwrap(), MySQLFederated::show_keys_block()),
             // DBeaver.
             (Regex::new("(?i)^(SHOW WARNINGS)").unwrap(), None),
             (Regex::new("(?i)^(/\\* ApplicationName=(.*)SHOW WARNINGS)").unwrap(), None),

src/query/service/src/servers/mysql/mysql_federated.rs

@@ -105,7 +105,7 @@ impl<W: AsyncWrite + Send + Sync + Unpin> AsyncMysqlShim<W> for InteractiveWorke
     }
 
     #[async_backtrace::framed]
-    async fn auth_plugin_for_username(&self, _user: &[u8]) -> &str {
+    async fn auth_plugin_for_username(&self, _user: &[u8]) -> &'static str {
         "mysql_native_password"
     }
 
@@ -218,7 +218,6 @@ impl<W: AsyncWrite + Send + Sync + Unpin> AsyncMysqlShim<W> for InteractiveWorke
         let query_id = Uuid::now_v7();
         // Ensure the query id shares the same representation as trace_id.
         let query_id_str = query_id.simple().to_string();
-
         let sampled =
             thread_rng().gen_range(0..100) <= self.base.session.get_trace_sample_rate()?;
         let span_context =
@@ -402,7 +401,11 @@ impl InteractiveWorkerBase {
                 if data_block.num_rows() > 0 {
                     info!("Federated response: {:?}", data_block);
                 }
-                let has_result = data_block.num_rows() > 0;
+                // Fix for MySQL ODBC 9.5 + PowerBI compatibility:
+                // Empty result sets (0 rows but with schema) must set has_result=true,
+                // otherwise ODBC throws "invalid cursor state" error on commands like "SHOW KEYS FROM table".
+                // Check for column definitions instead of row count.
+                let has_result = !schema.fields().is_empty() || data_block.num_columns() > 0;
                 Ok((
                     QueryResult::create(
                         DataBlockStream::create(None, vec![data_block]).boxed(),

src/query/service/src/servers/mysql/mysql_interactive_worker.rs

@@ -201,10 +201,55 @@ impl<'a, W: AsyncWrite + Send + Unpin> DFQueryResultWriter<'a, W> {
             }
         }
 
+        fn compute_length(data_type: &DataType) -> u32 {
+            match data_type {
+                DataType::Null | DataType::EmptyArray | DataType::EmptyMap => 0,
+                DataType::Boolean => 1,
+                DataType::Binary => 16382,
+                DataType::String => 16382 * 4,
+                DataType::Number(num_ty) => match num_ty {
+                    NumberDataType::Int8 | NumberDataType::UInt8 => 3,
+                    NumberDataType::Int16 | NumberDataType::UInt16 => 5,
+                    NumberDataType::Int32 | NumberDataType::UInt32 => 10,
+                    NumberDataType::Int64 => 19,
+                    NumberDataType::UInt64 => 20,
+                    NumberDataType::Float32 => 12,
+                    NumberDataType::Float64 => 22,
+                },
+                DataType::Decimal(size) => size.precision() as u32,
+                DataType::Date => 10,
+                DataType::Timestamp => 26,
+                DataType::Interval => 64,
+                DataType::Geometry | DataType::Geography => 1024,
+                DataType::Vector(_) => 1024,
+                DataType::Bitmap | DataType::Variant | DataType::StageLocation => 1024,
+                DataType::Array(inner) | DataType::Map(inner) => {
+                    let inner_len = compute_length(inner);
+                    (inner_len.saturating_mul(4)).min(16382)
+                }
+                DataType::Tuple(fields) => fields
+                    .iter()
+                    .map(compute_length)
+                    .max()
+                    .unwrap_or(1024)
+                    .min(1024),
+                DataType::Opaque(_) => 1024,
+                DataType::Nullable(inner) => compute_length(inner),
+                DataType::Generic(_) => 1024,
+                DataType::TimestampTz => 1024,
+            }
+        }
+
+        fn column_length(field: &DataField) -> u32 {
+            let length = compute_length(field.data_type());
+            length.clamp(1, 16382)
+        }
+
         fn make_column_from_field(field: &DataField) -> Result<Column> {
             convert_field_type(field).map(|column_type| Column {
                 table: "".to_string(),
                 column: field.name().to_string(),
+                collen: column_length(field),
                 coltype: column_type,
                 colflags: ColumnFlags::empty(),
             })

src/query/service/src/servers/mysql/writers/query_result_writer.rs

@@ -32,6 +32,8 @@ use databend_common_meta_app::storage::StorageFsConfig;
 use databend_common_meta_app::storage::StorageParams;
 use databend_common_meta_app::storage::StorageS3Config;
 use databend_common_sql::executor::table_read_plan::ToReadDataSourcePlan;
+use databend_common_storages_basic::view_table::QUERY;
+use databend_common_storages_information_schema::CharacterSetsTable;
 use databend_common_storages_system::BuildOptionsTable;
 use databend_common_storages_system::CachesTable;
 use databend_common_storages_system::CatalogsTable;
@@ -139,6 +141,21 @@ async fn test_columns_table() -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn test_information_schema_character_sets_table_metadata() -> Result<()> {
+    let table = CharacterSetsTable::create(1, "default");
+    let table_info = table.get_table_info();
+    assert_eq!(table_info.name, "character_sets");
+    let query = table_info
+        .meta
+        .options
+        .get(QUERY)
+        .expect("view query must be set");
+    assert!(query.contains("utf8mb4"));
+    assert!(query.contains("character_set_name"));
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_clusters_table() -> Result<()> {
     let fixture = TestFixture::setup().await?;