add block serializer

Author: zhyass

Cargo.lock

@@ -76,8 +76,8 @@ impl<W: Write> NativeWriter<W> {
     }
 
     /// Consumes itself into the inner writer
-    pub fn into_inner(self) -> W {
-        self.writer.w
+    pub fn inner_mut(&mut self) -> &mut W {
+        &mut self.writer.w
     }
 
     /// Writes the header and first (schema) message to the file.

src/common/native/src/write/writer.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::Ordering;
 use std::collections::HashMap;
 
 use databend_common_exception::ErrorCode;
@@ -132,15 +131,6 @@ pub fn u64_literal(val: u64) -> Scalar {
     Scalar::Number(NumberScalar::UInt64(val))
 }
 
-pub fn cmp_with_null(v1: &Scalar, v2: &Scalar) -> Ordering {
-    match (v1.is_null(), v2.is_null()) {
-        (true, true) => Ordering::Equal,
-        (true, false) => Ordering::Greater,
-        (false, true) => Ordering::Less,
-        (false, false) => v1.cmp(v2),
-    }
-}
-
 pub fn parse_sequence_args(table_args: &TableArgs, func_name: &str) -> Result<String> {
     let args = table_args.expect_all_positioned(func_name, Some(1))?;
     let sequence = string_value(&args[0])?;

src/query/catalog/src/table_args.rs

@@ -21,6 +21,7 @@ use databend_common_io::constants::DEFAULT_BLOCK_BUFFER_SIZE;
 use databend_common_io::constants::DEFAULT_BLOCK_INDEX_BUFFER_SIZE;
 use databend_common_sql::BloomIndexColumns;
 use databend_common_storages_fuse::io::serialize_block;
+use databend_common_storages_fuse::io::BloomIndexBuilder;
 use databend_common_storages_fuse::io::TableMetaLocationGenerator;
 use databend_common_storages_fuse::io::WriteSettings;
 use databend_common_storages_fuse::FuseStorageFormat;
@@ -127,12 +128,9 @@ impl<'a> BlockWriter<'a> {
         let bloom_index_cols = BloomIndexColumns::All;
         let bloom_columns_map =
             bloom_index_cols.bloom_index_fields(schema.clone(), BloomIndex::supported_type)?;
-        let maybe_bloom_index = BloomIndex::try_create(
-            FunctionContext::default(),
-            location.1,
-            block,
-            bloom_columns_map,
-        )?;
+        let mut builder = BloomIndexBuilder::create(FunctionContext::default(), bloom_columns_map);
+        builder.add_block(block)?;
+        let maybe_bloom_index = builder.finalize()?;
         if let Some(bloom_index) = maybe_bloom_index {
             let index_block = bloom_index.serialize_to_data_block()?;
             let filter_schema = bloom_index.filter_schema;

src/query/service/src/test_kits/block_writer.rs

@@ -1,4 +1,4 @@
-// Copyright 2022 Datafuse Labs.
+// 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.
@@ -17,14 +17,23 @@ use std::collections::HashMap;
 use std::io::Write;
 use std::sync::Arc;
 
+use databend_common_column::buffer::Buffer;
 use databend_common_expression::type_check;
 use databend_common_expression::type_check::check_function;
-use databend_common_expression::types::array::ArrayColumn;
 use databend_common_expression::types::map::KvColumn;
 use databend_common_expression::types::map::KvPair;
-use databend_common_expression::types::number::NumberScalar;
-use databend_common_expression::types::number::UInt8Type;
-use databend_common_expression::types::*;
+use databend_common_expression::types::AnyType;
+use databend_common_expression::types::ArrayColumn;
+use databend_common_expression::types::DataType;
+use databend_common_expression::types::DateType;
+use databend_common_expression::types::Int16Type;
+use databend_common_expression::types::Int32Type;
+use databend_common_expression::types::Int8Type;
+use databend_common_expression::types::NumberDataType;
+use databend_common_expression::types::NumberScalar;
+use databend_common_expression::types::StringType;
+use databend_common_expression::types::UInt8Type;
+use databend_common_expression::types::VariantType;
 use databend_common_expression::BlockEntry;
 use databend_common_expression::Column;
 use databend_common_expression::ColumnRef;
@@ -42,18 +51,17 @@ use databend_common_expression::TableSchema;
 use databend_common_expression::Value;
 use databend_common_functions::test_utils::parse_raw_expr;
 use databend_common_functions::BUILTIN_FUNCTIONS;
-use databend_storages_common_index::filters::BlockFilter as LatestBloom;
+use databend_common_storages_fuse::io::BloomIndexBuilder;
 use databend_storages_common_index::filters::Xor8Filter;
 use databend_storages_common_index::BloomIndex;
 use databend_storages_common_index::FilterEvalResult;
 use databend_storages_common_index::Index;
 use databend_storages_common_table_meta::meta::ColumnStatistics;
-use databend_storages_common_table_meta::meta::Versioned;
 use goldenfile::Mint;
 
 #[test]
 fn test_bloom_filter() {
-    let mut mint = Mint::new("tests/it/testdata");
+    let mut mint = Mint::new("tests/it/storages/testdata");
     let file = &mut mint.new_goldenfile("test_bloom_filter.txt").unwrap();
 
     test_base(file);
@@ -478,14 +486,10 @@ fn eval_index_expr(
             BloomIndex::calculate_scalar_digest(&func_ctx, scalar, &ty).unwrap()
         });
     }
-    let index = BloomIndex::try_create(
-        func_ctx.clone(),
-        LatestBloom::VERSION,
-        block,
-        bloom_columns.clone(),
-    )
-    .unwrap()
-    .unwrap();
+
+    let mut builder = BloomIndexBuilder::create(func_ctx.clone(), bloom_columns.clone());
+    builder.add_block(block).unwrap();
+    let index = builder.finalize().unwrap().unwrap();
 
     let column_stats = block
         .columns()

src/query/storages/common/index/tests/it/bloom_pruner.rs renamed to src/query/service/tests/it/storages/fuse/bloom_pruner.rs

@@ -14,6 +14,7 @@
 
 #![allow(clippy::too_many_arguments)]
 mod bloom_index_meta_size;
+mod bloom_pruner;
 mod conflict;
 mod io;
 mod meta;

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

@@ -84,8 +84,8 @@ fn test_ft_stats_block_stats() -> databend_common_exception::Result<()> {
 #[test]
 fn test_ft_stats_block_stats_with_column_distinct_count() -> databend_common_exception::Result<()> {
     let schema = Arc::new(TableSchema::new(vec![
-        TableField::new("a", TableDataType::Number(NumberDataType::Int32)),
-        TableField::new("b", TableDataType::String),
+        TableField::new_from_column_id("a", TableDataType::Number(NumberDataType::Int32), 0),
+        TableField::new_from_column_id("b", TableDataType::String, 1),
     ]));
 
     let block = DataBlock::new_from_columns(vec![