replace RoaringBitmap with plain old HashSet

I think we could use RoaringBimtmap to maintain the set of
snapshots that references a segment, but not sure about which crate
to use, and if they meet the performance assumptions and implemented correctly.

let's re-enable RoaringBitmap later(after evaluations).

Author: dantengsky

Cargo.lock

@@ -42,7 +42,6 @@ futures-util = "0.3.24"
 itertools = "0.10.5"
 metrics = "0.20.1"
 opendal = "0.22"
-roaring = "0.10.1"
 serde = { workspace = true }
 serde_json = { workspace = true }
 tracing = "0.1.36"

src/query/storages/fuse/fuse/Cargo.toml

@@ -31,7 +31,6 @@ pub use read::TableSnapshotReader;
 pub use segments::try_join_futures;
 pub use segments::SegmentsIO;
 pub use snapshots::ListSnapshotLiteOption;
-pub use snapshots::PositionTagged;
 pub use snapshots::SnapshotLiteListExtended;
 pub use snapshots::SnapshotsIO;
 pub use write::write_block;

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

@@ -26,14 +26,14 @@ use common_catalog::table_context::TableContext;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_storages_table_meta::meta::Location;
+use common_storages_table_meta::meta::SnapshotId;
 use common_storages_table_meta::meta::TableSnapshot;
 use common_storages_table_meta::meta::TableSnapshotLite;
 use futures::stream::StreamExt;
 use futures_util::future;
 use futures_util::TryStreamExt;
 use opendal::ObjectMode;
 use opendal::Operator;
-use roaring::RoaringBitmap;
 use tracing::info;
 use tracing::warn;
 use tracing::Instrument;
@@ -49,11 +49,10 @@ pub struct SnapshotsIO {
     format_version: u64,
 }
 
-pub type PositionTagged<T> = (T, usize);
 pub struct SnapshotLiteListExtended {
     pub chained_snapshot_lites: Vec<TableSnapshotLite>,
-    pub segment_locations: HashMap<Location, RoaringBitmap>,
-    pub orphan_snapshot_lites: Vec<PositionTagged<TableSnapshotLite>>,
+    pub segment_locations: HashMap<Location, HashSet<SnapshotId>>,
+    pub orphan_snapshot_lites: Vec<TableSnapshotLite>,
 }
 
 pub enum ListSnapshotLiteOption<'a> {
@@ -173,7 +172,8 @@ impl SnapshotsIO {
         // List all the snapshot file paths
         // note that snapshot file paths of ongoing txs might be included
         let mut snapshot_files = vec![];
-        let mut segment_location_with_index: HashMap<Location, RoaringBitmap> = HashMap::new();
+        let mut segment_location_with_index: HashMap<Location, HashSet<SnapshotId>> =
+            HashMap::new();
         if let Some(prefix) = Self::get_s3_prefix_from_file(&root_snapshot_file) {
             snapshot_files = self.list_files(&prefix, limit, None).await?;
         }
@@ -198,11 +198,9 @@ impl SnapshotsIO {
                     continue;
                 }
                 let snapshot_lite = TableSnapshotLite::from(snapshot.as_ref());
+                let snapshot_id = snapshot_lite.snapshot_id;
                 snapshot_lites.push(snapshot_lite);
 
-                // since we use u32 RoaringBitmap to index the snapshots
-                // just in case, here we check if number of snapshot is within upper bound
-                let idx = u32::try_from(snapshot_lites.len() - 1)?;
                 if let ListSnapshotLiteOption::NeedSegmentsWithExclusion(filter) = list_options {
                     // collects segments, and the snapshots that reference them.
                     for segment_location in &snapshot.segments {
@@ -214,9 +212,9 @@ impl SnapshotsIO {
                         segment_location_with_index
                             .entry(segment_location.clone())
                             .and_modify(|v| {
-                                v.insert(idx);
+                                v.insert(snapshot_id);
                             })
-                            .or_insert_with(|| RoaringBitmap::from_iter(vec![idx]));
+                            .or_insert_with(|| HashSet::from_iter(vec![snapshot_id]));
                     }
                 }
             }
@@ -255,11 +253,11 @@ impl SnapshotsIO {
     fn chain_snapshots(
         snapshot_lites: Vec<TableSnapshotLite>,
         root_snapshot: &TableSnapshot,
-    ) -> (Vec<TableSnapshotLite>, Vec<(TableSnapshotLite, usize)>) {
+    ) -> (Vec<TableSnapshotLite>, Vec<TableSnapshotLite>) {
         let mut snapshot_map = HashMap::new();
         let mut chained_snapshot_lites = vec![];
-        for (idx, snapshot_lite) in snapshot_lites.into_iter().enumerate() {
-            snapshot_map.insert(snapshot_lite.snapshot_id, (snapshot_lite, idx));
+        for snapshot_lite in snapshot_lites.into_iter() {
+            snapshot_map.insert(snapshot_lite.snapshot_id, snapshot_lite);
         }
         let root_snapshot_lite = TableSnapshotLite::from(root_snapshot);
         let mut prev_snapshot_id_tuple = root_snapshot_lite.prev_snapshot_id;
@@ -270,7 +268,7 @@ impl SnapshotsIO {
                 None => {
                     break;
                 }
-                Some((prev_snapshot, _idx)) => {
+                Some(prev_snapshot) => {
                     prev_snapshot_id_tuple = prev_snapshot.prev_snapshot_id;
                     chained_snapshot_lites.push(prev_snapshot);
                 }

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

@@ -28,13 +28,11 @@ use common_storages_table_meta::caches::CacheManager;
 use common_storages_table_meta::meta::Location;
 use common_storages_table_meta::meta::SnapshotId;
 use common_storages_table_meta::meta::TableSnapshotLite;
-use roaring::RoaringBitmap;
 use tracing::info;
 use tracing::warn;
 
 use crate::io::Files;
 use crate::io::ListSnapshotLiteOption;
-use crate::io::PositionTagged;
 use crate::io::SegmentsIO;
 use crate::io::SnapshotsIO;
 use crate::FuseTable;
@@ -131,16 +129,13 @@ impl FuseTable {
                 partitioned_snapshots
                     .within_retention
                     .into_iter()
-                    .map(|(_, index)| index as u32),
+                    .map(|snapshot| snapshot.snapshot_id)
+                    .collect(),
                 snapshot_lites_extended.segment_locations,
             );
 
             // orphan_snapshots that beyond retention period are allowed to be collected
-            orphan_snapshots = partitioned_snapshots
-                .beyond_retention
-                .into_iter()
-                .map(|(v, _)| v)
-                .collect();
+            orphan_snapshots = partitioned_snapshots.beyond_retention;
 
             // FIXME: we do not need to write last snapshot hint here(since last snapshot never changed
             // during gc). introduce a dedicated stmt to refresh the hint file instead pls.
@@ -360,14 +355,14 @@ impl FuseTable {
     fn apply_retention_rule(
         ctx: &dyn TableContext,
         base_timestamp: Option<DateTime<Utc>>,
-        snapshot_lites: Vec<(TableSnapshotLite, usize)>,
+        snapshot_lites: Vec<TableSnapshotLite>,
     ) -> Result<RetentionPartition> {
         //    let retention_interval = Duration::hours(DEFAULT_RETENTION_PERIOD_HOURS as i64);
         let retention_interval = Duration::hours(ctx.get_settings().get_retention_period()? as i64);
         let retention_point = base_timestamp.map(|s| s - retention_interval);
         let (beyond_retention, within_retention) = snapshot_lites
             .into_iter()
-            .partition(|(lite, _idx)| lite.timestamp < retention_point);
+            .partition(|lite| lite.timestamp < retention_point);
         Ok(RetentionPartition {
             beyond_retention,
             within_retention,
@@ -377,12 +372,15 @@ impl FuseTable {
     // filter out segments that are referenced by orphan snapshots
     // which are within retention period
     fn filter_out_segments_within_retention(
-        orphan_snapshot_index: impl IntoIterator<Item = u32>,
-        mut segment_with_refer_index: HashMap<Location, RoaringBitmap>,
+        // orphan_snapshot_index: impl IntoIterator<Item = SnapshotId>,
+        orphan_snapshot_index: HashSet<SnapshotId>,
+        mut segment_with_refer_index: HashMap<Location, HashSet<SnapshotId>>,
     ) -> HashSet<Location> {
-        let orphan_snapshot_index_bitmap = RoaringBitmap::from_iter(orphan_snapshot_index);
+        // let orphan_snapshot_index_bitmap = RoaringBitmap::from_iter(orphan_snapshot_index);
+        // segment_with_refer_index
+        //    .retain(|_location, refer_map| orphan_snapshot_index_bitmap.is_disjoint(refer_map));
         segment_with_refer_index
-            .retain(|_location, refer_map| orphan_snapshot_index_bitmap.is_disjoint(refer_map));
+            .retain(|_location, refer_map| orphan_snapshot_index.is_disjoint(refer_map));
         segment_with_refer_index.into_keys().collect()
     }
 
@@ -452,6 +450,6 @@ impl FuseTable {
 }
 
 struct RetentionPartition {
-    beyond_retention: Vec<PositionTagged<TableSnapshotLite>>,
-    within_retention: Vec<PositionTagged<TableSnapshotLite>>,
+    beyond_retention: Vec<TableSnapshotLite>,
+    within_retention: Vec<TableSnapshotLite>,
 }