fix: them map build_segment_id_map returns should contain format ver (

#11212)

src/query/storages/common/table-meta/src/meta/v3/snapshot.rs

@@ -167,11 +167,11 @@ impl TableSnapshot {
         Encoding::default()
     }
 
-    pub fn build_segment_id_map(&self) -> HashMap<String, usize> {
+    pub fn build_segment_id_map(&self) -> HashMap<Location, usize> {
         let segment_count = self.segments.len();
         let mut segment_id_map = HashMap::new();
         for (i, segment_loc) in self.segments.iter().enumerate() {
-            segment_id_map.insert(segment_loc.0.to_string(), segment_count - i - 1);
+            segment_id_map.insert(segment_loc.clone(), segment_count - i - 1);
         }
         segment_id_map
     }

src/query/storages/fuse/src/operations/read/fuse_row_fetcher.rs

@@ -173,12 +173,12 @@ where F: RowFetcher + Send + Sync + 'static
             MetaReaders::segment_info_reader(self.table.operator.clone(), table_schema);
         let mut map = HashMap::new();
 
-        for (location, seg_id) in segment_id_map.iter() {
+        for ((location, ver), seg_id) in segment_id_map.into_iter() {
             let segment_info = segment_reader
                 .read(&LoadParams {
-                    location: location.clone(),
+                    location,
                     len_hint: None,
-                    ver: snapshot.format_version, // TODO this is buggy, should use segment version
+                    ver,
                     put_cache: true,
                 })
                 .await?;
@@ -190,7 +190,7 @@ where F: RowFetcher + Send + Sync + 'static
                     None,
                     &self.projection,
                 );
-                let part_id = compute_row_id_prefix(*seg_id as u64, block_idx as u64);
+                let part_id = compute_row_id_prefix(seg_id as u64, block_idx as u64);
                 map.insert(part_id, part_info);
             }
         }

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

@@ -114,7 +114,7 @@ impl FuseTable {
         table_info: TableInfo,
         segments_location: Vec<Location>,
         summary: usize,
-        segment_id_map: Option<HashMap<String, usize>>,
+        segment_id_map: Option<HashMap<Location, usize>>,
     ) -> Result<(PartStatistics, Partitions)> {
         let start = Instant::now();
         info!(

src/query/storages/fuse/src/pruning/fuse_pruner.rs

@@ -170,7 +170,7 @@ impl FusePruner {
         &self,
         segment_locs: Vec<Location>,
         snapshot_loc: Option<String>,
-        segment_id_map: Option<HashMap<String, usize>>,
+        segment_id_map: Option<HashMap<Location, usize>>,
     ) -> Result<Vec<(BlockMetaIndex, Arc<BlockMeta>)>> {
         let segment_locs =
             create_segment_location_vector(segment_locs, snapshot_loc, segment_id_map);

src/query/storages/fuse/src/pruning/pruner_location.rs

@@ -26,26 +26,26 @@ pub struct SegmentLocation {
 pub fn create_segment_location_vector(
     locations: Vec<Location>,
     snapshot_loc: Option<String>,
-    segment_id_map: Option<HashMap<String, usize>>,
+    segment_id_map: Option<HashMap<Location, usize>>,
 ) -> Vec<SegmentLocation> {
     let segment_count = locations.len();
     if let Some(segment_id_map) = segment_id_map {
         let mut seg_locations = Vec::with_capacity(segment_count);
-        for (location, version) in locations {
+        for location in locations {
             seg_locations.push(SegmentLocation {
                 segment_id: *segment_id_map.get(&location).unwrap(),
-                location: (location.clone(), version),
+                location,
                 snapshot_loc: snapshot_loc.clone(),
             });
         }
 
         seg_locations
     } else {
         let mut seg_locations = Vec::with_capacity(segment_count);
-        for (i, location) in locations.iter().enumerate() {
+        for (i, location) in locations.into_iter().enumerate() {
             seg_locations.push(SegmentLocation {
                 segment_id: segment_count - i - 1,
-                location: location.to_owned(),
+                location,
                 snapshot_loc: snapshot_loc.clone(),
             });
         }