feat(storage): add fuse_tag table function to list snapshot tags (#19664)

* feat: support desc table tags

* fix

* fix

Author: zhyass

Cargo.lock

@@ -123,7 +123,6 @@ futures = { workspace = true }
 futures-util = { workspace = true }
 geo = { workspace = true }
 geo-index = { workspace = true }
-geozero = { workspace = true }
 headers = { workspace = true }
 hex = { workspace = true }
 http = { workspace = true }

src/query/service/Cargo.toml

@@ -48,6 +48,7 @@ use super::others::UdfEchoTable;
 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::FuseTagFunc;
 use crate::table_functions::TableFunction;
 use crate::table_functions::async_crash_me::AsyncCrashMeTable;
 use crate::table_functions::cloud::TaskDependentsEnableTable;
@@ -141,6 +142,14 @@ impl TableFunctionFactory {
             ),
         );
 
+        creators.insert(
+            "fuse_tag".to_string(),
+            (
+                next_id(),
+                Arc::new(TableFunctionTemplate::<FuseTagFunc>::create),
+            ),
+        );
+
         creators.insert(
             "fuse_dump_snapshots".to_string(),
             (

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

@@ -0,0 +1,130 @@
+// 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::plan::DataSourcePlan;
+use databend_common_catalog::table_args::TableArgs;
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_expression::DataBlock;
+use databend_common_expression::FromData;
+use databend_common_expression::TableDataType;
+use databend_common_expression::TableField;
+use databend_common_expression::TableSchema;
+use databend_common_expression::TableSchemaRefExt;
+use databend_common_expression::types::StringType;
+use databend_common_expression::types::TimestampType;
+use databend_common_meta_app::schema::ListTableTagsReq;
+
+use crate::operations::check_table_ref_access;
+use crate::sessions::TableContext;
+use crate::table_functions::SimpleTableFunc;
+use crate::table_functions::parse_db_tb_args;
+use crate::table_functions::string_literal;
+
+struct FuseTagArgs {
+    database_name: String,
+    table_name: String,
+}
+
+impl From<&FuseTagArgs> for TableArgs {
+    fn from(args: &FuseTagArgs) -> Self {
+        TableArgs::new_positioned(vec![
+            string_literal(args.database_name.as_str()),
+            string_literal(args.table_name.as_str()),
+        ])
+    }
+}
+
+pub struct FuseTagFunc {
+    args: FuseTagArgs,
+}
+
+#[async_trait::async_trait]
+impl SimpleTableFunc for FuseTagFunc {
+    fn table_args(&self) -> Option<TableArgs> {
+        Some((&self.args).into())
+    }
+
+    fn schema(&self) -> Arc<TableSchema> {
+        TableSchemaRefExt::create(vec![
+            TableField::new("name", TableDataType::String),
+            TableField::new("snapshot_location", TableDataType::String),
+            TableField::new("expire_at", TableDataType::Timestamp.wrap_nullable()),
+        ])
+    }
+
+    async fn apply(
+        &self,
+        ctx: &Arc<dyn TableContext>,
+        _plan: &DataSourcePlan,
+    ) -> Result<Option<DataBlock>> {
+        check_table_ref_access(ctx.as_ref())?;
+
+        let catalog = ctx.get_current_catalog();
+        let tenant = ctx.get_tenant();
+        let tbl = ctx
+            .get_catalog(&catalog)
+            .await?
+            .get_table(&tenant, &self.args.database_name, &self.args.table_name)
+            .await?;
+        if tbl.engine() != "FUSE" {
+            return Err(ErrorCode::TableEngineNotSupported(
+                "Invalid table engine, only FUSE table supports fuse_tag",
+            ));
+        }
+        let table_id = tbl.get_id();
+
+        let tags = ctx
+            .get_catalog(&catalog)
+            .await?
+            .list_table_tags(ListTableTagsReq {
+                table_id,
+                include_expired: true,
+            })
+            .await?;
+
+        if tags.is_empty() {
+            return Ok(Some(DataBlock::empty_with_schema(&self.schema().into())));
+        }
+
+        let mut names: Vec<String> = Vec::with_capacity(tags.len());
+        let mut snapshot_locations: Vec<String> = Vec::with_capacity(tags.len());
+        let mut expire_ats: Vec<Option<i64>> = Vec::with_capacity(tags.len());
+
+        for (tag_name, seq_tag) in &tags {
+            names.push(tag_name.clone());
+            snapshot_locations.push(seq_tag.data.snapshot_loc.clone());
+            expire_ats.push(seq_tag.data.expire_at.map(|t| t.timestamp_micros()));
+        }
+
+        Ok(Some(DataBlock::new_from_columns(vec![
+            StringType::from_data(names),
+            StringType::from_data(snapshot_locations),
+            TimestampType::from_opt_data(expire_ats),
+        ])))
+    }
+
+    fn create(func_name: &str, table_args: TableArgs) -> Result<Self>
+    where Self: Sized {
+        let (arg_database_name, arg_table_name) = parse_db_tb_args(&table_args, func_name)?;
+        Ok(Self {
+            args: FuseTagArgs {
+                database_name: arg_database_name,
+                table_name: arg_table_name,
+            },
+        })
+    }
+}

src/query/storages/fuse/src/table_functions/fuse_tag.rs

@@ -24,6 +24,7 @@ mod fuse_page;
 mod fuse_segment;
 mod fuse_snapshot;
 mod fuse_statistic;
+mod fuse_tag;
 mod fuse_time_travel_size;
 mod fuse_vacuum_drop_aggregating_index;
 mod fuse_vacuum_drop_inverted_index;
@@ -47,6 +48,7 @@ pub use fuse_page::FusePageFunc;
 pub use fuse_segment::FuseSegmentFunc;
 pub use fuse_snapshot::FuseSnapshotFunc;
 pub use fuse_statistic::FuseStatisticsFunc;
+pub use fuse_tag::FuseTagFunc;
 pub use fuse_time_travel_size::FuseTimeTravelSize;
 pub use fuse_time_travel_size::FuseTimeTravelSizeFunc;
 pub use fuse_vacuum_drop_aggregating_index::FuseVacuumDropAggregatingIndex;

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

@@ -17,6 +17,7 @@ fuse_page
 fuse_segment
 fuse_snapshot
 fuse_statistic
+fuse_tag
 fuse_time_travel_size
 fuse_vacuum2
 fuse_vacuum_drop_aggregating_index

tests/sqllogictests/suites/base/06_show/06_0014_show_table_functions.test

@@ -102,15 +102,25 @@ SELECT SLEEP(2)
 statement error 2748
 SELECT * FROM t1 at (tag => "temp_tag")
 
+query TB
+select name, expire_at is not null from fuse_tag('test_tag','t1') order by name
+----
+temp_tag 1
+v1_0 0
+v1_0_copy 0
+
 statement ok
 optimize table t1 compact
 
 statement ok
 select * from fuse_vacuum2('test_tag', 't1') ignore_result;
 
 ## temp_tag was purged.
-statement error 2745
-SELECT * FROM t1 at (tag => "temp_tag")
+query TB
+select name, expire_at is not null from fuse_tag('test_tag','t1') order by name
+----
+v1_0 0
+v1_0_copy 0
 
 query IT
 SELECT * FROM t1 at (tag => "v1_0") ORDER BY a