update

Author: zhyass

src/common/storage/src/meta_hll.rs

@@ -117,18 +117,6 @@ impl MetaHLL {
     pub fn num_empty_registers(&self) -> usize {
         self.registers.iter().filter(|x| **x == 0).count()
     }
-
-    pub fn from_hll(hll: MetaHLL12) -> Self {
-        let registers = hll.get_registers();
-        let mut new_registers = vec![0; M];
-        let group_size = registers.len() / M;
-        for i in 0..M {
-            for j in 0..group_size {
-                new_registers[i] = new_registers[i].max(registers[i * group_size + j]);
-            }
-        }
-        Self::with_registers(new_registers)
-    }
 }
 
 /// Helper function sigma as defined in
@@ -180,6 +168,20 @@ fn hll_tau(x: f64) -> f64 {
     }
 }
 
+impl From<MetaHLL12> for MetaHLL {
+    fn from(value: MetaHLL12) -> Self {
+        let registers = value.get_registers();
+        let mut new_registers = vec![0; M];
+        let group_size = registers.len() / M;
+        for i in 0..M {
+            for j in 0..group_size {
+                new_registers[i] = new_registers[i].max(registers[i * group_size + j]);
+            }
+        }
+        Self::with_registers(new_registers)
+    }
+}
+
 #[derive(serde::Serialize, borsh::BorshSerialize)]
 enum MetaHLLVariantRef<'a> {
     Empty,
@@ -393,7 +395,7 @@ mod tests {
             hll.add_object(&i);
         }
 
-        let hll = MetaHLL::from_hll(hll);
+        let hll = MetaHLL::from(hll);
         let count = hll.count();
         let error_rate = 1.04 / ((M as f64).sqrt());
         let diff = count as f64 / 100_000f64;

src/query/service/src/interpreters/interpreter_table_analyze.rs

@@ -136,6 +136,11 @@ impl Interpreter for AnalyzeTableInterpreter {
         let table_statistics = table
             .read_table_snapshot_statistics(Some(&snapshot))
             .await?;
+        if let Some(table_statistics) = &table_statistics {
+            if table_statistics.snapshot_id == snapshot.snapshot_id {
+                return Ok(PipelineBuildResult::create());
+            }
+        }
 
         // plan sql
         let (is_full, temporal_str) = if let Some(table_statistics) = &table_statistics {

src/query/service/src/pipelines/builders/builder_mutation_source.rs

@@ -80,7 +80,7 @@ impl PipelineBuilder {
                 Vec::with_capacity(mutation_source.partitions.partitions.len());
             for part in &mutation_source.partitions.partitions {
                 // Safe to downcast because we know the partition is lazy
-                let part: &FuseLazyPartInfo = FuseLazyPartInfo::from_part(part)?;
+                let part = FuseLazyPartInfo::from_part(part)?;
                 segment_locations.push(SegmentLocation {
                     segment_idx: part.segment_index,
                     location: part.segment_location.clone(),

src/query/service/tests/it/storages/fuse/operations/table_analyze.rs

@@ -219,7 +219,7 @@ async fn test_table_analyze_without_prev_table_seq() -> Result<()> {
     let hll: HashMap<ColumnId, MetaHLL12> =
         HashMap::from([(0, borsh_deserialize_from_slice(&col)?)]);
     let table_statistics =
-        TableSnapshotStatistics::new(hll, HashMap::new(), snapshot_1.snapshot_id);
+        TableSnapshotStatistics::new(hll, HashMap::new(), snapshot_1.snapshot_id, 14);
     let table_statistics_location = location_gen.snapshot_statistics_location_from_uuid(
         &table_statistics.snapshot_id,
         table_statistics.format_version(),

src/query/storages/common/table_meta/src/meta/current/mod.rs

@@ -25,14 +25,13 @@ pub use v2::SegmentStatistics;
 pub use v2::Statistics;
 pub use v2::VirtualBlockMeta;
 pub use v2::VirtualColumnMeta;
-pub use v3::TableSnapshotStatistics;
 pub use v4::CompactSegmentInfo;
 pub use v4::RawBlockMeta;
 pub use v4::SegmentInfo;
 pub use v4::TableSnapshot;
 pub use v4::TableSnapshotLite;
+pub use v4::TableSnapshotStatistics;
 
 use super::v0;
 use super::v2;
-use super::v3;
 use super::v4;

src/query/storages/common/table_meta/src/meta/v3/table_snapshot_statistics.rs

@@ -68,7 +68,7 @@ impl From<v2::TableSnapshotStatistics> for TableSnapshotStatistics {
         Self {
             format_version: TableSnapshotStatistics::VERSION,
             snapshot_id: value.snapshot_id,
-            hll: HashMap::new(),
+            hll: value.hll,
             histograms: HashMap::new(),
         }
     }

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

@@ -21,3 +21,4 @@ pub use segment::RawBlockMeta;
 pub use segment::SegmentInfo;
 pub use snapshot::TableSnapshot;
 pub use snapshot::TableSnapshotLite;
+pub use table_snapshot_statistics::TableSnapshotStatistics;

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

@@ -18,8 +18,12 @@ use databend_common_expression::ColumnId;
 use databend_common_storage::Histogram;
 use databend_common_storage::MetaHLL;
 
+use crate::meta::v1;
+use crate::meta::v2;
+use crate::meta::v3;
 use crate::meta::FormatVersion;
 use crate::meta::SnapshotId;
+use crate::meta::Versioned;
 
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug)]
 pub struct TableSnapshotStatistics {
@@ -31,3 +35,69 @@ pub struct TableSnapshotStatistics {
     pub hll: HashMap<ColumnId, MetaHLL>,
     pub histograms: HashMap<ColumnId, Histogram>,
 }
+
+impl TableSnapshotStatistics {
+    pub fn new(
+        hll: HashMap<ColumnId, MetaHLL>,
+        histograms: HashMap<ColumnId, Histogram>,
+        snapshot_id: SnapshotId,
+        row_count: usize,
+    ) -> Self {
+        Self {
+            format_version: TableSnapshotStatistics::VERSION,
+            snapshot_id,
+            hll,
+            histograms,
+            row_count,
+        }
+    }
+
+    pub fn format_version(&self) -> u64 {
+        self.format_version
+    }
+
+    pub fn column_distinct_values(&self) -> HashMap<ColumnId, u64> {
+        self.hll
+            .iter()
+            .map(|hll| (*hll.0, hll.1.count() as u64))
+            .collect()
+    }
+}
+
+impl From<v1::TableSnapshotStatistics> for TableSnapshotStatistics {
+    fn from(value: v1::TableSnapshotStatistics) -> Self {
+        Self {
+            format_version: TableSnapshotStatistics::VERSION,
+            snapshot_id: value.snapshot_id,
+            row_count: 0,
+            hll: HashMap::new(),
+            histograms: HashMap::new(),
+        }
+    }
+}
+
+impl From<v2::TableSnapshotStatistics> for TableSnapshotStatistics {
+    fn from(value: v2::TableSnapshotStatistics) -> Self {
+        let hll = value.hll.into_iter().map(|(k, v)| (k, v.into())).collect();
+        Self {
+            format_version: TableSnapshotStatistics::VERSION,
+            snapshot_id: value.snapshot_id,
+            row_count: 0,
+            hll,
+            histograms: HashMap::new(),
+        }
+    }
+}
+
+impl From<v3::TableSnapshotStatistics> for TableSnapshotStatistics {
+    fn from(value: v3::TableSnapshotStatistics) -> Self {
+        let hll = value.hll.into_iter().map(|(k, v)| (k, v.into())).collect();
+        Self {
+            format_version: TableSnapshotStatistics::VERSION,
+            snapshot_id: value.snapshot_id,
+            row_count: 0,
+            hll,
+            histograms: value.histograms,
+        }
+    }
+}

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

@@ -105,13 +105,15 @@ impl SnapshotVersion {
 impl Versioned<0> for v1::TableSnapshotStatistics {}
 impl Versioned<2> for v2::TableSnapshotStatistics {}
 impl Versioned<3> for v3::TableSnapshotStatistics {}
+impl Versioned<4> for v4::TableSnapshotStatistics {}
 
 impl Versioned<2> for DataBlock {}
 
 pub enum TableSnapshotStatisticsVersion {
     V0(PhantomData<v1::TableSnapshotStatistics>),
     V2(PhantomData<v2::TableSnapshotStatistics>),
     V3(PhantomData<v3::TableSnapshotStatistics>),
+    V4(PhantomData<v4::TableSnapshotStatistics>),
 }
 
 impl TableSnapshotStatisticsVersion {
@@ -120,6 +122,7 @@ impl TableSnapshotStatisticsVersion {
             TableSnapshotStatisticsVersion::V0(a) => Self::ver(a),
             TableSnapshotStatisticsVersion::V2(a) => Self::ver(a),
             TableSnapshotStatisticsVersion::V3(a) => Self::ver(a),
+            TableSnapshotStatisticsVersion::V4(a) => Self::ver(a),
         }
     }
 
@@ -206,8 +209,11 @@ mod converters {
                 3 => Ok(TableSnapshotStatisticsVersion::V3(testify_version::<_, 3>(
                     PhantomData,
                 ))),
+                4 => Ok(TableSnapshotStatisticsVersion::V4(testify_version::<_, 4>(
+                    PhantomData,
+                ))),
                 _ => Err(ErrorCode::Internal(format!(
-                    "unknown table snapshot statistics version {value}, versions supported: 0, 2, 3"
+                    "unknown table snapshot statistics version {value}, versions supported: 0, 2, 3, 4"
                 ))),
             }
         }

src/query/storages/common/table_meta/src/readers/versioned_reader.rs

@@ -43,7 +43,11 @@ impl VersionedReader<TableSnapshotStatistics> for TableSnapshotStatisticsVersion
                 let ts = load_json(reader, v)?;
                 TableSnapshotStatistics::from(ts)
             }
-            TableSnapshotStatisticsVersion::V3(v) => load_json(reader, v)?,
+            TableSnapshotStatisticsVersion::V3(v) => {
+                let ts = load_json(reader, v)?;
+                TableSnapshotStatistics::from(ts)
+            }
+            TableSnapshotStatisticsVersion::V4(v) => load_json(reader, v)?,
         };
         Ok(r)
     }