feat(query): add table statistics admin api (#18967)

Author: zhang2014

src/query/service/src/servers/admin/admin_service.rs

@@ -110,6 +110,10 @@ impl AdminService {
                 .at(
                     "/v1/tenants/:tenant/user_functions",
                     get(super::v1::user_functions::user_functions),
+                )
+                .at(
+                    "/v1/tenants/:tenant/databases/:database/tables/:table/stats",
+                    get(super::v1::table_statistics::get_table_stats_handler),
                 );
         }
 

src/query/service/src/servers/admin/v1/mod.rs

@@ -20,6 +20,7 @@ pub mod query_dump;
 pub mod settings;
 pub mod stream_status;
 pub mod system;
+pub mod table_statistics;
 pub mod tenant_table_stats;
 pub mod tenant_tables;
 pub mod user_functions;

src/query/service/src/servers/admin/v1/table_statistics.rs

@@ -0,0 +1,63 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use databend_common_catalog::catalog::CatalogManager;
+use databend_common_catalog::session_type::SessionType;
+use databend_common_catalog::table_context::TableContext;
+use databend_common_exception::Result;
+use databend_common_meta_app::tenant::Tenant;
+use databend_common_storages_system::StatisticsTable;
+use databend_common_storages_system::TableColumnStatistics;
+use poem::web::Json;
+use poem::web::Path;
+use poem::IntoResponse;
+
+use crate::sessions::SessionManager;
+
+#[poem::handler]
+#[async_backtrace::framed]
+pub async fn get_table_stats_handler(
+    Path((tenant, database, table)): Path<(String, String, String)>,
+) -> poem::Result<impl IntoResponse> {
+    match dump_columns(&tenant, &database, &table).await {
+        Ok(columns) => Ok(Json(columns)),
+        Err(error) => Err(poem::error::InternalServerError(error)),
+    }
+}
+
+async fn dump_columns(
+    tenant: &str,
+    database: &str,
+    table: &str,
+) -> Result<Vec<TableColumnStatistics>> {
+    let dummy_session = SessionManager::instance()
+        .create_session(SessionType::Dummy)
+        .await?;
+
+    let session = SessionManager::instance().register_session(dummy_session)?;
+
+    let ctx: Arc<dyn TableContext> = session
+        .create_query_context(&databend_common_version::BUILD_INFO)
+        .await?;
+
+    let catalog = CatalogManager::instance().get_default_catalog(Default::default())?;
+
+    let table = catalog
+        .get_table(&Tenant::new_literal(tenant), database, table)
+        .await?;
+
+    StatisticsTable::dump_table_columns(&ctx, &catalog, database, &table).await
+}

src/query/storages/system/src/lib.rs

@@ -118,6 +118,7 @@ pub use roles_table::RolesTable;
 pub use settings_table::SettingsTable;
 pub use stages_table::StagesTable;
 pub use statistics_table::StatisticsTable;
+pub use statistics_table::TableColumnStatistics;
 pub use streams_table::FullStreamsTable;
 pub use streams_table::TerseStreamsTable;
 pub use table::SyncOneBlockSystemTable;

src/query/storages/system/src/statistics_table.rs

@@ -51,8 +51,8 @@ pub struct StatisticsTable {
     table_info: TableInfo,
 }
 
-#[derive(Default)]
-struct TableColumnStatistics {
+#[derive(Default, serde::Serialize, serde::Deserialize)]
+pub struct TableColumnStatistics {
     database_name: String,
     table_name: String,
     column_name: String,
@@ -119,7 +119,7 @@ impl StatisticsTable {
     }
 
     #[async_backtrace::framed]
-    async fn dump_table_columns(
+    async fn dump_tables_columns(
         &self,
         ctx: Arc<dyn TableContext>,
         push_downs: Option<PushDownInfo>,
@@ -129,139 +129,146 @@ impl StatisticsTable {
         let mut rows: Vec<TableColumnStatistics> = vec![];
         for (database, tables) in database_and_tables {
             for table in tables {
-                match table.engine() {
-                    VIEW_ENGINE => {
-                        let fields = if let Some(query) = table.options().get(QUERY) {
-                            let mut planner = Planner::new(ctx.clone());
-                            match planner.plan_sql(query).await {
-                                Ok((plan, _)) => {
-                                    infer_table_schema(&plan.schema())?.fields().clone()
-                                }
-                                Err(e) => {
-                                    // If VIEW SELECT QUERY plan err, should return empty. not destroy the query.
-                                    warn!(
-                                        "failed to get columns for {}: {}",
-                                        table.get_table_info().desc,
-                                        e
-                                    );
-                                    vec![]
-                                }
-                            }
-                        } else {
+                rows.extend(Self::dump_table_columns(&ctx, catalog, &database, &table).await?);
+            }
+        }
+        Ok(rows)
+    }
+
+    pub async fn dump_table_columns(
+        ctx: &Arc<dyn TableContext>,
+        catalog: &Arc<dyn Catalog>,
+        database: &str,
+        table: &Arc<dyn Table>,
+    ) -> Result<Vec<TableColumnStatistics>> {
+        let mut columns = vec![];
+        match table.engine() {
+            VIEW_ENGINE => {
+                let fields = if let Some(query) = table.options().get(QUERY) {
+                    let mut planner = Planner::new(ctx.clone());
+                    match planner.plan_sql(query).await {
+                        Ok((plan, _)) => infer_table_schema(&plan.schema())?.fields().clone(),
+                        Err(e) => {
+                            // If VIEW SELECT QUERY plan err, should return empty. not destroy the query.
+                            warn!(
+                                "failed to get columns for {}: {}",
+                                table.get_table_info().desc,
+                                e
+                            );
                             vec![]
-                        };
-                        for field in fields {
-                            rows.push(TableColumnStatistics {
-                                database_name: database.clone(),
-                                table_name: table.name().into(),
-                                column_name: field.name,
-                                ..Default::default()
-                            })
                         }
                     }
-                    _ => {
-                        let schema = table.schema();
-                        // attach table should not collect statistics, source table column already collect them.
-                        let columns_statistics = if !FuseTable::is_table_attached(
-                            &table.get_table_info().meta.options,
+                } else {
+                    vec![]
+                };
+                for field in fields {
+                    columns.push(TableColumnStatistics {
+                        database_name: database.to_string(),
+                        table_name: table.name().into(),
+                        column_name: field.name,
+                        ..Default::default()
+                    })
+                }
+            }
+            _ => {
+                let schema = table.schema();
+                // attach table should not collect statistics, source table column already collect them.
+                let columns_statistics =
+                    if !FuseTable::is_table_attached(&table.get_table_info().meta.options) {
+                        table
+                            .column_statistics_provider(ctx.clone())
+                            .await
+                            .unwrap_or_else(|e| {
+                                let msg = format!(
+                                    "Collect {}.{}.{} column statistics with error: {}",
+                                    catalog.name(),
+                                    database,
+                                    table.name(),
+                                    e
+                                );
+                                warn!("{}", msg);
+                                ctx.push_warning(msg);
+                                Box::new(DummyColumnStatisticsProvider)
+                            })
+                    } else {
+                        Box::new(DummyColumnStatisticsProvider)
+                    };
+                let stats_row_count = columns_statistics.stats_num_rows();
+                let actual_row_count = columns_statistics.num_rows();
+                for field in schema.fields() {
+                    let column_id = field.column_id;
+                    let column_statistics = columns_statistics.column_statistics(column_id);
+                    let his_info = columns_statistics.histogram(column_id);
+                    let histogram = if let Some(his_info) = his_info {
+                        let mut his_infos = vec![];
+                        for (i, bucket) in his_info.buckets.iter().enumerate() {
+                            let min = bucket.lower_bound().to_string()?;
+                            let max = bucket.upper_bound().to_string()?;
+                            let ndv = bucket.num_distinct();
+                            let count = bucket.num_values();
+                            let his_info = format!(
+                                "[bucket id: {:?}, min: {:?}, max: {:?}, ndv: {:?}, count: {:?}]",
+                                i, min, max, ndv, count
+                            );
+                            his_infos.push(his_info);
+                        }
+                        his_infos.join(", ")
+                    } else {
+                        "".to_string()
+                    };
+                    columns.push(TableColumnStatistics {
+                        database_name: database.to_string(),
+                        table_name: table.name().into(),
+                        column_name: field.name().clone(),
+                        stats_row_count,
+                        actual_row_count,
+                        distinct_count: column_statistics.and_then(|v| v.ndv),
+                        null_count: column_statistics.map(|v| v.null_count),
+                        min: column_statistics
+                            .and_then(|s| s.min.clone())
+                            .map(|v| v.to_string().unwrap()),
+                        max: column_statistics
+                            .and_then(|s| s.max.clone())
+                            .map(|v| v.to_string().unwrap()),
+                        avg_size: columns_statistics.average_size(column_id),
+                        histogram,
+                    })
+                }
+                // add virtual column statistics
+                let table_info = table.get_table_info();
+                if let Some(virtual_schema) = &table_info.meta.virtual_schema {
+                    for virtual_field in virtual_schema.fields() {
+                        if let (Ok(source_field), Some(column_statistics)) = (
+                            schema.field_of_column_id(virtual_field.source_column_id),
+                            columns_statistics.column_statistics(virtual_field.column_id),
                         ) {
-                            table
-                                .column_statistics_provider(ctx.clone())
-                                .await
-                                .unwrap_or_else(|e| {
-                                    let msg = format!(
-                                        "Collect {}.{}.{} column statistics with error: {}",
-                                        catalog.name(),
-                                        database,
-                                        table.name(),
-                                        e
-                                    );
-                                    warn!("{}", msg);
-                                    ctx.push_warning(msg);
-                                    Box::new(DummyColumnStatisticsProvider)
-                                })
-                        } else {
-                            Box::new(DummyColumnStatisticsProvider)
-                        };
-                        let stats_row_count = columns_statistics.stats_num_rows();
-                        let actual_row_count = columns_statistics.num_rows();
-                        for field in schema.fields() {
-                            let column_id = field.column_id;
-                            let column_statistics = columns_statistics.column_statistics(column_id);
-                            let his_info = columns_statistics.histogram(column_id);
-                            let histogram = if let Some(his_info) = his_info {
-                                let mut his_infos = vec![];
-                                for (i, bucket) in his_info.buckets.iter().enumerate() {
-                                    let min = bucket.lower_bound().to_string()?;
-                                    let max = bucket.upper_bound().to_string()?;
-                                    let ndv = bucket.num_distinct();
-                                    let count = bucket.num_values();
-                                    let his_info = format!(
-                                        "[bucket id: {:?}, min: {:?}, max: {:?}, ndv: {:?}, count: {:?}]",
-                                        i, min, max, ndv, count
-                                    );
-                                    his_infos.push(his_info);
-                                }
-                                his_infos.join(", ")
-                            } else {
-                                "".to_string()
-                            };
-                            rows.push(TableColumnStatistics {
-                                database_name: database.clone(),
+                            let column_name =
+                                format!("{}{}", source_field.name, virtual_field.name);
+                            columns.push(TableColumnStatistics {
+                                database_name: database.to_string(),
                                 table_name: table.name().into(),
-                                column_name: field.name().clone(),
+                                column_name,
                                 stats_row_count,
                                 actual_row_count,
-                                distinct_count: column_statistics.and_then(|v| v.ndv),
-                                null_count: column_statistics.map(|v| v.null_count),
+                                distinct_count: column_statistics.ndv,
+                                null_count: Some(column_statistics.null_count),
                                 min: column_statistics
-                                    .and_then(|s| s.min.clone())
+                                    .min
+                                    .clone()
                                     .map(|v| v.to_string().unwrap()),
                                 max: column_statistics
-                                    .and_then(|s| s.max.clone())
+                                    .max
+                                    .clone()
                                     .map(|v| v.to_string().unwrap()),
-                                avg_size: columns_statistics.average_size(column_id),
-                                histogram,
+                                avg_size: columns_statistics.average_size(virtual_field.column_id),
+                                histogram: "".to_string(),
                             })
                         }
-                        // add virtual column statistics
-                        let table_info = table.get_table_info();
-                        if let Some(virtual_schema) = &table_info.meta.virtual_schema {
-                            for virtual_field in virtual_schema.fields() {
-                                if let (Ok(source_field), Some(column_statistics)) = (
-                                    schema.field_of_column_id(virtual_field.source_column_id),
-                                    columns_statistics.column_statistics(virtual_field.column_id),
-                                ) {
-                                    let column_name =
-                                        format!("{}{}", source_field.name, virtual_field.name);
-                                    rows.push(TableColumnStatistics {
-                                        database_name: database.clone(),
-                                        table_name: table.name().into(),
-                                        column_name,
-                                        stats_row_count,
-                                        actual_row_count,
-                                        distinct_count: column_statistics.ndv,
-                                        null_count: Some(column_statistics.null_count),
-                                        min: column_statistics
-                                            .min
-                                            .clone()
-                                            .map(|v| v.to_string().unwrap()),
-                                        max: column_statistics
-                                            .max
-                                            .clone()
-                                            .map(|v| v.to_string().unwrap()),
-                                        avg_size: columns_statistics
-                                            .average_size(virtual_field.column_id),
-                                        histogram: "".to_string(),
-                                    })
-                                }
-                            }
-                        }
                     }
                 }
             }
         }
-        Ok(rows)
+        Ok(columns)
     }
 }
 
@@ -287,7 +294,7 @@ impl AsyncSystemTable for StatisticsTable {
                 ctx.session_state()?,
             )
             .await?;
-        let rows = self.dump_table_columns(ctx, push_downs, &catalog).await?;
+        let rows = self.dump_tables_columns(ctx, push_downs, &catalog).await?;
 
         let mut names = Vec::with_capacity(rows.len());
         let mut databases = Vec::with_capacity(rows.len());