fix review comment

Author: zhyass

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

@@ -7,7 +7,7 @@ Returns the block information of a snapshot of table.
 ## Syntax
 
 ```sql
-FUSE_BLOCK('<database_name>', '<table_name>','<snapshot_id>')
+FUSE_BLOCK('<database_name>', '<table_name>'[, '<snapshot_id>'])
 ```
 
 ## Examples
@@ -17,17 +17,7 @@ CREATE TABLE mytable(c int);
 INSERT INTO mytable values(1);
 INSERT INTO mytable values(2); 
 
--- Obtain a snapshot ID
-SELECT snapshot_id FROM FUSE_SNAPSHOT('default', 'mytable') limit 1;
-
----
-+----------------------------------+
-| snapshot_id                      |
-+----------------------------------+
-| 51e84b56458f44269b05a059b364a659 |
-+----------------------------------+
-
-SELECT * FROM FUSE_BLOCK('default', 'mytable', '51e84b56458f44269b05a059b364a659');
+SELECT * FROM FUSE_BLOCK('default', 'mytable');
 
 ---
 +----------------------------------+----------------------------+----------------------------------------------------+------------+----------------------------------------------------+-------------------+

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

@@ -41,13 +41,17 @@ impl OneBlockProcedure for FuseBlockProcedure {
     }
 
     fn features(&self) -> ProcedureFeatures {
-        ProcedureFeatures::default().num_arguments(3)
+        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 = args[2].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())?

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

@@ -123,13 +123,13 @@ async fn test_call_fuse_block_interpreter() -> Result<()> {
         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 3 arguments, but got 0.";
+        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, xxxx)";
+        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");
@@ -143,7 +143,7 @@ async fn test_call_fuse_block_interpreter() -> Result<()> {
 
     // BadArguments
     {
-        let query = "call system$fuse_block(system, tables, xxxx)";
+        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");
@@ -167,7 +167,7 @@ async fn test_call_fuse_block_interpreter() -> Result<()> {
 
     // fuse_block
     {
-        let query = "call system$fuse_block(default, a, xxxx)";
+        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?;

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;

src/query/storages/fuse/src/table_functions/fuse_blocks/fuse_block.rs

@@ -17,6 +17,7 @@ use std::sync::Arc;
 use common_datablocks::DataBlock;
 use common_datavalues::prelude::*;
 use common_exception::Result;
+use common_fuse_meta::meta::TableSnapshot;
 use futures_util::TryStreamExt;
 
 use crate::io::MetaReaders;
@@ -27,11 +28,15 @@ use crate::FuseTable;
 pub struct FuseBlock<'a> {
     pub ctx: Arc<dyn TableContext>,
     pub table: &'a FuseTable,
-    pub snapshot_id: String,
+    pub snapshot_id: Option<String>,
 }
 
 impl<'a> FuseBlock<'a> {
-    pub fn new(ctx: Arc<dyn TableContext>, table: &'a FuseTable, snapshot_id: String) -> Self {
+    pub fn new(
+        ctx: Arc<dyn TableContext>,
+        table: &'a FuseTable,
+        snapshot_id: Option<String>,
+    ) -> Self {
         Self {
             ctx,
             table,
@@ -43,6 +48,10 @@ impl<'a> FuseBlock<'a> {
         let tbl = self.table;
         let maybe_snapshot = tbl.read_table_snapshot(self.ctx.clone()).await?;
         if let Some(snapshot) = maybe_snapshot {
+            if self.snapshot_id.is_none() {
+                return self.to_block(snapshot).await;
+            }
+
             // prepare the stream of snapshot
             let snapshot_version = tbl.snapshot_format_version();
             let snapshot_location = tbl
@@ -57,46 +66,49 @@ impl<'a> FuseBlock<'a> {
 
             // find the element by snapshot_id in stream
             while let Some(snapshot) = snapshot_stream.try_next().await? {
-                if snapshot.snapshot_id.simple().to_string() == self.snapshot_id {
-                    let len = snapshot.summary.block_count as usize;
-                    let snapshot_id = vec![self.snapshot_id.clone().into_bytes()];
-                    let timestamp =
-                        vec![snapshot.timestamp.map(|dt| (dt.timestamp_micros()) as i64)];
-                    let mut block_location: Vec<Vec<u8>> = Vec::with_capacity(len);
-                    let mut block_size: Vec<u64> = Vec::with_capacity(len);
-                    let mut bloom_filter_location: Vec<Option<Vec<u8>>> = Vec::with_capacity(len);
-                    let mut bloom_filter_size: Vec<u64> = Vec::with_capacity(len);
-
-                    let reader = MetaReaders::segment_info_reader(self.ctx.as_ref());
-                    for (x, ver) in &snapshot.segments {
-                        let segment = reader.read(x, None, *ver).await?;
-                        segment.blocks.clone().into_iter().for_each(|block| {
-                            block_location.push(block.location.0.into_bytes());
-                            block_size.push(block.block_size);
-                            bloom_filter_location.push(
-                                block
-                                    .bloom_filter_index_location
-                                    .map(|(s, _)| s.into_bytes()),
-                            );
-                            bloom_filter_size.push(block.bloom_filter_index_size);
-                        });
-                    }
-
-                    return Ok(DataBlock::create(FuseBlock::schema(), vec![
-                        Arc::new(ConstColumn::new(Series::from_data(snapshot_id), len)),
-                        Arc::new(ConstColumn::new(Series::from_data(timestamp), len)),
-                        Series::from_data(block_location),
-                        Series::from_data(block_size),
-                        Series::from_data(bloom_filter_location),
-                        Series::from_data(bloom_filter_size),
-                    ]));
+                if snapshot.snapshot_id.simple().to_string() == self.snapshot_id.clone().unwrap() {
+                    return self.to_block(snapshot).await;
                 }
             }
         }
 
         Ok(DataBlock::empty_with_schema(Self::schema()))
     }
 
+    async fn to_block(&self, snapshot: Arc<TableSnapshot>) -> Result<DataBlock> {
+        let len = snapshot.summary.block_count as usize;
+        let snapshot_id = vec![snapshot.snapshot_id.simple().to_string().into_bytes()];
+        let timestamp = vec![snapshot.timestamp.map(|dt| (dt.timestamp_micros()) as i64)];
+        let mut block_location: Vec<Vec<u8>> = Vec::with_capacity(len);
+        let mut block_size: Vec<u64> = Vec::with_capacity(len);
+        let mut bloom_filter_location: Vec<Option<Vec<u8>>> = Vec::with_capacity(len);
+        let mut bloom_filter_size: Vec<u64> = Vec::with_capacity(len);
+
+        let reader = MetaReaders::segment_info_reader(self.ctx.as_ref());
+        for (x, ver) in &snapshot.segments {
+            let segment = reader.read(x, None, *ver).await?;
+            segment.blocks.clone().into_iter().for_each(|block| {
+                block_location.push(block.location.0.into_bytes());
+                block_size.push(block.block_size);
+                bloom_filter_location.push(
+                    block
+                        .bloom_filter_index_location
+                        .map(|(s, _)| s.into_bytes()),
+                );
+                bloom_filter_size.push(block.bloom_filter_index_size);
+            });
+        }
+
+        Ok(DataBlock::create(FuseBlock::schema(), vec![
+            Arc::new(ConstColumn::new(Series::from_data(snapshot_id), len)),
+            Arc::new(ConstColumn::new(Series::from_data(timestamp), len)),
+            Series::from_data(block_location),
+            Series::from_data(block_size),
+            Series::from_data(bloom_filter_location),
+            Series::from_data(bloom_filter_size),
+        ]))
+    }
+
     pub fn schema() -> Arc<DataSchema> {
         DataSchemaRefExt::create(vec![
             DataField::new("snapshot_id", Vu8::to_data_type()),