refactor: fuse table funcs (#16149)

* refact: rename generic type names

* wip

* refact: fuse_block

* refact: fuse_column

* wip

* wip

* introduce common arg func

* refactor: fuse_block/columns

* refactor: fuse_segment

* refactor: fuse_snapshot

* resolve merge conflict

* refact: simple arg func

* refactor: cluster infor funcs

* refactor: fuse_encoding/stats

* refactor: squash mods

* cargo clippy

* cleanup

* fix logic test 09_0014_func_clustering_information_function.test

* fix: logic test 09_0006_func_fuse_history.test

Change error code from 1001 (internal error) to 4000( storage error),
if fuse_snapshot is applied to non-fuse tables

Author: dantengsky

src/query/service/src/interpreters/interpreter_select.rs

@@ -145,14 +145,13 @@ impl SelectInterpreter {
             .main_pipeline
             .set_on_finished(move |info: &ExecutionInfo| match &info.res {
                 Ok(_) => GlobalIORuntime::instance().block_on(async move {
-                    info!("Updating the stream meta to consume data");
-
                     match update_stream_metas {
                         Some(streams) => {
                             let r = UpdateMultiTableMetaReq {
                                 update_table_metas: streams.update_table_metas,
                                 ..Default::default()
                             };
+                            info!("Updating the stream meta to consume data");
                             catalog.update_multi_table_meta(r).await.map(|_| ())
                         }
                         None => Ok(()),

src/query/service/src/table_functions/table_function_factory.rs

@@ -19,11 +19,12 @@ use databend_common_catalog::table_args::TableArgs;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_types::MetaId;
-use databend_common_storages_fuse::table_functions::ClusteringStatisticsTable;
+use databend_common_storages_fuse::table_functions::ClusteringStatisticsFunc;
 use databend_common_storages_fuse::table_functions::FuseAmendTable;
-use databend_common_storages_fuse::table_functions::FuseColumnTable;
-use databend_common_storages_fuse::table_functions::FuseEncodingTable;
-use databend_common_storages_fuse::table_functions::SetCacheCapacity;
+use databend_common_storages_fuse::table_functions::FuseBlockFunc;
+use databend_common_storages_fuse::table_functions::FuseColumnFunc;
+use databend_common_storages_fuse::table_functions::FuseEncodingFunc;
+use databend_common_storages_fuse::table_functions::FuseStatisticsFunc;
 use databend_common_storages_fuse::table_functions::TableFunctionTemplate;
 use databend_common_storages_stream::stream_status_table_func::StreamStatusTable;
 use itertools::Itertools;
@@ -35,11 +36,10 @@ use super::SuggestedBackgroundTasksTable;
 use super::TenantQuotaTable;
 use crate::catalogs::SYS_TBL_FUC_ID_END;
 use crate::catalogs::SYS_TBL_FUNC_ID_BEGIN;
-use crate::storages::fuse::table_functions::ClusteringInformationTable;
-use crate::storages::fuse::table_functions::FuseBlockTable;
-use crate::storages::fuse::table_functions::FuseSegmentTable;
-use crate::storages::fuse::table_functions::FuseSnapshotTable;
-use crate::storages::fuse::table_functions::FuseStatisticTable;
+use crate::storages::fuse::table_functions::ClusteringInformationFunc;
+use crate::storages::fuse::table_functions::FuseSegmentFunc;
+use crate::storages::fuse::table_functions::FuseSnapshotFunc;
+use crate::storages::fuse::table_functions::SetCacheCapacityFunc;
 use crate::table_functions::async_crash_me::AsyncCrashMeTable;
 use crate::table_functions::cloud::TaskDependentsEnableTable;
 use crate::table_functions::cloud::TaskDependentsTable;
@@ -121,7 +121,10 @@ impl TableFunctionFactory {
 
         creators.insert(
             "fuse_snapshot".to_string(),
-            (next_id(), Arc::new(FuseSnapshotTable::create)),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<FuseSnapshotFunc>::create),
+            ),
         );
 
         creators.insert(
@@ -136,34 +139,56 @@ impl TableFunctionFactory {
             "set_cache_capacity".to_string(),
             (
                 next_id(),
-                Arc::new(TableFunctionTemplate::<SetCacheCapacity>::create),
+                Arc::new(TableFunctionTemplate::<SetCacheCapacityFunc>::create),
             ),
         );
 
         creators.insert(
             "fuse_segment".to_string(),
-            (next_id(), Arc::new(FuseSegmentTable::create)),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<FuseSegmentFunc>::create),
+            ),
         );
+
         creators.insert(
             "fuse_block".to_string(),
-            (next_id(), Arc::new(FuseBlockTable::create)),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<FuseBlockFunc>::create),
+            ),
         );
+
         creators.insert(
             "fuse_column".to_string(),
-            (next_id(), Arc::new(FuseColumnTable::create)),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<FuseColumnFunc>::create),
+            ),
         );
+
         creators.insert(
             "fuse_statistic".to_string(),
-            (next_id(), Arc::new(FuseStatisticTable::create)),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<FuseStatisticsFunc>::create),
+            ),
         );
 
         creators.insert(
             "clustering_information".to_string(),
-            (next_id(), Arc::new(ClusteringInformationTable::create)),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<ClusteringInformationFunc>::create),
+            ),
         );
+
         creators.insert(
             "clustering_statistics".to_string(),
-            (next_id(), Arc::new(ClusteringStatisticsTable::create)),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<ClusteringStatisticsFunc>::create),
+            ),
         );
 
         creators.insert(
@@ -232,7 +257,10 @@ impl TableFunctionFactory {
 
         creators.insert(
             "fuse_encoding".to_string(),
-            (next_id(), Arc::new(FuseEncodingTable::create)),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<FuseEncodingFunc>::create),
+            ),
         );
 
         creators.insert(

src/query/storages/fuse/src/table_functions/cache_admin/mod.rs

@@ -19,6 +19,9 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use chrono::Utc;
+use databend_common_catalog::catalog::CATALOG_DEFAULT;
+use databend_common_catalog::plan::DataSourcePlan;
+use databend_common_catalog::table_args::TableArgs;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::compare_scalars;
@@ -42,6 +45,7 @@ use databend_common_expression::Scalar;
 use databend_common_expression::TableDataType;
 use databend_common_expression::TableField;
 use databend_common_expression::TableSchema;
+use databend_common_expression::TableSchemaRef;
 use databend_common_expression::TableSchemaRefExt;
 use databend_common_expression::Value;
 use databend_common_functions::BUILTIN_FUNCTIONS;
@@ -57,9 +61,81 @@ use serde_json::Value as JsonValue;
 use crate::io::SegmentsIO;
 use crate::sessions::TableContext;
 use crate::table_functions::cmp_with_null;
+use crate::table_functions::parse_db_tb_opt_args;
+use crate::table_functions::string_literal;
+use crate::table_functions::SimpleArgFunc;
+use crate::table_functions::SimpleArgFuncTemplate;
 use crate::FuseTable;
 use crate::Table;
 
+pub struct ClusteringInformationArgs {
+    database_name: String,
+    table_name: String,
+    cluster_key: Option<String>,
+}
+
+impl From<&ClusteringInformationArgs> for TableArgs {
+    fn from(args: &ClusteringInformationArgs) -> Self {
+        let mut tbl_args = Vec::new();
+        tbl_args.push(string_literal(args.database_name.as_str()));
+        tbl_args.push(string_literal(args.table_name.as_str()));
+        if let Some(arg_cluster_key) = &args.cluster_key {
+            tbl_args.push(string_literal(arg_cluster_key));
+        }
+        TableArgs::new_positioned(tbl_args)
+    }
+}
+impl TryFrom<(&str, TableArgs)> for ClusteringInformationArgs {
+    type Error = ErrorCode;
+    fn try_from(
+        (func_name, table_args): (&str, TableArgs),
+    ) -> std::result::Result<Self, Self::Error> {
+        let (database_name, table_name, cluster_key) =
+            parse_db_tb_opt_args(&table_args, func_name)?;
+
+        Ok(Self {
+            database_name,
+            table_name,
+            cluster_key,
+        })
+    }
+}
+
+pub type ClusteringInformationFunc = SimpleArgFuncTemplate<ClusteringInformationNew>;
+pub struct ClusteringInformationNew;
+
+#[async_trait::async_trait]
+impl SimpleArgFunc for ClusteringInformationNew {
+    type Args = ClusteringInformationArgs;
+
+    fn schema() -> TableSchemaRef {
+        ClusteringInformation::schema()
+    }
+
+    async fn apply(
+        ctx: &Arc<dyn TableContext>,
+        args: &Self::Args,
+        _plan: &DataSourcePlan,
+    ) -> Result<DataBlock> {
+        let tenant_id = ctx.get_tenant();
+        let tbl = ctx
+            .get_catalog(CATALOG_DEFAULT)
+            .await?
+            .get_table(
+                &tenant_id,
+                args.database_name.as_str(),
+                args.table_name.as_str(),
+            )
+            .await?;
+
+        let tbl = FuseTable::try_from_table(tbl.as_ref())?;
+
+        ClusteringInformation::new(ctx.clone(), tbl, args.cluster_key.clone())
+            .get_clustering_info()
+            .await
+    }
+}
+
 pub struct ClusteringInformation<'a> {
     pub ctx: Arc<dyn TableContext>,
     pub table: &'a FuseTable,