feat(wal): increase recovery parallelism (GreptimeTeam#4689)

* Refactor RaftEngineLogStore to use references for config

 - Updated `RaftEngineLogStore::try_new` to accept a reference to `RaftEngineConfig` instead of taking ownership.
 - Replaced direct usage of `config` with individual fields (`sync_write`, `sync_period`, `read_batch_size`).
 - Adjusted test cases to pass references to `RaftEngineConfig`.

* Add parallelism configuration for WAL recovery

 - Introduced `recovery_parallelism` setting in `datanode.example.toml` and `standalone.example.toml` for configuring parallelism during WAL recovery.
 - Updated `Cargo.lock` and `Cargo.toml` to include `num_cpus` dependency.
 - Modified `RaftEngineConfig` to include `recovery_parallelism` with a default value set to the number of CP

* feat/wal-recovery-parallelism:
 Add `wal.recovery_parallelism` configuration option

 - Introduced `wal.recovery_parallelism` to config.md for specifying parallelism during WAL recovery.
 - Updated `RaftEngineLogStore` to include `recovery_threads` from the new configuration.

* fix: ut

config/config.md

@@ -68,6 +68,7 @@
 | `wal.enable_log_recycle` | Bool | `true` | Whether to reuse logically truncated log files.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.prefill_log_files` | Bool | `false` | Whether to pre-create log files on start up.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.sync_period` | String | `10s` | Duration for fsyncing log files.<br/>**It's only used when the provider is `raft_engine`**. |
+| `wal.recovery_parallelism` | Integer | `2` | Parallelism during WAL recovery. |
 | `wal.broker_endpoints` | Array | -- | The Kafka broker endpoints.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.auto_create_topics` | Bool | `true` | Automatically create topics for WAL.<br/>Set to `true` to automatically create topics for WAL.<br/>Otherwise, use topics named `topic_name_prefix_[0..num_topics)` |
 | `wal.num_topics` | Integer | `64` | Number of topics.<br/>**It's only used when the provider is `kafka`**. |
@@ -381,6 +382,7 @@
 | `wal.enable_log_recycle` | Bool | `true` | Whether to reuse logically truncated log files.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.prefill_log_files` | Bool | `false` | Whether to pre-create log files on start up.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.sync_period` | String | `10s` | Duration for fsyncing log files.<br/>**It's only used when the provider is `raft_engine`**. |
+| `wal.recovery_parallelism` | Integer | `2` | Parallelism during WAL recovery. |
 | `wal.broker_endpoints` | Array | -- | The Kafka broker endpoints.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.max_batch_bytes` | String | `1MB` | The max size of a single producer batch.<br/>Warning: Kafka has a default limit of 1MB per message in a topic.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.consumer_wait_timeout` | String | `100ms` | The consumer wait timeout.<br/>**It's only used when the provider is `kafka`**. |

config/datanode.example.toml

@@ -170,6 +170,9 @@ prefill_log_files = false
 ## **It's only used when the provider is `raft_engine`**.
 sync_period = "10s"
 
+## Parallelism during WAL recovery.
+recovery_parallelism = 2
+
 ## The Kafka broker endpoints.
 ## **It's only used when the provider is `kafka`**.
 broker_endpoints = ["127.0.0.1:9092"]

config/standalone.example.toml

@@ -174,6 +174,9 @@ prefill_log_files = false
 ## **It's only used when the provider is `raft_engine`**.
 sync_period = "10s"
 
+## Parallelism during WAL recovery.
+recovery_parallelism = 2
+
 ## The Kafka broker endpoints.
 ## **It's only used when the provider is `kafka`**.
 broker_endpoints = ["127.0.0.1:9092"]

src/cmd/tests/load_config_test.rs

@@ -65,6 +65,7 @@ fn test_load_datanode_example_config() {
             wal: DatanodeWalConfig::RaftEngine(RaftEngineConfig {
                 dir: Some("/tmp/greptimedb/wal".to_string()),
                 sync_period: Some(Duration::from_secs(10)),
+                recovery_parallelism: 2,
                 ..Default::default()
             }),
             storage: StorageConfig {
@@ -207,6 +208,7 @@ fn test_load_standalone_example_config() {
             wal: DatanodeWalConfig::RaftEngine(RaftEngineConfig {
                 dir: Some("/tmp/greptimedb/wal".to_string()),
                 sync_period: Some(Duration::from_secs(10)),
+                recovery_parallelism: 2,
                 ..Default::default()
             }),
             region_engine: vec![

src/common/wal/Cargo.toml

@@ -17,6 +17,7 @@ common-macro.workspace = true
 common-telemetry.workspace = true
 futures-util.workspace = true
 humantime-serde.workspace = true
+num_cpus.workspace = true
 rskafka.workspace = true
 rustls = { version = "0.23", default-features = false, features = ["ring", "logging", "std", "tls12"] }
 rustls-native-certs = "0.7"

src/common/wal/src/config/raft_engine.rs

@@ -41,6 +41,8 @@ pub struct RaftEngineConfig {
     /// Duration for fsyncing log files.
     #[serde(with = "humantime_serde")]
     pub sync_period: Option<Duration>,
+    /// Parallelism during log recovery.
+    pub recovery_parallelism: usize,
 }
 
 impl Default for RaftEngineConfig {
@@ -55,6 +57,7 @@ impl Default for RaftEngineConfig {
             enable_log_recycle: true,
             prefill_log_files: false,
             sync_period: None,
+            recovery_parallelism: num_cpus::get(),
         }
     }
 }

src/datanode/src/datanode.rs

@@ -454,7 +454,7 @@ impl DatanodeBuilder {
             "Creating raft-engine logstore with config: {:?} and storage path: {}",
             config, &wal_dir
         );
-        let logstore = RaftEngineLogStore::try_new(wal_dir, config.clone())
+        let logstore = RaftEngineLogStore::try_new(wal_dir, config)
             .await
             .map_err(Box::new)
             .context(OpenLogStoreSnafu)?;

src/log-store/src/raft_engine/log_store.rs

@@ -16,6 +16,7 @@ use std::collections::{hash_map, HashMap};
 use std::fmt::{Debug, Formatter};
 use std::sync::atomic::{AtomicI64, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 use async_stream::stream;
 use common_runtime::{RepeatedTask, TaskFunction};
@@ -40,7 +41,9 @@ use crate::raft_engine::protos::logstore::{EntryImpl, NamespaceImpl};
 const NAMESPACE_PREFIX: &str = "$sys/";
 
 pub struct RaftEngineLogStore {
-    config: RaftEngineConfig,
+    sync_write: bool,
+    sync_period: Option<Duration>,
+    read_batch_size: usize,
     engine: Arc<Engine>,
     gc_task: RepeatedTask<Error>,
     last_sync_time: AtomicI64,
@@ -76,7 +79,7 @@ impl TaskFunction<Error> for PurgeExpiredFilesFunction {
 }
 
 impl RaftEngineLogStore {
-    pub async fn try_new(dir: String, config: RaftEngineConfig) -> Result<Self> {
+    pub async fn try_new(dir: String, config: &RaftEngineConfig) -> Result<Self> {
         let raft_engine_config = Config {
             dir,
             purge_threshold: ReadableSize(config.purge_threshold.0),
@@ -85,6 +88,7 @@ impl RaftEngineLogStore {
             target_file_size: ReadableSize(config.file_size.0),
             enable_log_recycle: config.enable_log_recycle,
             prefill_for_recycle: config.prefill_log_files,
+            recovery_threads: config.recovery_parallelism,
             ..Default::default()
         };
         let engine = Arc::new(Engine::open(raft_engine_config).context(RaftEngineSnafu)?);
@@ -96,7 +100,9 @@ impl RaftEngineLogStore {
         );
 
         let log_store = Self {
-            config,
+            sync_write: config.sync_write,
+            sync_period: config.sync_period,
+            read_batch_size: config.read_batch_size,
             engine,
             gc_task,
             last_sync_time: AtomicI64::new(0),
@@ -196,7 +202,9 @@ impl RaftEngineLogStore {
 impl Debug for RaftEngineLogStore {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("RaftEngineLogsStore")
-            .field("config", &self.config)
+            .field("sync_write", &self.sync_write)
+            .field("sync_period", &self.sync_period)
+            .field("read_batch_size", &self.read_batch_size)
             .field("started", &self.gc_task.started())
             .finish()
     }
@@ -228,9 +236,9 @@ impl LogStore for RaftEngineLogStore {
 
         let (mut batch, last_entry_ids) = self.entries_to_batch(entries)?;
 
-        let mut sync = self.config.sync_write;
+        let mut sync = self.sync_write;
 
-        if let Some(sync_period) = &self.config.sync_period {
+        if let Some(sync_period) = &self.sync_period {
             let now = common_time::util::current_time_millis();
             if now - self.last_sync_time.load(Ordering::Relaxed) >= sync_period.as_millis() as i64 {
                 self.last_sync_time.store(now, Ordering::Relaxed);
@@ -276,7 +284,7 @@ impl LogStore for RaftEngineLogStore {
             entry_id,
             self.span(ns)
         );
-        let max_batch_size = self.config.read_batch_size;
+        let max_batch_size = self.read_batch_size;
         let (tx, mut rx) = tokio::sync::mpsc::channel(max_batch_size);
         let _handle = common_runtime::spawn_global(async move {
             while start_index <= last_index {
@@ -489,7 +497,7 @@ mod tests {
         let dir = create_temp_dir("raft-engine-logstore-test");
         let logstore = RaftEngineLogStore::try_new(
             dir.path().to_str().unwrap().to_string(),
-            RaftEngineConfig::default(),
+            &RaftEngineConfig::default(),
         )
         .await
         .unwrap();
@@ -502,7 +510,7 @@ mod tests {
         let dir = create_temp_dir("raft-engine-logstore-test");
         let logstore = RaftEngineLogStore::try_new(
             dir.path().to_str().unwrap().to_string(),
-            RaftEngineConfig::default(),
+            &RaftEngineConfig::default(),
         )
         .await
         .unwrap();
@@ -528,7 +536,7 @@ mod tests {
         let dir = create_temp_dir("raft-engine-logstore-test");
         let logstore = RaftEngineLogStore::try_new(
             dir.path().to_str().unwrap().to_string(),
-            RaftEngineConfig::default(),
+            &RaftEngineConfig::default(),
         )
         .await
         .unwrap();
@@ -570,7 +578,7 @@ mod tests {
         {
             let logstore = RaftEngineLogStore::try_new(
                 dir.path().to_str().unwrap().to_string(),
-                RaftEngineConfig::default(),
+                &RaftEngineConfig::default(),
             )
             .await
             .unwrap();
@@ -590,7 +598,7 @@ mod tests {
 
         let logstore = RaftEngineLogStore::try_new(
             dir.path().to_str().unwrap().to_string(),
-            RaftEngineConfig::default(),
+            &RaftEngineConfig::default(),
         )
         .await
         .unwrap();
@@ -634,7 +642,7 @@ mod tests {
             ..Default::default()
         };
 
-        RaftEngineLogStore::try_new(path, config).await.unwrap()
+        RaftEngineLogStore::try_new(path, &config).await.unwrap()
     }
 
     #[tokio::test]

src/log-store/src/test_util/log_store_util.rs

@@ -29,7 +29,7 @@ pub async fn create_tmp_local_file_log_store<P: AsRef<Path>>(path: P) -> RaftEng
         file_size: ReadableSize::kb(128),
         ..Default::default()
     };
-    RaftEngineLogStore::try_new(path, cfg).await.unwrap()
+    RaftEngineLogStore::try_new(path, &cfg).await.unwrap()
 }
 
 /// Create a [KafkaLogStore].

tests-integration/tests/http.rs

@@ -906,6 +906,7 @@ fn drop_lines_with_inconsistent_results(input: String) -> String {
         "metadata_cache_size =",
         "content_cache_size =",
         "name =",
+        "recovery_parallelism =",
     ];
 
     input