feat(storage): store compaction group id in per table info

proto/hummock.proto

@@ -69,25 +69,28 @@ message IntraLevelDelta {
 enum CompatibilityVersion {
   VERSION_UNSPECIFIED = 0;
   NO_TRIVIAL_SPLIT = 1;
+  NO_MEMBER_TABLE_IDS = 2;
 }
 
 message GroupConstruct {
   CompactionConfig group_config = 1;
   // If parent_group_id is not 0, it means parent_group_id splits into parent_group_id and this group, so this group is not empty initially.
   uint64 parent_group_id = 2;
-  repeated uint32 table_ids = 3;
+  repeated uint32 table_ids = 3 [deprecated = true];
   uint64 group_id = 4;
   uint64 new_sst_start_id = 5;
   CompatibilityVersion version = 6;
 }
 
 message GroupMetaChange {
-  repeated uint32 table_ids_add = 1;
-  repeated uint32 table_ids_remove = 2;
+  option deprecated = true;
+  repeated uint32 table_ids_add = 1 [deprecated = true];
+  repeated uint32 table_ids_remove = 2 [deprecated = true];
 }
 
 message GroupTableChange {
-  repeated uint32 table_ids = 1;
+  option deprecated = true;
+  repeated uint32 table_ids = 1 [deprecated = true];
   uint64 target_group_id = 2;
   uint64 origin_group_id = 3;
   uint64 new_sst_start_id = 4;
@@ -102,7 +105,7 @@ message GroupDelta {
     GroupConstruct group_construct = 2;
     GroupDestroy group_destroy = 3;
     GroupMetaChange group_meta_change = 4;
-    GroupTableChange group_table_change = 5;
+    GroupTableChange group_table_change = 5 [deprecated = true];
   }
 }
 
@@ -152,11 +155,13 @@ message TableChangeLog {
 message StateTableInfo {
   uint64 committed_epoch = 1;
   uint64 safe_epoch = 2;
+  uint64 compaction_group_id = 3;
 }
 
 message StateTableInfoDelta {
   uint64 committed_epoch = 1;
   uint64 safe_epoch = 2;
+  uint64 compaction_group_id = 3;
 }
 
 message HummockVersion {
@@ -165,7 +170,7 @@ message HummockVersion {
     OverlappingLevel l0 = 2;
     uint64 group_id = 3;
     uint64 parent_group_id = 4;
-    repeated uint32 member_table_ids = 5;
+    repeated uint32 member_table_ids = 5 [deprecated = true];
   }
   uint64 id = 1;
   // Levels of each compaction group

src/meta/src/hummock/compaction/mod.rs

@@ -16,12 +16,12 @@
 
 pub mod compaction_config;
 mod overlap_strategy;
-use risingwave_common::catalog::TableOption;
+use risingwave_common::catalog::{TableId, TableOption};
 use risingwave_pb::hummock::compact_task::{self, TaskType};
 
 mod picker;
 pub mod selector;
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeSet, HashMap, HashSet};
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
@@ -92,6 +92,7 @@ impl CompactStatus {
     pub fn get_compact_task(
         &mut self,
         levels: &Levels,
+        member_table_ids: &BTreeSet<TableId>,
         task_id: HummockCompactionTaskId,
         group: &CompactionGroup,
         stats: &mut LocalSelectorStatistic,
@@ -106,6 +107,7 @@ impl CompactStatus {
             task_id,
             group,
             levels,
+            member_table_ids,
             &mut self.level_handlers,
             stats,
             table_id_to_options.clone(),
@@ -121,6 +123,7 @@ impl CompactStatus {
                     task_id,
                     group,
                     levels,
+                    member_table_ids,
                     &mut self.level_handlers,
                     stats,
                     table_id_to_options,

src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs

@@ -326,7 +326,6 @@ pub mod tests {
                     generate_table(1, 1, 201, 210, 1),
                 ],
             )],
-            member_table_ids: vec![1],
             ..Default::default()
         };
         let mut local_stats = LocalPickerStatistic::default();
@@ -419,7 +418,6 @@ pub mod tests {
                 total_file_size: 0,
                 uncompressed_file_size: 0,
             }),
-            member_table_ids: vec![1],
             ..Default::default()
         };
         push_tables_level0_nonoverlapping(&mut levels, vec![generate_table(1, 1, 50, 140, 2)]);
@@ -482,7 +480,6 @@ pub mod tests {
                 total_file_size: 0,
                 uncompressed_file_size: 0,
             }),
-            member_table_ids: vec![1],
             ..Default::default()
         };
         push_tables_level0_nonoverlapping(
@@ -586,7 +583,6 @@ pub mod tests {
         let levels = Levels {
             l0: Some(l0),
             levels: vec![generate_level(1, vec![generate_table(3, 1, 0, 100000, 1)])],
-            member_table_ids: vec![1],
             ..Default::default()
         };
         let levels_handler = vec![LevelHandler::new(0), LevelHandler::new(1)];
@@ -664,7 +660,6 @@ pub mod tests {
         let levels = Levels {
             l0: Some(l0),
             levels: vec![generate_level(1, vec![generate_table(3, 1, 0, 100000, 1)])],
-            member_table_ids: vec![1],
             ..Default::default()
         };
         let mut levels_handler = vec![LevelHandler::new(0), LevelHandler::new(1)];
@@ -731,7 +726,6 @@ pub mod tests {
         let levels = Levels {
             l0: Some(l0),
             levels: vec![generate_level(1, vec![generate_table(3, 1, 1, 100, 1)])],
-            member_table_ids: vec![1],
             ..Default::default()
         };
         let mut levels_handler = vec![LevelHandler::new(0), LevelHandler::new(1)];

src/meta/src/hummock/compaction/picker/intra_compaction_picker.rs

@@ -428,7 +428,6 @@ pub mod tests {
                 total_file_size: 0,
                 uncompressed_file_size: 0,
             }),
-            member_table_ids: vec![1],
             ..Default::default()
         };
         push_tables_level0_nonoverlapping(
@@ -473,7 +472,6 @@ pub mod tests {
                 generate_table(1, 1, 100, 210, 2),
                 generate_table(2, 1, 200, 250, 2),
             ])),
-            member_table_ids: vec![1],
             ..Default::default()
         };
         let mut levels_handler = vec![LevelHandler::new(0), LevelHandler::new(1)];
@@ -511,7 +509,6 @@ pub mod tests {
             let levels = Levels {
                 l0: Some(l0),
                 levels: vec![generate_level(1, vec![generate_table(100, 1, 0, 1000, 1)])],
-                member_table_ids: vec![1],
                 ..Default::default()
             };
             let mut levels_handler = vec![LevelHandler::new(0), LevelHandler::new(1)];
@@ -560,7 +557,6 @@ pub mod tests {
             let levels = Levels {
                 l0: Some(l0),
                 levels: vec![generate_level(1, vec![generate_table(100, 1, 0, 1000, 1)])],
-                member_table_ids: vec![1],
                 ..Default::default()
             };
             let mut levels_handler = vec![LevelHandler::new(0), LevelHandler::new(1)];
@@ -632,7 +628,6 @@ pub mod tests {
             let levels = Levels {
                 l0: Some(l0),
                 levels: vec![generate_level(1, vec![generate_table(100, 1, 0, 1000, 1)])],
-                member_table_ids: vec![1],
                 ..Default::default()
             };
             let mut levels_handler = vec![LevelHandler::new(0), LevelHandler::new(1)];
@@ -708,7 +703,6 @@ pub mod tests {
         let levels = Levels {
             l0: Some(l0),
             levels: vec![generate_level(1, vec![generate_table(100, 1, 0, 1000, 1)])],
-            member_table_ids: vec![1],
             ..Default::default()
         };
         levels_handler[1].add_pending_task(100, 1, levels.levels[0].get_table_infos());
@@ -729,7 +723,6 @@ pub mod tests {
         let mut levels = Levels {
             l0: Some(l0),
             levels: vec![generate_level(1, vec![generate_table(100, 1, 0, 1000, 1)])],
-            member_table_ids: vec![1],
             ..Default::default()
         };
         assert!(picker
@@ -827,7 +820,6 @@ pub mod tests {
         let levels = Levels {
             l0: Some(l0),
             levels: vec![generate_level(1, vec![generate_table(100, 1, 0, 1000, 1)])],
-            member_table_ids: vec![1],
             ..Default::default()
         };
 

src/meta/src/hummock/compaction/picker/manual_compaction_picker.rs

@@ -326,7 +326,7 @@ impl CompactionPicker for ManualCompactionPicker {
 
 #[cfg(test)]
 pub mod tests {
-    use std::collections::HashMap;
+    use std::collections::{BTreeSet, HashMap};
 
     use risingwave_pb::hummock::compact_task;
     pub use risingwave_pb::hummock::KeyRange;
@@ -1198,6 +1198,7 @@ pub mod tests {
                     1,
                     &group_config,
                     &levels,
+                    &BTreeSet::new(),
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),
@@ -1235,6 +1236,7 @@ pub mod tests {
                     2,
                     &group_config,
                     &levels,
+                    &BTreeSet::new(),
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),
@@ -1308,6 +1310,7 @@ pub mod tests {
                     1,
                     &group_config,
                     &levels,
+                    &BTreeSet::new(),
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),
@@ -1347,6 +1350,7 @@ pub mod tests {
                     1,
                     &group_config,
                     &levels,
+                    &BTreeSet::new(),
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),

src/meta/src/hummock/compaction/picker/space_reclaim_compaction_picker.rs

@@ -168,10 +168,11 @@ impl SpaceReclaimCompactionPicker {
 #[cfg(test)]
 mod test {
 
-    use std::collections::HashMap;
+    use std::collections::{BTreeSet, HashMap};
     use std::sync::Arc;
 
     use itertools::Itertools;
+    use risingwave_common::catalog::TableId;
     use risingwave_pb::hummock::compact_task;
     pub use risingwave_pb::hummock::{Level, LevelType};
 
@@ -234,11 +235,12 @@ mod test {
         }
 
         assert_eq!(levels.len(), 4);
-        let mut levels = Levels {
+        let levels = Levels {
             levels,
             l0: Some(l0),
             ..Default::default()
         };
+        let mut member_table_ids = BTreeSet::new();
         let mut levels_handler = (0..5).map(LevelHandler::new).collect_vec();
         let mut local_stats = LocalSelectorStatistic::default();
 
@@ -252,6 +254,7 @@ mod test {
                     1,
                     &group_config,
                     &levels,
+                    &member_table_ids,
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),
@@ -268,6 +271,7 @@ mod test {
                     1,
                     &group_config,
                     &levels,
+                    &member_table_ids,
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),
@@ -308,6 +312,7 @@ mod test {
                     1,
                     &group_config,
                     &levels,
+                    &member_table_ids,
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),
@@ -334,6 +339,7 @@ mod test {
                     1,
                     &group_config,
                     &levels,
+                    &member_table_ids,
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),
@@ -351,12 +357,17 @@ mod test {
                 }
             }
 
-            levels.member_table_ids = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
+            member_table_ids = BTreeSet::from_iter(
+                [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+                    .into_iter()
+                    .map(TableId::new),
+            );
             // pick space reclaim
             let task = selector.pick_compaction(
                 1,
                 &group_config,
                 &levels,
+                &member_table_ids,
                 &mut levels_handler,
                 &mut local_stats,
                 HashMap::default(),
@@ -372,13 +383,15 @@ mod test {
                 }
             }
 
-            levels.member_table_ids = vec![2, 3, 4, 5, 6, 7, 8, 9];
+            member_table_ids =
+                BTreeSet::from_iter([2, 3, 4, 5, 6, 7, 8, 9].into_iter().map(TableId::new));
             // pick space reclaim
             let task = selector
                 .pick_compaction(
                     1,
                     &group_config,
                     &levels,
+                    &member_table_ids,
                     &mut levels_handler,
                     &mut local_stats,
                     HashMap::default(),
@@ -407,7 +420,7 @@ mod test {
             // rebuild selector
             selector = SpaceReclaimCompactionSelector::default();
             // cut range [3,4] [6] [8,9,10]
-            levels.member_table_ids = vec![0, 1, 2, 5, 7];
+            member_table_ids = BTreeSet::from_iter([0, 1, 2, 5, 7].into_iter().map(TableId::new));
             let expect_task_file_count = [2, 1, 4];
             let expect_task_sst_id_range = [vec![3, 4], vec![6], vec![8, 9, 10, 11]];
             for (index, x) in expect_task_file_count.iter().enumerate() {
@@ -417,6 +430,7 @@ mod test {
                         1,
                         &group_config,
                         &levels,
+                        &member_table_ids,
                         &mut levels_handler,
                         &mut local_stats,
                         HashMap::default(),
@@ -458,12 +472,13 @@ mod test {
             // rebuild selector
             selector = SpaceReclaimCompactionSelector::default();
             // cut range [3,4] [6] [8,9,10]
-            levels.member_table_ids = vec![0, 1, 2, 5, 7];
+
+            member_table_ids = BTreeSet::from_iter([0, 1, 2, 5, 7].into_iter().map(TableId::new));
             let expect_task_file_count = [2, 1, 5];
             let expect_task_sst_id_range = [vec![3, 4], vec![6], vec![7, 8, 9, 10, 11]];
             for (index, x) in expect_task_file_count.iter().enumerate() {
                 if index == expect_task_file_count.len() - 1 {
-                    levels.member_table_ids = vec![2, 5];
+                    member_table_ids = BTreeSet::from_iter([2, 5].into_iter().map(TableId::new));
                 }
 
                 // // pick space reclaim
@@ -472,6 +487,7 @@ mod test {
                         1,
                         &group_config,
                         &levels,
+                        &member_table_ids,
                         &mut levels_handler,
                         &mut local_stats,
                         HashMap::default(),