fix: deserialization error of AdditionalStatsMeta.location (#18618)

* Fix deserialization error of AdditionalStatsMeta.location

* fix

* fix

* fix test

Author: zhyass

src/query/ee/src/storages/fuse/operations/table_index.rs

@@ -145,10 +145,8 @@ pub async fn do_refresh_table_index(
                 put_cache: false,
             })
             .await?;
-        let stats = match &segment_info.summary.additional_stats_meta {
-            Some(meta) if meta.location.is_some() => {
-                Some(read_segment_stats(operator.clone(), meta.location.clone().unwrap()).await?)
-            }
+        let stats = match segment_info.summary.additional_stats_loc() {
+            Some(loc) => Some(read_segment_stats(operator.clone(), loc).await?),
             _ => None,
         };
 

src/query/ee/src/storages/fuse/operations/virtual_columns.rs

@@ -136,10 +136,8 @@ pub async fn do_refresh_virtual_column(
                 put_cache: false,
             })
             .await?;
-        let stats = match &segment_info.summary.additional_stats_meta {
-            Some(meta) if meta.location.is_some() => {
-                Some(read_segment_stats(operator.clone(), meta.location.clone().unwrap()).await?)
-            }
+        let stats = match segment_info.summary.additional_stats_loc() {
+            Some(loc) => Some(read_segment_stats(operator.clone(), loc).await?),
             _ => None,
         };
 

src/query/service/src/test_kits/fuse.rs

@@ -130,7 +130,7 @@ pub async fn generate_segments_v2(
             TableMetaLocationGenerator::gen_segment_stats_location_from_segment_location(&location);
         let additional_stats_meta = AdditionalStatsMeta {
             size: stats.len() as u64,
-            location: Some((stats_location.clone(), SegmentStatistics::VERSION)),
+            location: (stats_location.clone(), SegmentStatistics::VERSION),
             ..Default::default()
         };
         dal.write(&stats_location, stats).await?;
@@ -173,7 +173,7 @@ pub async fn generate_segments(
             TableMetaLocationGenerator::gen_segment_stats_location_from_segment_location(&location);
         let additional_stats_meta = AdditionalStatsMeta {
             size: stats.len() as u64,
-            location: Some((stats_location.clone(), SegmentStatistics::VERSION)),
+            location: (stats_location.clone(), SegmentStatistics::VERSION),
             ..Default::default()
         };
         dal.write(&stats_location, stats).await?;

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

@@ -76,14 +76,19 @@ pub struct AdditionalStatsMeta {
     /// The size of the stats data in bytes.
     pub size: u64,
     /// The file location of the stats data.
-    pub location: Option<Location>,
+    #[serde(default = "default_location")]
+    pub location: Location,
     /// An optional HyperLogLog data structure.
     pub hll: Option<RawBlockHLL>,
     /// The count of the stats rows.
     #[serde(default)]
     pub row_count: u64,
 }
 
+fn default_location() -> Location {
+    ("".to_string(), 0)
+}
+
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, Default)]
 pub struct Statistics {
     pub row_count: u64,
@@ -273,6 +278,13 @@ impl Statistics {
             additional_stats_meta: None,
         }
     }
+
+    pub fn additional_stats_loc(&self) -> Option<Location> {
+        match &self.additional_stats_meta {
+            Some(meta) if !meta.location.0.is_empty() => Some(meta.location.clone()),
+            _ => None,
+        }
+    }
 }
 
 /// Serializes a `Scalar` value by first converting it to `IndexScalar`.

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

@@ -667,13 +667,12 @@ impl Processor for AnalyzeCollectNDVSource {
                 let compact_segment_info = self.segment_reader.read(&load_param).await?;
 
                 let block_count = compact_segment_info.summary.block_count as usize;
-                let block_hlls = match &compact_segment_info.summary.additional_stats_meta {
-                    Some(meta) if meta.location.is_some() => {
-                        let (path, ver) = meta.location.as_ref().unwrap();
+                let block_hlls = match compact_segment_info.summary.additional_stats_loc() {
+                    Some((path, ver)) => {
                         let load_param = LoadParams {
-                            location: path.clone(),
+                            location: path,
                             len_hint: None,
-                            ver: *ver,
+                            ver,
                             put_cache: true,
                         };
                         let stats = self.stats_reader.read(&load_param).await?;
@@ -744,7 +743,7 @@ impl Processor for AnalyzeCollectNDVSource {
                     );
                 let additional_stats_meta = AdditionalStatsMeta {
                     size: data.len() as u64,
-                    location: Some((segment_stats_location.clone(), SegmentStatistics::VERSION)),
+                    location: (segment_stats_location.clone(), SegmentStatistics::VERSION),
                     ..Default::default()
                 };
                 self.dal.write(&segment_stats_location, data).await?;

src/query/storages/fuse/src/operations/common/processors/transform_mutation_aggregator.rs

@@ -503,10 +503,8 @@ impl TableMutationAggregator {
                         SegmentsIO::read_compact_segment(op.clone(), loc, schema, false).await?;
                     let mut segment_info = SegmentInfo::try_from(compact_segment_info)?;
 
-                    let stats = match &segment_info.summary.additional_stats_meta {
-                        Some(meta) if meta.location.is_some() => Some(
-                            read_segment_stats(op.clone(), meta.location.clone().unwrap()).await?,
-                        ),
+                    let stats = match segment_info.summary.additional_stats_loc() {
+                        Some(loc) => Some(read_segment_stats(op.clone(), loc).await?),
                         _ => None,
                     };
 
@@ -865,7 +863,7 @@ async fn write_segment(
             );
         let additional_stats_meta = AdditionalStatsMeta {
             size: stats.len() as u64,
-            location: Some((segment_stats_location.clone(), SegmentStatistics::VERSION)),
+            location: (segment_stats_location.clone(), SegmentStatistics::VERSION),
             ..Default::default()
         };
         dal.write(&segment_stats_location, stats).await?;

src/query/storages/fuse/src/operations/common/processors/transform_serialize_segment.rs

@@ -282,10 +282,7 @@ impl<B: SegmentBuilder> Processor for TransformSerializeSegment<B> {
                     let stats_summary = self.hll_accumulator.take_summary();
                     additional_stats_meta = Some(AdditionalStatsMeta {
                         size: stats_data.len() as u64,
-                        location: Some((
-                            segment_stats_location.clone(),
-                            SegmentStatistics::VERSION,
-                        )),
+                        location: (segment_stats_location.clone(), SegmentStatistics::VERSION),
                         ..Default::default()
                     });
                     stats = Some((segment_stats_location, stats_data, stats_summary));

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

@@ -847,10 +847,8 @@ impl TryFrom<Arc<CompactSegmentInfo>> for LocationTuple {
                 bloom_location.insert(bloom_loc.0.clone());
             }
         }
-        if let Some(meta) = &value.as_ref().summary.additional_stats_meta {
-            if let Some(loc) = &meta.location {
-                hll_location.insert(loc.0.clone());
-            }
+        if let Some(loc) = value.as_ref().summary.additional_stats_loc() {
+            hll_location.insert(loc.0);
         }
         Ok(Self {
             block_location,
@@ -884,10 +882,8 @@ impl TryFrom<Arc<ColumnOrientedSegment>> for LocationTuple {
             }
         }
 
-        if let Some(meta) = &value.as_ref().summary.additional_stats_meta {
-            if let Some(loc) = &meta.location {
-                hll_location.insert(loc.0.clone());
-            }
+        if let Some(loc) = value.as_ref().summary.additional_stats_loc() {
+            hll_location.insert(loc.0);
         }
         Ok(Self {
             block_location,

src/query/storages/fuse/src/operations/mutation/mutator/block_compact_mutator.rs

@@ -570,10 +570,8 @@ impl CompactTaskBuilder {
             let permit = acquire_task_permit(semaphore.clone()).await?;
             let op = self.dal.clone();
             let handler = runtime.spawn(async move {
-                let stats = match &segment.summary.additional_stats_meta {
-                    Some(meta) if meta.location.is_some() => {
-                        Some(read_segment_stats(op.clone(), meta.location.clone().unwrap()).await?)
-                    }
+                let stats = match segment.summary.additional_stats_loc() {
+                    Some(loc) => Some(read_segment_stats(op.clone(), loc).await?),
                     _ => None,
                 };
                 let blocks = segment.block_metas()?;

src/query/storages/fuse/src/operations/mutation/mutator/recluster_mutator.rs

@@ -188,7 +188,7 @@ impl ReclusterMutator {
                     info.summary
                         .additional_stats_meta
                         .as_ref()
-                        .and_then(|v| v.location.clone()),
+                        .map(|v| v.location.clone()),
                 ));
                 (loc.segment_idx, info)
             })