use metahll

Author: zhyass

src/query/storages/common/table_meta/src/meta/statistics.rs

@@ -25,7 +25,6 @@ pub type SnapshotId = Uuid;
 pub type Location = (String, FormatVersion);
 pub type ClusterKey = (u32, String);
 pub type StatisticsOfColumns = HashMap<ColumnId, ColumnStatistics>;
-pub type ColumnDistinctHLL = simple_hll::HyperLogLog<10>;
 
 // Assigned to executors, describes that which blocks of given segment, an executor should take care of
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]

src/query/storages/common/table_meta/src/meta/v4/block_statistics.rs

@@ -26,20 +26,20 @@ use crate::meta::format::compress;
 use crate::meta::format::encode;
 use crate::meta::format::read_and_deserialize;
 use crate::meta::versions::Versioned;
-use crate::meta::ColumnDistinctHLL;
 use crate::meta::FormatVersion;
 use crate::meta::MetaCompression;
 use crate::meta::MetaEncoding;
+use crate::meta::MetaHLL;
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct BlockStatistics {
     pub format_version: FormatVersion,
 
-    pub hll: HashMap<ColumnId, ColumnDistinctHLL>,
+    pub hll: HashMap<ColumnId, MetaHLL>,
 }
 
 impl BlockStatistics {
-    pub fn new(hll: HashMap<ColumnId, ColumnDistinctHLL>) -> Self {
+    pub fn new(hll: HashMap<ColumnId, MetaHLL>) -> Self {
         Self {
             format_version: BlockStatistics::VERSION,
             hll,

src/query/storages/fuse/src/io/write/stream/block_builder.rs

@@ -318,7 +318,9 @@ impl StreamBlockBuilder {
             .block_stats_location(&block_id);
         let block_stats_state = self.block_stats_builder.finalize(block_stats_location)?;
         if let Some(state) = &block_stats_state {
-            column_distinct_count.extend(state.column_distinct_count.clone());
+            for (key, val) in &state.column_distinct_count {
+                column_distinct_count.entry(*key).or_insert(*val);
+            }
         }
         let col_stats = self.column_stats_state.finalize(column_distinct_count)?;
 

src/query/storages/fuse/src/io/write/stream/column_ndv_estimator.rs

@@ -39,15 +39,15 @@ use databend_common_expression::with_number_type;
 use databend_common_expression::Column;
 use databend_common_expression::ScalarRef;
 use databend_common_expression::SELECTIVITY_THRESHOLD;
-use databend_storages_common_table_meta::meta::ColumnDistinctHLL;
+use databend_storages_common_table_meta::meta::MetaHLL;
 use enum_dispatch::enum_dispatch;
 
 #[enum_dispatch]
 pub trait ColumnNDVEstimatorOps: Send + Sync {
     fn update_column(&mut self, column: &Column);
     fn update_scalar(&mut self, scalar: &ScalarRef);
     fn finalize(&self) -> usize;
-    fn hll(self) -> ColumnDistinctHLL;
+    fn hll(self) -> MetaHLL;
 }
 
 #[enum_dispatch(ColumnNDVEstimatorOps)]
@@ -111,7 +111,7 @@ where
     T: ValueType + Send + Sync,
     for<'a> T::ScalarRef<'a>: Hash,
 {
-    hll: ColumnDistinctHLL,
+    hll: MetaHLL,
     _phantom: PhantomData<T>,
 }
 
@@ -122,7 +122,7 @@ where
 {
     pub fn new() -> Self {
         Self {
-            hll: ColumnDistinctHLL::new(),
+            hll: MetaHLL::new(),
             _phantom: Default::default(),
         }
     }
@@ -181,7 +181,7 @@ where
         self.hll.count()
     }
 
-    fn hll(self) -> ColumnDistinctHLL {
+    fn hll(self) -> MetaHLL {
         self.hll
     }
 }