wip

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/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/storages/common/table_meta/src/meta/current/mod.rs

@@ -20,7 +20,9 @@ pub use v2::ColumnMeta;
 pub use v2::ColumnStatistics;
 pub use v2::Statistics;
 pub use v4::CompactSegmentInfo;
+pub use v4::SegmentDescriptor;
 pub use v4::SegmentInfo;
+pub use v4::SegmentSummary;
 pub use v4::TableSnapshot;
 pub use v4::TableSnapshotLite;
 

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

@@ -17,5 +17,7 @@ mod snapshot;
 
 pub use segment::CompactSegmentInfo;
 pub use segment::SegmentInfo;
+pub use snapshot::SegmentDescriptor;
+pub use snapshot::SegmentSummary;
 pub use snapshot::TableSnapshot;
 pub use snapshot::TableSnapshotLite;

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

@@ -78,21 +78,42 @@ pub struct TableSnapshot {
     ///
     /// We rely on background merge tasks to keep merging segments, so that
     /// this the size of this vector could be kept reasonable
-    pub segments: Vec<Location>,
+    pub segments: Vec<SegmentDescriptor>,
 
     // The metadata of the cluster keys.
     pub cluster_key_meta: Option<ClusterKey>,
     pub table_statistics_location: Option<String>,
 }
 
+// TODO remove PartialEq
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
+pub struct SegmentSummary {
+    pub row_count: u64,
+    pub block_count: u64,
+}
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
+pub struct SegmentDescriptor {
+    pub location: Location,
+    pub summary: Option<SegmentSummary>,
+}
+
+impl From<Location> for SegmentDescriptor {
+    fn from(location: Location) -> Self {
+        SegmentDescriptor {
+            location,
+            summary: None,
+        }
+    }
+}
+
 impl TableSnapshot {
     pub fn new(
         snapshot_id: SnapshotId,
         prev_timestamp: &Option<DateTime<Utc>>,
         prev_snapshot_id: Option<(SnapshotId, FormatVersion)>,
         schema: TableSchema,
         summary: Statistics,
-        segments: Vec<Location>,
+        segments: Vec<SegmentDescriptor>,
         cluster_key_meta: Option<ClusterKey>,
         table_statistics_location: Option<String>,
     ) -> Self {
@@ -218,7 +239,7 @@ impl From<v2::TableSnapshot> for TableSnapshot {
             prev_snapshot_id: s.prev_snapshot_id,
             schema: s.schema,
             summary: s.summary,
-            segments: s.segments,
+            segments: s.segments.into_iter().map(|v| v.into()).collect(),
             cluster_key_meta: s.cluster_key_meta,
             table_statistics_location: s.table_statistics_location,
         }
@@ -239,7 +260,7 @@ where T: Into<v3::TableSnapshot>
             prev_snapshot_id: s.prev_snapshot_id,
             schema: s.schema.into(),
             summary: s.summary.into(),
-            segments: s.segments,
+            segments: s.segments.into_iter().map(|v| v.into()).collect(),
             cluster_key_meta: s.cluster_key_meta,
             table_statistics_location: s.table_statistics_location,
         }

src/query/storages/fuse/src/io/snapshots.rs

@@ -267,10 +267,10 @@ impl SnapshotsIO {
         let mut segments = HashSet::new();
         // collects extended segments.
         for segment_location in &snapshot.segments {
-            if root_snapshot.segments.contains(segment_location) {
+            if root_snapshot.segments.contains(&segment_location.location) {
                 continue;
             }
-            segments.insert(segment_location.clone());
+            segments.insert(segment_location.location.clone());
         }
         let table_statistics_location =
             if snapshot.table_statistics_location != root_snapshot.table_statistics_location {

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

@@ -72,8 +72,9 @@ impl FuseTable {
                     blocks_cluster_stats.push(cluster_stats.clone());
                 }
 
+                let locations = chunk.iter().map(|v| v.location.clone()).collect::<Vec<_>>();
                 let segments = segments_io
-                    .read_segments::<SegmentInfo>(chunk, true)
+                    .read_segments::<SegmentInfo>(&locations, true)
                     .await?;
                 for segment in segments {
                     let segment = segment?;

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

@@ -37,7 +37,7 @@ use databend_common_pipeline_transforms::processors::AsyncAccumulatingTransforme
 use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_storages_common_cache::CacheAccessor;
 use databend_storages_common_cache_manager::CachedObject;
-use databend_storages_common_table_meta::meta::Location;
+use databend_storages_common_table_meta::meta::SegmentDescriptor;
 use databend_storages_common_table_meta::meta::SegmentInfo;
 use databend_storages_common_table_meta::meta::SnapshotId;
 use databend_storages_common_table_meta::meta::Statistics;
@@ -287,7 +287,7 @@ impl FuseTable {
         &self,
         ctx: &Arc<dyn TableContext>,
         base_snapshot: Arc<TableSnapshot>,
-        base_segments: &[Location],
+        base_segments: &[SegmentDescriptor],
         base_summary: Statistics,
         abort_operation: AbortOperation,
         max_retry_elapsed: Option<Duration>,
@@ -303,7 +303,7 @@ impl FuseTable {
         let mut latest_table_ref: Arc<dyn Table>;
 
         // potentially concurrently appended segments, init it to empty
-        let mut concurrently_appended_segment_locations: &[Location] = &[];
+        let mut concurrently_appended_segment_locations: &[SegmentDescriptor] = &[];
 
         // Status
         ctx.set_status_info("mutation: begin try to commit");
@@ -422,12 +422,12 @@ impl FuseTable {
     async fn merge_with_base(
         ctx: Arc<dyn TableContext>,
         operator: Operator,
-        base_segments: &[Location],
+        base_segments: &[SegmentDescriptor],
         base_summary: &Statistics,
-        concurrently_appended_segment_locations: &[Location],
+        concurrently_appended_segment_locations: &[SegmentDescriptor],
         schema: TableSchemaRef,
         default_cluster_key_id: Option<u32>,
-    ) -> Result<(Vec<Location>, Statistics)> {
+    ) -> Result<(Vec<SegmentDescriptor>, Statistics)> {
         if concurrently_appended_segment_locations.is_empty() {
             Ok((base_segments.to_owned(), base_summary.clone()))
         } else {
@@ -439,8 +439,13 @@ impl FuseTable {
                 .collect();
 
             let fuse_segment_io = SegmentsIO::create(ctx, operator, schema);
+            // TODO
+            let locations = concurrently_appended_segment_locations
+                .iter()
+                .map(|v| v.location.clone())
+                .collect::<Vec<_>>();
             let concurrent_appended_segment_infos = fuse_segment_io
-                .read_segments::<SegmentInfo>(concurrently_appended_segment_locations, true)
+                .read_segments::<SegmentInfo>(&locations, true)
                 .await?;
 
             let mut new_statistics = base_summary.clone();

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

@@ -29,8 +29,9 @@ use databend_common_expression::TableSchemaRef;
 use databend_common_pipeline_transforms::processors::AsyncAccumulatingTransform;
 use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_storages_common_table_meta::meta::BlockMeta;
-use databend_storages_common_table_meta::meta::Location;
+use databend_storages_common_table_meta::meta::SegmentDescriptor;
 use databend_storages_common_table_meta::meta::SegmentInfo;
+use databend_storages_common_table_meta::meta::SegmentSummary;
 use databend_storages_common_table_meta::meta::Statistics;
 use databend_storages_common_table_meta::meta::Versioned;
 use itertools::Itertools;
@@ -62,10 +63,10 @@ pub struct TableMutationAggregator {
     location_gen: TableMetaLocationGenerator,
     thresholds: BlockThresholds,
     default_cluster_key_id: Option<u32>,
-    base_segments: Vec<Location>,
+    base_segments: Vec<SegmentDescriptor>,
 
     mutations: HashMap<SegmentIndex, BlockMutations>,
-    appended_segments: Vec<Location>,
+    appended_segments: Vec<SegmentDescriptor>,
     appended_statistics: Statistics,
     removed_segment_indexes: Vec<SegmentIndex>,
     removed_statistics: Statistics,
@@ -105,7 +106,7 @@ impl TableMutationAggregator {
     pub fn new(
         table: &FuseTable,
         ctx: Arc<dyn TableContext>,
-        base_segments: Vec<Location>,
+        base_segments: Vec<SegmentDescriptor>,
         kind: MutationKind,
     ) -> Self {
         TableMutationAggregator {
@@ -184,8 +185,16 @@ impl TableMutationAggregator {
                     self.default_cluster_key_id,
                 );
 
+                let desc = SegmentDescriptor {
+                    location: (segment_location, format_version),
+                    summary: Some(SegmentSummary {
+                        row_count: summary.row_count,
+                        block_count: summary.block_count,
+                    }),
+                };
                 self.appended_segments
-                    .push((segment_location, format_version))
+                    //.push((segment_location, format_version))
+                    .push(desc)
             }
             MutationLogEntry::CompactExtras { extras } => {
                 match self.mutations.entry(extras.segment_index) {
@@ -241,8 +250,14 @@ impl TableMutationAggregator {
                                 &summary,
                                 self.default_cluster_key_id,
                             );
-                            replaced_segments
-                                .insert(result.index, (location, SegmentInfo::VERSION));
+                            let desc = SegmentDescriptor {
+                                location: (location, SegmentInfo::VERSION),
+                                summary: Some(SegmentSummary {
+                                    row_count: summary.row_count,
+                                    block_count: summary.block_count,
+                                }),
+                            };
+                            replaced_segments.insert(result.index, desc);
                         } else {
                             self.removed_segment_indexes.push(result.index);
                         }
@@ -311,7 +326,7 @@ impl TableMutationAggregator {
                 let (new_blocks, origin_summary) = if let Some(loc) = location {
                     // read the old segment
                     let compact_segment_info =
-                        SegmentsIO::read_compact_segment(op.clone(), loc, schema, false).await?;
+                        SegmentsIO::read_compact_segment(op.clone(), loc.location, schema, false).await?;
                     let mut segment_info = SegmentInfo::try_from(compact_segment_info)?;
 
                     // take away the blocks, they are being mutated