refactor(meta): limit max inflight time travel query (#18810)

src/common/src/config.rs

@@ -196,6 +196,10 @@ pub struct MetaConfig {
     #[serde(default = "default::meta::full_gc_object_limit")]
     pub full_gc_object_limit: u64,
 
+    /// Max number of inflight time travel query.
+    #[serde(default = "default::meta::max_inflight_time_travel_query")]
+    pub max_inflight_time_travel_query: u64,
+
     /// Schedule compaction for all compaction groups with this interval.
     #[serde(default = "default::meta::periodic_compaction_interval_sec")]
     pub periodic_compaction_interval_sec: u64,
@@ -1354,6 +1358,10 @@ pub mod default {
             100_000
         }
 
+        pub fn max_inflight_time_travel_query() -> u64 {
+            1000
+        }
+
         pub fn periodic_compaction_interval_sec() -> u64 {
             60
         }

src/config/docs.md

@@ -43,6 +43,7 @@ This page is automatically generated by `./risedev generate-example-config`
 | hummock_version_checkpoint_interval_sec | Interval of hummock version checkpoint. | 30 |
 | hybrid_partition_vnode_count | Count of partitions of tables in default group and materialized view group. The meta node will decide according to some strategy whether to cut the boundaries of the file according to the vnode alignment. Each partition contains aligned data of `vnode_count / hybrid_partition_vnode_count` consecutive virtual-nodes of one state table. Set it zero to disable this feature. | 4 |
 | max_heartbeat_interval_secs | Maximum allowed heartbeat interval in seconds. | 60 |
+| max_inflight_time_travel_query | Max number of inflight time travel query. | 1000 |
 | meta_leader_lease_secs |  | 30 |
 | min_delta_log_num_for_hummock_version_checkpoint | The minimum delta log number a new checkpoint should compact, otherwise the checkpoint attempt is rejected. | 10 |
 | min_sst_retention_time_sec | Objects within `min_sst_retention_time_sec` won't be deleted by hummock full GC, even they are dangling. | 86400 |

src/config/example.toml

@@ -17,6 +17,7 @@ dir = "./"
 min_sst_retention_time_sec = 86400
 full_gc_interval_sec = 86400
 full_gc_object_limit = 100000
+max_inflight_time_travel_query = 1000
 periodic_compaction_interval_sec = 60
 vacuum_interval_sec = 30
 vacuum_spin_interval_ms = 200

src/meta/node/src/lib.rs

@@ -396,6 +396,7 @@ pub fn start(
                 min_sst_retention_time_sec: config.meta.min_sst_retention_time_sec,
                 full_gc_interval_sec: config.meta.full_gc_interval_sec,
                 full_gc_object_limit: config.meta.full_gc_object_limit,
+                max_inflight_time_travel_query: config.meta.max_inflight_time_travel_query,
                 enable_committed_sst_sanity_check: config.meta.enable_committed_sst_sanity_check,
                 periodic_compaction_interval_sec: config.meta.periodic_compaction_interval_sec,
                 node_num_monitor_interval_sec: config.meta.node_num_monitor_interval_sec,

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

@@ -36,7 +36,7 @@ use risingwave_pb::hummock::{
 };
 use risingwave_pb::meta::subscribe_response::Operation;
 use tokio::sync::mpsc::UnboundedSender;
-use tokio::sync::Mutex;
+use tokio::sync::{Mutex, Semaphore};
 use tonic::Streaming;
 
 use crate::hummock::compaction::CompactStatus;
@@ -111,6 +111,7 @@ pub struct HummockManager {
     pub compaction_state: CompactionState,
     full_gc_state: FullGcState,
     now: Mutex<u64>,
+    inflight_time_travel_query: Semaphore,
 }
 
 pub type HummockManagerRef = Arc<HummockManager>;
@@ -245,6 +246,7 @@ impl HummockManager {
         let version_archive_dir = version_archive_dir(state_store_dir);
         let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
         let full_gc_object_limit = env.opts.full_gc_object_limit;
+        let inflight_time_travel_query = env.opts.max_inflight_time_travel_query;
         let instance = HummockManager {
             env,
             versioning: MonitoredRwLock::new(
@@ -282,6 +284,7 @@ impl HummockManager {
             compaction_state: CompactionState::new(),
             full_gc_state: FullGcState::new(Some(full_gc_object_limit)),
             now: Mutex::new(0),
+            inflight_time_travel_query: Semaphore::new(inflight_time_travel_query as usize),
         };
         let instance = Arc::new(instance);
         instance.init_time_travel_state().await?;

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

@@ -283,6 +283,12 @@ impl HummockManager {
         table_id: u32,
     ) -> Result<HummockVersion> {
         let sql_store = self.sql_store().ok_or_else(require_sql_meta_store_err)?;
+        let _permit = self.inflight_time_travel_query.try_acquire().map_err(|_| {
+            anyhow!(format!(
+                "too many inflight time travel queries, max_inflight_time_travel_query={}",
+                self.env.opts.max_inflight_time_travel_query
+            ))
+        })?;
         let epoch_to_version = hummock_epoch_to_version::Entity::find()
             .filter(
                 Condition::any()

src/meta/src/manager/env.rs

@@ -190,6 +190,8 @@ pub struct MetaOpts {
     pub full_gc_interval_sec: u64,
     /// Max number of object per full GC job can fetch.
     pub full_gc_object_limit: u64,
+    /// Max number of inflight time travel query.
+    pub max_inflight_time_travel_query: u64,
     /// Enable sanity check when SSTs are committed
     pub enable_committed_sst_sanity_check: bool,
     /// Schedule compaction for all compaction groups with this interval.
@@ -324,6 +326,7 @@ impl MetaOpts {
             min_sst_retention_time_sec: 3600 * 24 * 7,
             full_gc_interval_sec: 3600 * 24 * 7,
             full_gc_object_limit: 100_000,
+            max_inflight_time_travel_query: 1000,
             enable_committed_sst_sanity_check: false,
             periodic_compaction_interval_sec: 60,
             node_num_monitor_interval_sec: 10,