add SegmentDescriptor

Author: dantengsky

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

@@ -97,7 +97,7 @@ pub async fn get_snapshot_referenced_files(
         format_version: ver,
         snapshot_id: root_snapshot.snapshot_id,
         timestamp: root_snapshot.timestamp,
-        segments: HashSet::from_iter(root_snapshot.segments.clone()),
+        segments: HashSet::from_iter(root_snapshot.segments.iter().map(|v| v.location.clone())),
         table_statistics_location: root_snapshot.table_statistics_location.clone(),
     });
     drop(root_snapshot);

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

@@ -99,7 +99,11 @@ pub async fn do_refresh_virtual_column(
     let segment_locs = if let Some(segment_locs) = segment_locs {
         segment_locs
     } else {
-        snapshot.segments.clone()
+        snapshot
+            .segments
+            .iter()
+            .map(|v| v.location.clone())
+            .collect()
     };
 
     // Read source variant columns and extract inner fields as virtual columns.

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

@@ -55,13 +55,14 @@ async fn test_fuse_do_refresh_virtual_column() -> Result<()> {
     let write_settings = fuse_table.get_write_settings();
     let storage_format = write_settings.storage_format;
 
-    let segment_locs = Some(snapshot.segments.clone());
+    let segment_locs = Some(snapshot.segments.iter().map(|v| v.location.clone()).collect());
     do_refresh_virtual_column(fuse_table, table_ctx, virtual_columns, segment_locs).await?;
 
     let segment_reader =
         MetaReaders::segment_info_reader(fuse_table.get_operator(), table_schema.clone());
 
-    for (location, ver) in &snapshot.segments {
+    for seg in &snapshot.segments {
+        let (location, ver) = &seg.location;
         let segment_info = segment_reader
             .read(&LoadParams {
                 location: location.to_string(),

src/query/service/src/interpreters/interpreter_merge_into.rs

@@ -441,6 +441,7 @@ impl MergeIntoInterpreter {
             .segments
             .clone()
             .into_iter()
+            .map(|v| v.location)
             .enumerate()
             .collect();
 

src/query/service/src/interpreters/interpreter_replace.rs

@@ -308,6 +308,7 @@ impl ReplaceInterpreter {
                 .segments
                 .clone()
                 .into_iter()
+                .map(|v| v.location)
                 .enumerate()
                 .collect(),
             block_slots: None,

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

@@ -67,7 +67,7 @@ pub async fn generate_snapshot_with_segments(
     let operator = fuse_table.get_operator();
     let location_gen = fuse_table.meta_location_generator();
     let mut new_snapshot = TableSnapshot::from_previous(current_snapshot.as_ref());
-    new_snapshot.segments = segment_locations;
+    new_snapshot.segments = segment_locations.into_iter().map(|v| v.into()).collect();
     let new_snapshot_location = location_gen
         .snapshot_location_from_uuid(&new_snapshot.snapshot_id, TableSnapshot::VERSION)?;
     if let Some(ts) = time_stamp {
@@ -199,6 +199,7 @@ pub async fn generate_snapshots(fixture: &TestFixture) -> Result<()> {
 
     // create snapshot 1, the format version is 3.
     let locations = vec![segments_v3[0].0.clone(), segments_v2[0].0.clone()];
+    let locations = locations.into_iter().map(|v| v.into()).collect();
     let mut snapshot_1 = TableSnapshot::new(
         Uuid::new_v4(),
         &snapshot_0.timestamp,
@@ -224,7 +225,7 @@ pub async fn generate_snapshots(fixture: &TestFixture) -> Result<()> {
         segments_v2[0].0.clone(),
     ];
     let mut snapshot_2 = TableSnapshot::from_previous(&snapshot_1);
-    snapshot_2.segments = locations;
+    snapshot_2.segments = locations.into_iter().map(|v| v.into()).collect();
     snapshot_2.timestamp = Some(now);
     snapshot_2.summary = merge_statistics(&snapshot_1.summary, &segments_v3[1].1.summary, None);
     let new_snapshot_location = location_gen

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

@@ -37,13 +37,13 @@ use databend_storages_common_table_meta::meta::TableSnapshot;
 fn test_unresolvable_delete_conflict() {
     let mut base_snapshot = TableSnapshot::new_empty_snapshot(TableSchema::default());
     base_snapshot.segments = vec![
-        ("1".to_string(), 1),
-        ("2".to_string(), 1),
-        ("3".to_string(), 1),
+        ("1".to_string(), 1).into(),
+        ("2".to_string(), 1).into(),
+        ("3".to_string(), 1).into(),
     ];
 
     let mut latest_snapshot = TableSnapshot::new_empty_snapshot(TableSchema::default());
-    latest_snapshot.segments = vec![("1".to_string(), 1), ("4".to_string(), 1)];
+    latest_snapshot.segments = vec![("1".to_string(), 1).into(), ("4".to_string(), 1).into()];
 
     let ctx = ConflictResolveContext::ModifiedSegmentExistsInLatest(SnapshotChanges {
         appended_segments: vec![],
@@ -77,9 +77,9 @@ fn test_unresolvable_delete_conflict() {
 fn test_resolvable_delete_conflict() {
     let mut base_snapshot = TableSnapshot::new_empty_snapshot(TableSchema::default());
     base_snapshot.segments = vec![
-        ("1".to_string(), 1),
-        ("2".to_string(), 1),
-        ("3".to_string(), 1),
+        ("1".to_string(), 1).into(),
+        ("2".to_string(), 1).into(),
+        ("3".to_string(), 1).into(),
     ];
 
     base_snapshot.summary = Statistics {
@@ -95,9 +95,9 @@ fn test_resolvable_delete_conflict() {
 
     let mut latest_snapshot = TableSnapshot::new_empty_snapshot(TableSchema::default());
     latest_snapshot.segments = vec![
-        ("2".to_string(), 1),
-        ("3".to_string(), 1),
-        ("4".to_string(), 1),
+        ("2".to_string(), 1).into(),
+        ("3".to_string(), 1).into(),
+        ("4".to_string(), 1).into(),
     ];
 
     latest_snapshot.summary = Statistics {
@@ -135,7 +135,7 @@ fn test_resolvable_delete_conflict() {
 
     let ctx = ConflictResolveContext::ModifiedSegmentExistsInLatest(SnapshotChanges {
         appended_segments: vec![],
-        replaced_segments: HashMap::from([(2, ("8".to_string(), 1))]),
+        replaced_segments: HashMap::from([(2, ("8".to_string(), 1).into())]),
         removed_segment_indexes: vec![1],
         removed_statistics,
         merged_statistics,
@@ -150,7 +150,7 @@ fn test_resolvable_delete_conflict() {
         Some(Arc::new(latest_snapshot)),
     );
     let snapshot = result.unwrap();
-    let expected = vec![("8".to_string(), 1), ("4".to_string(), 1)];
+    let expected = vec![("8".to_string(), 1).into(), ("4".to_string(), 1).into()];
     assert_eq!(snapshot.segments, expected);
 
     let actual = snapshot.summary;
@@ -180,9 +180,9 @@ fn test_resolvable_delete_conflict() {
 fn test_resolvable_replace_conflict() {
     let mut base_snapshot = TableSnapshot::new_empty_snapshot(TableSchema::default());
     base_snapshot.segments = vec![
-        ("1".to_string(), 1),
-        ("2".to_string(), 1),
-        ("3".to_string(), 1),
+        ("1".to_string(), 1).into(),
+        ("2".to_string(), 1).into(),
+        ("3".to_string(), 1).into(),
     ];
 
     base_snapshot.summary = Statistics {
@@ -198,9 +198,9 @@ fn test_resolvable_replace_conflict() {
 
     let mut latest_snapshot = TableSnapshot::new_empty_snapshot(TableSchema::default());
     latest_snapshot.segments = vec![
-        ("2".to_string(), 1),
-        ("3".to_string(), 1),
-        ("4".to_string(), 1),
+        ("2".to_string(), 1).into(),
+        ("3".to_string(), 1).into(),
+        ("4".to_string(), 1).into(),
     ];
 
     latest_snapshot.summary = Statistics {
@@ -237,8 +237,8 @@ fn test_resolvable_replace_conflict() {
     };
 
     let ctx = ConflictResolveContext::ModifiedSegmentExistsInLatest(SnapshotChanges {
-        appended_segments: vec![("6".to_string(), 1)],
-        replaced_segments: HashMap::from([(2, ("5".to_string(), 1))]),
+        appended_segments: vec![("6".to_string(), 1).into()],
+        replaced_segments: HashMap::from([(2, ("5".to_string(), 1).into())]),
         removed_segment_indexes: vec![1],
         removed_statistics,
         merged_statistics,
@@ -254,9 +254,9 @@ fn test_resolvable_replace_conflict() {
     );
     let snapshot = result.unwrap();
     let expected = vec![
-        ("6".to_string(), 1),
-        ("5".to_string(), 1),
-        ("4".to_string(), 1),
+        ("6".to_string(), 1).into(), // TODO
+        ("5".to_string(), 1).into(),
+        ("4".to_string(), 1).into(),
     ];
     assert_eq!(snapshot.segments, expected);
 

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

@@ -98,9 +98,9 @@ async fn check_segment_column_ids(
             fuse_table.get_operator(),
             TestFixture::default_table_schema(),
         );
-        for (seg_loc, _) in &snapshot.segments {
+        for seg in &snapshot.segments {
             let params = LoadParams {
-                location: seg_loc.clone(),
+                location: seg.location.0.clone(),
                 len_hint: None,
                 ver: SegmentInfo::VERSION,
                 put_cache: false,

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

@@ -251,7 +251,7 @@ async fn test_commit_to_meta_server() -> Result<()> {
             let table = fixture.latest_default_table().await?;
             let fuse_table = FuseTable::try_from_table(table.as_ref())?;
 
-            let new_segments = vec![("do not care".to_string(), SegmentInfo::VERSION)];
+            let new_segments = vec![("do not care".to_string(), SegmentInfo::VERSION).into()];
             let new_snapshot = TableSnapshot::new(
                 Uuid::new_v4(),
                 &None,

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

@@ -103,14 +103,14 @@ async fn check_partitions(parts: &Partitions, fixture: &TestFixture) -> Result<(
 
     let snapshot = reader.read(&load_params).await?;
     for segment in &snapshot.segments {
-        segment_name.insert(segment.0.clone());
+        segment_name.insert(segment.location.0.clone());
 
         let compact_segment_reader = MetaReaders::segment_info_reader(
             fuse_table.get_operator(),
             TestFixture::default_table_schema(),
         );
         let params = LoadParams {
-            location: segment.0.clone(),
+            location: segment.location.0.clone(),
             len_hint: None,
             ver: SegmentInfo::VERSION,
             put_cache: false,