Merge pull request #7631 from zhyass/feature_fix

feat(storage): get all parquet file list for fuse engine

Author: mergify[bot]

docs/doc/30-reference/20-functions/111-system-functions/fuse_block.md

@@ -0,0 +1,29 @@
+---
+title: FUSE_BLOCK
+---
+
+Returns the block information of a snapshot of table.
+
+## Syntax
+
+```sql
+FUSE_BLOCK('<database_name>', '<table_name>'[, '<snapshot_id>'])
+```
+
+## Examples
+
+```sql
+CREATE TABLE mytable(c int);
+INSERT INTO mytable values(1);
+INSERT INTO mytable values(2); 
+
+SELECT * FROM FUSE_BLOCK('default', 'mytable');
+
+---
++----------------------------------+----------------------------+----------------------------------------------------+------------+----------------------------------------------------+-------------------+
+| snapshot_id                      | timestamp                  | block_location                                     | block_size | bloom_filter_location                              | bloom_filter_size |
++----------------------------------+----------------------------+----------------------------------------------------+------------+----------------------------------------------------+-------------------+
+| 51e84b56458f44269b05a059b364a659 | 2022-09-15 07:14:14.137268 | 1/7/_b/39a6dbbfd9b44ad5a8ec8ab264c93cf5_v0.parquet |          4 | 1/7/_i/39a6dbbfd9b44ad5a8ec8ab264c93cf5_v1.parquet |               221 |
+| 51e84b56458f44269b05a059b364a659 | 2022-09-15 07:14:14.137268 | 1/7/_b/d0ee9688c4d24d6da86acd8b0d6f4fad_v0.parquet |          4 | 1/7/_i/d0ee9688c4d24d6da86acd8b0d6f4fad_v1.parquet |               219 |
++----------------------------------+----------------------------+----------------------------------------------------+------------+----------------------------------------------------+-------------------+
+```

src/query/service/src/procedures/systems/fuse_block.rs

@@ -0,0 +1,73 @@
+//  Copyright 2022 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 common_datablocks::DataBlock;
+use common_datavalues::DataSchema;
+use common_exception::Result;
+
+use crate::procedures::OneBlockProcedure;
+use crate::procedures::Procedure;
+use crate::procedures::ProcedureFeatures;
+use crate::sessions::QueryContext;
+use crate::sessions::TableContext;
+use crate::storages::fuse::table_functions::FuseBlock;
+use crate::storages::fuse::FuseTable;
+
+pub struct FuseBlockProcedure {}
+
+impl FuseBlockProcedure {
+    pub fn try_create() -> Result<Box<dyn Procedure>> {
+        Ok(FuseBlockProcedure {}.into_procedure())
+    }
+}
+
+#[async_trait::async_trait]
+impl OneBlockProcedure for FuseBlockProcedure {
+    fn name(&self) -> &str {
+        "FUSE_BLOCK"
+    }
+
+    fn features(&self) -> ProcedureFeatures {
+        ProcedureFeatures::default().variadic_arguments(2, 3)
+    }
+
+    async fn all_data(&self, ctx: Arc<QueryContext>, args: Vec<String>) -> Result<DataBlock> {
+        let database_name = args[0].clone();
+        let table_name = args[1].clone();
+        let snapshot_id = if args.len() > 2 {
+            Some(args[2].clone())
+        } else {
+            None
+        };
+        let tenant_id = ctx.get_tenant();
+        let tbl = ctx
+            .get_catalog(&ctx.get_current_catalog())?
+            .get_table(
+                tenant_id.as_str(),
+                database_name.as_str(),
+                table_name.as_str(),
+            )
+            .await?;
+
+        let tbl = FuseTable::try_from_table(tbl.as_ref())?;
+
+        Ok(FuseBlock::new(ctx, tbl, snapshot_id).get_blocks().await?)
+    }
+
+    fn schema(&self) -> Arc<DataSchema> {
+        FuseBlock::schema()
+    }
+}

src/query/service/src/procedures/systems/mod.rs

@@ -13,12 +13,14 @@
 // limitations under the License.
 
 mod clustering_information;
+mod fuse_block;
 mod fuse_segment;
 mod fuse_snapshot;
 mod search_tables;
 mod system;
 
 pub use clustering_information::ClusteringInformationProcedure;
+pub use fuse_block::FuseBlockProcedure;
 pub use fuse_segment::FuseSegmentProcedure;
 pub use fuse_snapshot::FuseSnapshotProcedure;
 pub use search_tables::SearchTablesProcedure;

src/query/service/src/procedures/systems/system.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use crate::procedures::systems::ClusteringInformationProcedure;
+use crate::procedures::systems::FuseBlockProcedure;
 use crate::procedures::systems::FuseSegmentProcedure;
 use crate::procedures::systems::FuseSnapshotProcedure;
 use crate::procedures::systems::SearchTablesProcedure;
@@ -34,6 +35,10 @@ impl SystemProcedure {
             "system$fuse_segment",
             Box::new(FuseSegmentProcedure::try_create),
         );
+        factory.register(
+            "system$fuse_block",
+            Box::new(FuseBlockProcedure::try_create),
+        );
         factory.register(
             "system$search_tables",
             Box::new(SearchTablesProcedure::try_create),

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

@@ -24,6 +24,7 @@ use parking_lot::RwLock;
 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::table_functions::async_crash_me::AsyncCrashMeTable;
@@ -105,6 +106,10 @@ impl TableFunctionFactory {
             "fuse_segment".to_string(),
             (next_id(), Arc::new(FuseSegmentTable::create)),
         );
+        creators.insert(
+            "fuse_block".to_string(),
+            (next_id(), Arc::new(FuseBlockTable::create)),
+        );
 
         creators.insert(
             "clustering_information".to_string(),

src/query/service/tests/it/interpreters/interpreter_call.rs

@@ -110,6 +110,72 @@ async fn test_call_fuse_snapshot_interpreter() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn test_call_fuse_block_interpreter() -> Result<()> {
+    let (_guard, ctx) = crate::tests::create_query_context().await?;
+    let mut planner = Planner::new(ctx.clone());
+
+    // NumberArgumentsNotMatch
+    {
+        let query = "call system$fuse_block()";
+        let (plan, _, _) = planner.plan_sql(query).await?;
+        let executor = InterpreterFactoryV2::get(ctx.clone(), &plan)?;
+        assert_eq!(executor.name(), "CallInterpreter");
+        let res = executor.execute().await;
+        assert_eq!(res.is_err(), true);
+        let expect = "Code: 1028, displayText = Function `FUSE_BLOCK` expect to have [2, 3] arguments, but got 0.";
+        assert_eq!(expect, res.err().unwrap().to_string());
+    }
+
+    // UnknownTable
+    {
+        let query = "call system$fuse_block(default, test)";
+        let (plan, _, _) = planner.plan_sql(query).await?;
+        let executor = InterpreterFactoryV2::get(ctx.clone(), &plan)?;
+        assert_eq!(executor.name(), "CallInterpreter");
+        let res = executor.execute().await;
+        assert_eq!(res.is_err(), true);
+        assert_eq!(
+            res.err().unwrap().code(),
+            ErrorCode::UnknownTable("").code()
+        );
+    }
+
+    // BadArguments
+    {
+        let query = "call system$fuse_block(system, tables)";
+        let (plan, _, _) = planner.plan_sql(query).await?;
+        let executor = InterpreterFactoryV2::get(ctx.clone(), &plan)?;
+        assert_eq!(executor.name(), "CallInterpreter");
+        let res = executor.execute().await;
+        assert_eq!(res.is_err(), true);
+        let expect =
+            "Code: 1015, displayText = expects table of engine FUSE, but got SystemTables.";
+        assert_eq!(expect, res.err().unwrap().to_string());
+    }
+
+    // Create table
+    {
+        let query = "\
+            CREATE TABLE default.a(a bigint)\
+        ";
+
+        let (plan, _, _) = planner.plan_sql(query).await?;
+        let executor = InterpreterFactoryV2::get(ctx.clone(), &plan)?;
+        let _ = executor.execute().await?;
+    }
+
+    // fuse_block
+    {
+        let query = "call system$fuse_block(default, a)";
+        let (plan, _, _) = planner.plan_sql(query).await?;
+        let executor = InterpreterFactoryV2::get(ctx.clone(), &plan)?;
+        let _ = executor.execute().await?;
+    }
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_call_clustering_information_interpreter() -> Result<()> {
     let (_guard, ctx) = crate::tests::create_query_context().await?;
@@ -188,7 +254,7 @@ async fn test_call_clustering_information_interpreter() -> Result<()> {
         let _ = executor.execute().await?;
     }
 
-    // FuseHistory
+    // clustering_information
     {
         let query = "call system$clustering_information(default, b)";
         let (plan, _, _) = planner.plan_sql(query).await?;

src/query/service/tests/it/storages/fuse/table_functions/clustering_information_table.rs

@@ -22,7 +22,6 @@ use databend_query::interpreters::CreateTableInterpreterV2;
 use databend_query::interpreters::Interpreter;
 use tokio_stream::StreamExt;
 
-use crate::storages::fuse::table_test_fixture::TestFixture;
 use crate::storages::fuse::table_test_fixture::*;
 
 #[tokio::test]

src/query/service/tests/it/storages/fuse/table_functions/fuse_block_table.rs

@@ -0,0 +1,99 @@
+//  Copyright 2022 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 common_base::base::tokio;
+use common_datablocks::DataBlock;
+use common_exception::ErrorCode;
+use common_exception::Result;
+use databend_query::interpreters::CreateTableInterpreterV2;
+use databend_query::interpreters::Interpreter;
+use tokio_stream::StreamExt;
+
+use crate::storages::fuse::table_test_fixture::*;
+
+#[tokio::test]
+async fn test_fuse_block_table() -> Result<()> {
+    let fixture = TestFixture::new().await;
+    let db = fixture.default_db_name();
+    let tbl = fixture.default_table_name();
+    let ctx = fixture.ctx();
+
+    // test db & table
+    let create_table_plan = fixture.default_crate_table_plan();
+    let interpreter = CreateTableInterpreterV2::try_create(ctx.clone(), create_table_plan)?;
+    interpreter.execute().await?;
+
+    {
+        let expected = vec![
+            "+-------+",
+            "| count |",
+            "+-------+",
+            "| 0     |",
+            "+-------+",
+        ];
+        let qry = format!(
+            "select count(1) as count from fuse_block('{}', '{}')",
+            db, tbl
+        );
+
+        expects_ok(
+            "count_should_be_0",
+            execute_query(ctx.clone(), qry.as_str()).await,
+            expected,
+        )
+        .await?;
+    }
+
+    {
+        let qry = format!("insert into {}.{} values(1, (2, 3)),(2, (4, 6))", db, tbl);
+        execute_query(ctx.clone(), qry.as_str()).await?;
+        let qry = format!("insert into {}.{} values(7, (8, 9))", db, tbl);
+        execute_query(ctx.clone(), qry.as_str()).await?;
+        let expected = vec![
+            "+-------+",
+            "| count |",
+            "+-------+",
+            "| 2     |",
+            "+-------+",
+        ];
+
+        let qry = format!(
+            "select count(1) as count from fuse_block('{}', '{}')",
+            db, tbl
+        );
+
+        expects_ok(
+            "count_should_be_2",
+            execute_query(ctx.clone(), qry.as_str()).await,
+            expected,
+        )
+        .await?;
+    }
+
+    {
+        // incompatible table engine
+        let qry = format!("create table {}.in_mem (a int) engine =Memory", db);
+        execute_query(ctx.clone(), qry.as_str()).await?;
+
+        let qry = format!("select * from fuse_block('{}', '{}')", db, "in_mem");
+        let output_stream = execute_query(ctx.clone(), qry.as_str()).await?;
+        expects_err(
+            "unsupported_table_engine",
+            ErrorCode::logical_error_code(),
+            output_stream.collect::<Result<Vec<DataBlock>>>().await,
+        );
+    }
+
+    Ok(())
+}

src/query/service/tests/it/storages/fuse/table_functions/fuse_snapshot_table.rs

@@ -22,7 +22,6 @@ use databend_query::interpreters::CreateTableInterpreterV2;
 use databend_query::interpreters::Interpreter;
 use tokio_stream::StreamExt;
 
-use crate::storages::fuse::table_test_fixture::TestFixture;
 use crate::storages::fuse::table_test_fixture::*;
 
 #[tokio::test]
@@ -182,7 +181,7 @@ async fn test_fuse_snapshot_table_read() -> Result<()> {
         let qry = format!("select * from fuse_snapshot('{}', '{}')", db, "in_mem");
         let output_stream = execute_query(ctx.clone(), qry.as_str()).await?;
         expects_err(
-            "check_row_and_block_count_after_append",
+            "unsupported_table_engine",
             ErrorCode::logical_error_code(),
             output_stream.collect::<Result<Vec<DataBlock>>>().await,
         );

src/query/service/tests/it/storages/fuse/table_functions/mod.rs

@@ -13,4 +13,5 @@
 //  limitations under the License.
 
 mod clustering_information_table;
+mod fuse_block_table;
 mod fuse_snapshot_table;