feat(storage): use distinct count calculated by the xor hash function

Author: b41sh

src/query/service/tests/it/storages/fuse/operations/mutation/segments_compact_mutator.rs

@@ -663,9 +663,10 @@ impl CompactSegmentTestFixture {
             let mut stats_acc = StatisticsAccumulator::default();
             for block in blocks {
                 let block = block?;
-                let col_stats = gen_columns_statistics(&block)?;
+                let col_stats = gen_columns_statistics(&block, None)?;
 
-                let mut block_statistics = BlockStatistics::from(&block, "".to_owned(), None)?;
+                let mut block_statistics =
+                    BlockStatistics::from(&block, "".to_owned(), None, None)?;
                 let block_meta = block_writer.write(block, col_stats, None).await?;
                 block_statistics.block_file_location = block_meta.location.0.clone();
 

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

@@ -48,7 +48,7 @@ use crate::storages::fuse::table_test_fixture::TestFixture;
 fn test_ft_stats_block_stats() -> common_exception::Result<()> {
     let schema = DataSchemaRefExt::create(vec![DataField::new("a", i32::to_data_type())]);
     let block = DataBlock::create(schema, vec![Series::from_data(vec![1, 2, 3])]);
-    let r = gen_columns_statistics(&block)?;
+    let r = gen_columns_statistics(&block, None)?;
     assert_eq!(1, r.len());
     let col_stats = r.get(&0).unwrap();
     assert_eq!(col_stats.min, DataValue::Int64(1));
@@ -69,7 +69,7 @@ fn test_ft_tuple_stats_block_stats() -> common_exception::Result<()> {
     let column = StructColumn::from_data(inner_columns, tuple_data_type).arc();
 
     let block = DataBlock::create(schema, vec![column]);
-    let r = gen_columns_statistics(&block)?;
+    let r = gen_columns_statistics(&block, None)?;
     assert_eq!(2, r.len());
     let col0_stats = r.get(&0).unwrap();
     assert_eq!(col0_stats.min, DataValue::Int64(1));
@@ -89,7 +89,7 @@ fn test_ft_stats_col_stats_reduce() -> common_exception::Result<()> {
     let blocks = TestFixture::gen_sample_blocks_ex(num_of_blocks, rows_per_block, val_start_with);
     let col_stats = blocks
         .iter()
-        .map(|b| gen_columns_statistics(&b.clone().unwrap()))
+        .map(|b| gen_columns_statistics(&b.clone().unwrap(), None))
         .collect::<common_exception::Result<Vec<_>>>()?;
     let r = reducers::reduce_block_statistics(&col_stats, None);
     assert!(r.is_ok());
@@ -158,8 +158,9 @@ async fn test_accumulator() -> common_exception::Result<()> {
 
     for item in blocks {
         let block = item?;
-        let col_stats = gen_columns_statistics(&block)?;
-        let block_statistics = BlockStatistics::from(&block, "does_not_matter".to_owned(), None)?;
+        let col_stats = gen_columns_statistics(&block, None)?;
+        let block_statistics =
+            BlockStatistics::from(&block, "does_not_matter".to_owned(), None, None)?;
         let block_writer = BlockWriter::new(&operator, &loc_generator);
         let block_meta = block_writer.write(block, col_stats, None).await?;
         stats_acc.add_with_block_meta(block_meta, block_statistics)?;
@@ -344,7 +345,7 @@ fn test_ft_stats_block_stats_string_columns_trimming_using_eval() -> common_exce
         let max_expr = max_col.get(0);
 
         // generate the statistics of column
-        let stats_of_columns = gen_columns_statistics(&block).unwrap();
+        let stats_of_columns = gen_columns_statistics(&block, None).unwrap();
 
         // check if the max value (untrimmed) is in degenerated condition:
         // - the length of string value is larger or equal than STRING_PREFIX_LEN

src/query/service/tests/it/storages/statistics/column_statistics.rs

@@ -88,15 +88,15 @@ fn test_column_traverse() -> Result<()> {
     let cols = traverse::traverse_columns_dfs(sample_block.columns())?;
 
     assert_eq!(5, cols.len());
-    (0..5).for_each(|i| assert_eq!(cols[i], sample_cols[i], "checking col {}", i));
+    (0..5).for_each(|i| assert_eq!(cols[i].1, sample_cols[i], "checking col {}", i));
 
     Ok(())
 }
 
 #[test]
 fn test_column_statistic() -> Result<()> {
     let (sample_block, sample_cols) = gen_sample_block();
-    let col_stats = gen_columns_statistics(&sample_block)?;
+    let col_stats = gen_columns_statistics(&sample_block, None)?;
 
     assert_eq!(5, col_stats.len());
 

src/query/storages/fuse/fuse-result/src/result_table_sink.rs

@@ -176,7 +176,7 @@ impl Processor for ResultTableSink {
         match std::mem::replace(&mut self.state, State::None) {
             State::NeedSerialize(block) => {
                 let location = self.locations.gen_block_location();
-                let block_statistics = BlockStatistics::from(&block, location.clone(), None)?;
+                let block_statistics = BlockStatistics::from(&block, location.clone(), None, None)?;
 
                 let mut data = Vec::with_capacity(100 * 1024 * 1024);
                 let schema = block.schema().clone();

src/query/storages/fuse/fuse-result/src/writer.rs

@@ -101,7 +101,7 @@ impl ResultTableWriter {
     pub async fn append_block(&mut self, block: DataBlock) -> Result<PartInfoPtr> {
         let location = self.locations.gen_block_location();
         let mut data = Vec::with_capacity(100 * 1024 * 1024);
-        let block_statistics = BlockStatistics::from(&block, location.clone(), None)?;
+        let block_statistics = BlockStatistics::from(&block, location.clone(), None, None)?;
         let schema = block.schema().clone();
         let (size, meta_data) = serialize_data_blocks(vec![block], &schema, &mut data)?;
 

src/query/storages/fuse/fuse/src/operations/fuse_sink.rs

@@ -13,6 +13,7 @@
 //  limitations under the License.
 
 use std::any::Any;
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use async_trait::async_trait;
@@ -51,6 +52,33 @@ struct BloomIndexState {
     location: Location,
 }
 
+impl BloomIndexState {
+    pub fn try_create(
+        block: &DataBlock,
+        location: Location,
+    ) -> Result<(Self, HashMap<usize, usize>)> {
+        // write index
+        let bloom_index = BlockFilter::try_create(&[block])?;
+        let index_block = bloom_index.filter_block;
+        let mut data = Vec::with_capacity(100 * 1024);
+        let index_block_schema = &bloom_index.filter_schema;
+        let (size, _) = serialize_data_blocks_with_compression(
+            vec![index_block],
+            index_block_schema,
+            &mut data,
+            CompressionOptions::Uncompressed,
+        )?;
+        Ok((
+            Self {
+                data,
+                size,
+                location,
+            },
+            bloom_index.column_distinct_count,
+        ))
+    }
+}
+
 enum State {
     None,
     NeedSerialize(DataBlock),
@@ -168,28 +196,16 @@ impl Processor for FuseTableSink {
 
                 let (block_location, block_id) = self.meta_locations.gen_block_location();
 
-                let bloom_index_state = {
-                    // write index
-                    let bloom_index = BlockFilter::try_create(&[&block])?;
-                    let index_block = bloom_index.filter_block;
-                    let location = self.meta_locations.block_bloom_index_location(&block_id);
-                    let mut data = Vec::with_capacity(100 * 1024);
-                    let index_block_schema = &bloom_index.filter_schema;
-                    let (size, _) = serialize_data_blocks_with_compression(
-                        vec![index_block],
-                        index_block_schema,
-                        &mut data,
-                        CompressionOptions::Uncompressed,
-                    )?;
-                    BloomIndexState {
-                        data,
-                        size,
-                        location,
-                    }
-                };
+                let location = self.meta_locations.block_bloom_index_location(&block_id);
+                let (bloom_index_state, column_distinct_count) =
+                    BloomIndexState::try_create(&block, location)?;
 
-                let block_statistics =
-                    BlockStatistics::from(&block, block_location.0, cluster_stats)?;
+                let block_statistics = BlockStatistics::from(
+                    &block,
+                    block_location.0,
+                    cluster_stats,
+                    Some(column_distinct_count),
+                )?;
                 // we need a configuration of block size threshold here
                 let mut data = Vec::with_capacity(100 * 1024 * 1024);
                 let schema = block.schema().clone();

src/query/storages/fuse/fuse/src/operations/mutation/deletion_mutator.rs

@@ -83,7 +83,7 @@ impl DeletionMutator {
                 &self.base_mutator.data_accessor,
                 &self.base_mutator.location_generator,
             );
-            let col_stats = gen_columns_statistics(&replace_with)?;
+            let col_stats = gen_columns_statistics(&replace_with, None)?;
             let cluster_stats = self
                 .cluster_stats_gen
                 .gen_with_origin_stats(&replace_with, origin_stats)?;

src/query/storages/fuse/fuse/src/statistics/block_statistics.rs

@@ -34,12 +34,16 @@ impl BlockStatistics {
         data_block: &DataBlock,
         location: String,
         cluster_stats: Option<ClusterStatistics>,
+        column_distinct_count: Option<HashMap<usize, usize>>,
     ) -> common_exception::Result<BlockStatistics> {
         Ok(BlockStatistics {
             block_file_location: location,
             block_rows_size: data_block.num_rows() as u64,
             block_bytes_size: data_block.memory_size() as u64,
-            block_column_statistics: column_statistic::gen_columns_statistics(data_block)?,
+            block_column_statistics: column_statistic::gen_columns_statistics(
+                data_block,
+                column_distinct_count,
+            )?,
             block_cluster_statistics: cluster_stats,
         })
     }

src/query/storages/fuse/fuse/src/statistics/column_statistic.rs

@@ -12,10 +12,12 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use common_datablocks::DataBlock;
 use common_datavalues::Column;
+use common_datavalues::ColumnRef;
 use common_datavalues::ColumnWithField;
 use common_datavalues::DataField;
 use common_datavalues::DataValue;
@@ -39,16 +41,19 @@ pub fn calc_column_distinct_of_values(
     distinct_values.get(0).as_u64()
 }
 
-pub fn get_traverse_columns_dfs(data_block: &DataBlock) -> Result<Vec<Arc<dyn Column>>> {
+pub fn get_traverse_columns_dfs(data_block: &DataBlock) -> Result<Vec<(Option<usize>, ColumnRef)>> {
     traverse::traverse_columns_dfs(data_block.columns())
 }
 
-pub fn gen_columns_statistics(data_block: &DataBlock) -> Result<StatisticsOfColumns> {
+pub fn gen_columns_statistics(
+    data_block: &DataBlock,
+    column_distinct_count: Option<HashMap<usize, usize>>,
+) -> Result<StatisticsOfColumns> {
     let mut statistics = StatisticsOfColumns::new();
 
     let leaves = traverse::traverse_columns_dfs(data_block.columns())?;
 
-    for (idx, col) in leaves.iter().enumerate() {
+    for (idx, (col_idx, col)) in leaves.iter().enumerate() {
         let col_data_type = col.data_type();
         if !MinMaxIndex::is_supported_type(&col_data_type) {
             continue;
@@ -81,7 +86,17 @@ pub fn gen_columns_statistics(data_block: &DataBlock) -> Result<StatisticsOfColu
             }
         }
 
-        let distinct_of_values = calc_column_distinct_of_values(col, column_field)?;
+        // use distinct count calculated by the xor hash function to avoid repetitive operation.
+        let distinct_of_values = match (col_idx, &column_distinct_count) {
+            (Some(col_idx), Some(ref column_distinct_count)) => {
+                if let Some(value) = column_distinct_count.get(col_idx) {
+                    *value as u64
+                } else {
+                    calc_column_distinct_of_values(col, column_field)?
+                }
+            }
+            (_, _) => calc_column_distinct_of_values(col, column_field)?,
+        };
 
         let (is_all_null, bitmap) = col.validity();
         let unset_bits = match (is_all_null, bitmap) {
@@ -113,25 +128,29 @@ pub mod traverse {
     use super::*;
 
     // traverses columns and collects the leaves in depth first manner
-    pub fn traverse_columns_dfs(columns: &[ColumnRef]) -> Result<Vec<ColumnRef>> {
+    pub fn traverse_columns_dfs(columns: &[ColumnRef]) -> Result<Vec<(Option<usize>, ColumnRef)>> {
         let mut leaves = vec![];
-        for f in columns {
-            traverse_recursive(f, &mut leaves)?;
+        for (idx, col) in columns.iter().enumerate() {
+            traverse_recursive(Some(idx), col, &mut leaves)?;
         }
         Ok(leaves)
     }
 
-    fn traverse_recursive(column: &ColumnRef, leaves: &mut Vec<ColumnRef>) -> Result<()> {
+    fn traverse_recursive(
+        idx: Option<usize>,
+        column: &ColumnRef,
+        leaves: &mut Vec<(Option<usize>, ColumnRef)>,
+    ) -> Result<()> {
         match column.data_type() {
             DataTypeImpl::Struct(_) => {
                 let full_column = column.convert_full_column();
                 let struct_col: &StructColumn = Series::check_get(&full_column)?;
-                for f in struct_col.values() {
-                    traverse_recursive(f, leaves)?
+                for col in struct_col.values() {
+                    traverse_recursive(None, col, leaves)?
                 }
             }
             _ => {
-                leaves.push(column.clone());
+                leaves.push((idx, column.clone()));
             }
         }
         Ok(())

src/query/storages/fuse/fuse/src/statistics/reducers.rs

@@ -95,7 +95,7 @@ pub fn reduce_block_statistics<T: Borrow<StatisticsOfColumns>>(
 
             let distinct_of_values = match data_block {
                 Some(_data_block) => {
-                    if let Some(col) = leaves.as_ref().unwrap().get(*id as usize) {
+                    if let Some((_, col)) = leaves.as_ref().unwrap().get(*id as usize) {
                         let col_data_type = col.data_type();
                         let data_field = DataField::new("", col_data_type);
                         let column_field = ColumnWithField::new(col.clone(), data_field);