feat(storage): optimize data alignment for default compaction group

proto/hummock.proto

@@ -310,6 +310,7 @@ message CompactTask {
   bool split_by_state_table = 21;
   // Compaction needs to cut the state table every time 1/weight of vnodes in the table have been processed.
   uint32 split_weight_by_vnode = 22;
+  map<uint32, uint32> table_vnode_partition = 23;
 }
 
 message LevelHandler {

src/common/src/config.rs

@@ -262,6 +262,9 @@ pub struct MetaConfig {
     #[serde(default = "default::meta::split_group_size_limit")]
     pub split_group_size_limit: u64,
 
+    #[serde(default = "default::meta::cut_table_size_limit")]
+    pub cut_table_size_limit: u64,
+
     #[serde(default, flatten)]
     pub unrecognized: Unrecognized<Self>,
 
@@ -937,7 +940,7 @@ pub mod default {
         }
 
         pub fn periodic_split_compact_group_interval_sec() -> u64 {
-            180 // 3mi
+            10 // 10s
         }
 
         pub fn periodic_tombstone_reclaim_compaction_interval_sec() -> u64 {
@@ -967,6 +970,10 @@ pub mod default {
         pub fn compaction_task_max_heartbeat_interval_secs() -> u64 {
             60 // 1min
         }
+
+        pub fn cut_table_size_limit() -> u64 {
+            1024 * 1024 * 1024 // 1GB
+        }
     }
 
     pub mod server {

src/config/example.toml

@@ -33,9 +33,10 @@ backend = "Mem"
 periodic_space_reclaim_compaction_interval_sec = 3600
 periodic_ttl_reclaim_compaction_interval_sec = 1800
 periodic_tombstone_reclaim_compaction_interval_sec = 600
-periodic_split_compact_group_interval_sec = 180
+periodic_split_compact_group_interval_sec = 10
 move_table_size_limit = 10737418240
 split_group_size_limit = 68719476736
+cut_table_size_limit = 1073741824
 do_not_config_object_storage_lifecycle = false
 partition_vnode_count = 64
 table_write_throughput_threshold = 16777216

src/meta/node/src/lib.rs

@@ -300,6 +300,7 @@ pub fn start(opts: MetaNodeOpts) -> Pin<Box<dyn Future<Output = ()> + Send>> {
                     .meta
                     .compaction_task_max_heartbeat_interval_secs,
                 compaction_config: Some(config.meta.compaction_config),
+                cut_table_size_limit: config.meta.cut_table_size_limit,
             },
             config.system.into_init_system_params(),
         )

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

@@ -77,7 +77,6 @@ pub struct CompactionTask {
     pub compression_algorithm: String,
     pub target_file_size: u64,
     pub compaction_task_type: compact_task::TaskType,
-    pub enable_split_by_table: bool,
 }
 
 pub fn create_overlap_strategy(compaction_mode: CompactionMode) -> Arc<dyn OverlapStrategy> {
@@ -150,6 +149,7 @@ impl CompactStatus {
             task_type: ret.compaction_task_type as i32,
             split_by_state_table: group.compaction_config.split_by_state_table,
             split_weight_by_vnode: group.compaction_config.split_weight_by_vnode,
+            table_vnode_partition: BTreeMap::default(),
         };
         Some(compact_task)
     }
@@ -237,7 +237,6 @@ pub fn create_compaction_task(
         input,
         target_file_size,
         compaction_task_type,
-        enable_split_by_table: false,
     }
 }
 

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

@@ -28,7 +28,6 @@ pub struct MinOverlappingPicker {
     level: usize,
     target_level: usize,
     max_select_bytes: u64,
-    split_by_table: bool,
     overlap_strategy: Arc<dyn OverlapStrategy>,
 }
 
@@ -37,14 +36,12 @@ impl MinOverlappingPicker {
         level: usize,
         target_level: usize,
         max_select_bytes: u64,
-        split_by_table: bool,
         overlap_strategy: Arc<dyn OverlapStrategy>,
     ) -> MinOverlappingPicker {
         MinOverlappingPicker {
             level,
             target_level,
             max_select_bytes,
-            split_by_table,
             overlap_strategy,
         }
     }
@@ -66,9 +63,6 @@ impl MinOverlappingPicker {
                 if level_handlers[self.level].is_pending_compact(&table.sst_id) {
                     break;
                 }
-                if self.split_by_table && table.table_ids != select_tables[left].table_ids {
-                    break;
-                }
                 if select_file_size > self.max_select_bytes {
                     break;
                 }
@@ -384,13 +378,8 @@ pub mod tests {
 
     #[test]
     fn test_compact_l1() {
-        let mut picker = MinOverlappingPicker::new(
-            1,
-            2,
-            10000,
-            false,
-            Arc::new(RangeOverlapStrategy::default()),
-        );
+        let mut picker =
+            MinOverlappingPicker::new(1, 2, 10000, Arc::new(RangeOverlapStrategy::default()));
         let levels = vec![
             Level {
                 level_idx: 1,
@@ -466,13 +455,8 @@ pub mod tests {
 
     #[test]
     fn test_expand_l1_files() {
-        let mut picker = MinOverlappingPicker::new(
-            1,
-            2,
-            10000,
-            false,
-            Arc::new(RangeOverlapStrategy::default()),
-        );
+        let mut picker =
+            MinOverlappingPicker::new(1, 2, 10000, Arc::new(RangeOverlapStrategy::default()));
         let levels = vec![
             Level {
                 level_idx: 1,
@@ -812,7 +796,7 @@ pub mod tests {
         ];
         // no limit
         let picker =
-            MinOverlappingPicker::new(2, 3, 1000, false, Arc::new(RangeOverlapStrategy::default()));
+            MinOverlappingPicker::new(2, 3, 1000, Arc::new(RangeOverlapStrategy::default()));
         let (select_files, target_files) = picker.pick_tables(
             &levels[1].table_infos,
             &levels[2].table_infos,

src/meta/src/hummock/compaction/selector/level_selector.rs

@@ -121,7 +121,6 @@ impl DynamicLevelSelectorCore {
                     picker_info.select_level,
                     picker_info.target_level,
                     self.config.max_bytes_for_level_base,
-                    self.config.split_by_state_table,
                     overlap_strategy,
                 ))
             }

src/meta/src/hummock/manager/mod.rs

@@ -80,7 +80,8 @@ use crate::hummock::metrics_utils::{
 };
 use crate::hummock::{CompactorManagerRef, TASK_NORMAL};
 use crate::manager::{
-    CatalogManagerRef, ClusterManagerRef, FragmentManagerRef, IdCategory, MetaSrvEnv, META_NODE_ID,
+    CatalogManagerRef, ClusterManagerRef, FragmentManagerRef, IdCategory, MetaSrvEnv, TableId,
+    META_NODE_ID,
 };
 use crate::model::{
     BTreeMapEntryTransaction, BTreeMapTransaction, ClusterId, MetadataModel, ValTransaction,
@@ -144,6 +145,9 @@ pub struct HummockManager {
     // `compaction_state` will record the types of compact tasks that can be triggered in `hummock`
     // and suggest types with a certain priority.
     pub compaction_state: CompactionState,
+
+    group_to_table_vnode_partition:
+        parking_lot::RwLock<HashMap<CompactionGroupId, BTreeMap<TableId, u32>>>,
 }
 
 pub type HummockManagerRef = Arc<HummockManager>;
@@ -390,6 +394,7 @@ impl HummockManager {
             history_table_throughput: parking_lot::RwLock::new(HashMap::default()),
             compactor_streams_change_tx,
             compaction_state: CompactionState::new(),
+            group_to_table_vnode_partition: parking_lot::RwLock::new(HashMap::default()),
         };
         let instance = Arc::new(instance);
         instance.start_worker(rx).await;
@@ -813,6 +818,14 @@ impl HummockManager {
         // lock in compaction_guard, take out all table_options in advance there may be a
         // waste of resources here, need to add a more efficient filter in catalog_manager
         let all_table_id_to_option = self.catalog_manager.get_all_table_options().await;
+        let mut table_to_vnode_partition = match self
+            .group_to_table_vnode_partition
+            .read()
+            .get(&compaction_group_id)
+        {
+            Some(table_to_vnode_partition) => table_to_vnode_partition.clone(),
+            None => BTreeMap::default(),
+        };
 
         let mut compaction_guard = write_lock!(self, compaction).await;
         let compaction = compaction_guard.deref_mut();
@@ -903,6 +916,7 @@ impl HummockManager {
             .unwrap()
             .member_table_ids
             .clone();
+
         let is_trivial_reclaim = CompactStatus::is_trivial_reclaim(&compact_task);
         let is_trivial_move = CompactStatus::is_trivial_move_task(&compact_task);
 
@@ -963,6 +977,18 @@ impl HummockManager {
             compact_task.current_epoch_time = Epoch::now().0;
             compact_task.compaction_filter_mask =
                 group_config.compaction_config.compaction_filter_mask;
+            table_to_vnode_partition
+                .retain(|table_id, _| compact_task.existing_table_ids.contains(table_id));
+
+            if group_config.compaction_config.split_weight_by_vnode > 0 {
+                // TODO: unify to table_to_vnode_partition directly
+                table_to_vnode_partition.insert(
+                    member_table_ids[0],
+                    group_config.compaction_config.split_weight_by_vnode,
+                );
+            }
+
+            compact_task.table_vnode_partition = table_to_vnode_partition;
 
             let mut compact_task_assignment =
                 BTreeMapTransaction::new(&mut compaction.compact_task_assignment);
@@ -2164,7 +2190,6 @@ impl HummockManager {
                 ));
                 split_group_trigger_interval
                     .set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-                split_group_trigger_interval.reset();
 
                 let split_group_trigger = IntervalStream::new(split_group_trigger_interval)
                     .map(|_| HummockTimerEvent::GroupSplit);
@@ -2484,31 +2509,60 @@ impl HummockManager {
         let mv_group_id: CompactionGroupId = StaticCompactionGroupId::MaterializedView.into();
         let partition_vnode_count = self.env.opts.partition_vnode_count;
         let window_size = HISTORY_TABLE_INFO_STATISTIC_TIME / (checkpoint_secs as usize);
+
+        let mut group_to_table_vnode_partition = HashMap::default();
+
         for group in &group_infos {
+            let table_vnode_partition_mapping = group_to_table_vnode_partition
+                .entry(group.group_id)
+                .or_insert(BTreeMap::default());
+
             if group.table_statistic.len() == 1 {
+                // no need to handle the dedication compaciton group
                 continue;
             }
 
             for (table_id, table_size) in &group.table_statistic {
-                if !created_tables.contains(table_id) {
-                    continue;
-                }
                 let mut is_high_write_throughput = false;
                 let mut is_low_write_throughput = true;
+                let is_creating_table = !created_tables.contains(table_id);
                 if let Some(history) = table_write_throughput.get(table_id) {
-                    if history.len() >= window_size {
-                        is_high_write_throughput = history.iter().all(|throughput| {
-                            *throughput / checkpoint_secs
-                                > self.env.opts.table_write_throughput_threshold
-                        });
-                        is_low_write_throughput = history.iter().any(|throughput| {
-                            *throughput / checkpoint_secs
-                                < self.env.opts.min_table_split_write_throughput
-                        });
+                    if !is_creating_table {
+                        if history.len() >= window_size {
+                            is_high_write_throughput = history.iter().all(|throughput| {
+                                *throughput / checkpoint_secs
+                                    > self.env.opts.table_write_throughput_threshold
+                            });
+                            is_low_write_throughput = history.iter().any(|throughput| {
+                                *throughput / checkpoint_secs
+                                    < self.env.opts.min_table_split_write_throughput
+                            });
+                        }
+                    } else {
+                        let sum = history.iter().sum::<u64>();
+                        is_low_write_throughput = sum
+                            < self.env.opts.min_table_split_write_throughput
+                                * history.len() as u64
+                                * checkpoint_secs;
                     }
                 }
                 let state_table_size = *table_size;
 
+                {
+                    if !is_low_write_throughput {
+                        table_vnode_partition_mapping.insert(*table_id, 4_u32);
+                    } else if state_table_size > self.env.opts.cut_table_size_limit {
+                        table_vnode_partition_mapping.insert(*table_id, 1_u32);
+                    } else {
+                        table_vnode_partition_mapping.remove(table_id);
+                    }
+                }
+
+                if !created_tables.contains(table_id) {
+                    // do not split the creating table
+                    continue;
+                }
+
                 if is_low_write_throughput {
                     continue;
                 }
@@ -2532,6 +2586,7 @@ impl HummockManager {
                     }
                 }
 
+                table_vnode_partition_mapping.insert(*table_id, partition_vnode_count);
                 let ret = self
                     .move_state_table_to_compaction_group(
                         parent_group_id,
@@ -2557,6 +2612,13 @@ impl HummockManager {
                 }
             }
         }
+
+        tracing::info!(
+            "group_to_table_vnode_partition {:?}",
+            group_to_table_vnode_partition
+        );
+
+        *self.group_to_table_vnode_partition.write() = group_to_table_vnode_partition;
     }
 
     #[named]

src/meta/src/manager/catalog/mod.rs

@@ -2675,6 +2675,7 @@ impl CatalogManager {
             .database
             .tables
             .values()
+            .filter(|table| table.get_stream_job_status() != Ok(StreamJobStatus::Creating))
             .map(|table| table.id)
             .collect()
     }

src/meta/src/manager/env.rs

@@ -169,6 +169,9 @@ pub struct MetaOpts {
 
     pub compaction_task_max_heartbeat_interval_secs: u64,
     pub compaction_config: Option<CompactionConfig>,
+
+    /// The size limit to split a state-table to independent sstable.
+    pub cut_table_size_limit: u64,
 }
 
 impl MetaOpts {
@@ -209,6 +212,7 @@ impl MetaOpts {
             partition_vnode_count: 32,
             compaction_task_max_heartbeat_interval_secs: 0,
             compaction_config: None,
+            cut_table_size_limit: 1024 * 1024 * 1024,
         }
     }
 }

src/storage/src/hummock/compactor/compaction_utils.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashSet;
+use std::collections::{BTreeMap, HashSet};
 use std::marker::PhantomData;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
@@ -119,9 +119,9 @@ pub struct TaskConfig {
     pub stats_target_table_ids: Option<HashSet<u32>>,
     pub task_type: compact_task::TaskType,
     pub is_target_l0_or_lbase: bool,
-    pub split_by_table: bool,
-    pub split_weight_by_vnode: u32,
     pub use_block_based_filter: bool,
+
+    pub table_vnode_partition: BTreeMap<u32, u32>,
 }
 
 pub fn build_multi_compaction_filter(compact_task: &CompactTask) -> MultiCompactionFilter {

src/storage/src/hummock/compactor/compactor_runner.rs

@@ -104,9 +104,8 @@ impl CompactorRunner {
                 task_type: task.task_type(),
                 is_target_l0_or_lbase: task.target_level == 0
                     || task.target_level == task.base_level,
-                split_by_table: task.split_by_state_table,
-                split_weight_by_vnode: task.split_weight_by_vnode,
                 use_block_based_filter,
+                table_vnode_partition: task.table_vnode_partition.clone(),
             },
             object_id_getter,
         );