refactor(meta): use insert_many to reduce latency (#20123)

src/common/src/config.rs

@@ -532,6 +532,10 @@ pub struct MetaDeveloperConfig {
     #[serde(default = "default::developer::hummock_time_travel_sst_info_insert_batch_size")]
     /// Max number of SSTs inserted into meta store per INSERT, during time travel metadata writing.
     pub hummock_time_travel_sst_info_insert_batch_size: usize,
+
+    #[serde(default = "default::developer::hummock_time_travel_epoch_version_insert_batch_size")]
+    /// Max number of epoch-to-version inserted into meta store per INSERT, during time travel metadata writing.
+    pub hummock_time_travel_epoch_version_insert_batch_size: usize,
 }
 
 /// The section `[server]` in `risingwave.toml`.
@@ -2054,6 +2058,10 @@ pub mod default {
             100
         }
 
+        pub fn hummock_time_travel_epoch_version_insert_batch_size() -> usize {
+            1000
+        }
+
         pub fn memory_controller_threshold_aggressive() -> f64 {
             0.9
         }

src/config/example.toml

@@ -93,6 +93,7 @@ meta_actor_cnt_per_worker_parallelism_soft_limit = 100
 meta_actor_cnt_per_worker_parallelism_hard_limit = 400
 meta_hummock_time_travel_sst_info_fetch_batch_size = 10000
 meta_hummock_time_travel_sst_info_insert_batch_size = 100
+meta_hummock_time_travel_epoch_version_insert_batch_size = 1000
 
 [meta.meta_store_config]
 max_connections = 10

src/meta/node/src/lib.rs

@@ -370,6 +370,10 @@ pub fn start(
                     .meta
                     .developer
                     .hummock_time_travel_sst_info_insert_batch_size,
+                hummock_time_travel_epoch_version_insert_batch_size: config
+                    .meta
+                    .developer
+                    .hummock_time_travel_epoch_version_insert_batch_size,
                 min_delta_log_num_for_hummock_version_checkpoint: config
                     .meta
                     .min_delta_log_num_for_hummock_version_checkpoint,

src/meta/src/hummock/manager/time_travel.rs

@@ -430,15 +430,32 @@ impl HummockManager {
             Ok(count)
         }
 
+        let mut batch = vec![];
         for (table_id, _cg_id, committed_epoch) in tables_to_commit {
             let version_id: u64 = delta.id.to_u64();
             let m = hummock_epoch_to_version::ActiveModel {
                 epoch: Set(committed_epoch.try_into().unwrap()),
                 table_id: Set(table_id.table_id.into()),
                 version_id: Set(version_id.try_into().unwrap()),
             };
+            batch.push(m);
+            if batch.len()
+                >= self
+                    .env
+                    .opts
+                    .hummock_time_travel_epoch_version_insert_batch_size
+            {
+                // There should be no conflict rows.
+                hummock_epoch_to_version::Entity::insert_many(std::mem::take(&mut batch))
+                    .do_nothing()
+                    .exec(txn)
+                    .await?;
+            }
+        }
+        if !batch.is_empty() {
             // There should be no conflict rows.
-            hummock_epoch_to_version::Entity::insert(m)
+            hummock_epoch_to_version::Entity::insert_many(batch)
+                .do_nothing()
                 .exec(txn)
                 .await?;
         }

src/meta/src/manager/env.rs

@@ -115,6 +115,7 @@ pub struct MetaOpts {
     pub hummock_time_travel_snapshot_interval: u64,
     pub hummock_time_travel_sst_info_fetch_batch_size: usize,
     pub hummock_time_travel_sst_info_insert_batch_size: usize,
+    pub hummock_time_travel_epoch_version_insert_batch_size: usize,
     /// The minimum delta log number a new checkpoint should compact, otherwise the checkpoint
     /// attempt is rejected. Greater value reduces object store IO, meanwhile it results in
     /// more loss of in memory `HummockVersionCheckpoint::stale_objects` state when meta node is
@@ -274,6 +275,7 @@ impl MetaOpts {
             hummock_time_travel_snapshot_interval: 0,
             hummock_time_travel_sst_info_fetch_batch_size: 10_000,
             hummock_time_travel_sst_info_insert_batch_size: 10,
+            hummock_time_travel_epoch_version_insert_batch_size: 1000,
             min_delta_log_num_for_hummock_version_checkpoint: 1,
             min_sst_retention_time_sec: 3600 * 24 * 7,
             full_gc_interval_sec: 3600 * 24 * 7,