feat: Implement WAL-based RaftLog storage

Cargo.toml

@@ -289,6 +289,7 @@ flagset = "0.4"
 flatbuffers = "24" # Must use the same version with arrow-ipc
 flate2 = "1"
 foreign_vec = "0.1.0"
+fs_extra = "1.3.0"
 futures = "0.3.24"
 futures-async-stream = { version = "0.2.7" }
 futures-util = "0.3.24"
@@ -408,6 +409,7 @@ prost = { version = "0.13" }
 prost-build = { version = "0.13" }
 prqlc = "0.11.3"
 quanta = "0.11.1"
+raft-log = { version = "0.2.1" }
 rand = { version = "0.8.5", features = ["small_rng"] }
 rayon = "1.9.0"
 recursive = "0.1.1"
@@ -617,7 +619,7 @@ deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "3038c145" }
 ethnum = { git = "https://github.com/datafuse-extras/ethnum-rs", rev = "4cb05f1" }
 jsonb = { git = "https://github.com/databendlabs/jsonb", rev = "ada713c" }
 openai_api_rust = { git = "https://github.com/datafuse-extras/openai-api", rev = "819a0ed" }
-openraft = { git = "https://github.com/databendlabs/openraft", tag = "v0.10.0-alpha.6" }
+openraft = { git = "https://github.com/databendlabs/openraft", tag = "v0.10.0-alpha.7" }
 orc-rust = { git = "https://github.com/datafusion-contrib/orc-rust", rev = "dfb1ede" }
 recursive = { git = "https://github.com/datafuse-extras/recursive.git", rev = "6af35a1" }
 sled = { git = "https://github.com/datafuse-extras/sled", tag = "v0.34.7-datafuse.1" }

src/meta/binaries/Cargo.toml

@@ -43,6 +43,7 @@ databend-meta = { workspace = true }
 fastrace = { workspace = true }
 futures = { workspace = true }
 log = { workspace = true }
+raft-log = { workspace = true }
 rand = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }

src/meta/binaries/metactl/export_from_disk.rs

@@ -36,7 +36,7 @@ pub async fn export_from_dir(args: &ExportArgs) -> anyhow::Result<()> {
     eprintln!();
     eprintln!("Export:");
 
-    let sto_inn = StoreInner::open_create(&raft_config, Some(()), None).await?;
+    let sto_inn = StoreInner::open(&raft_config).await?;
     let mut lines = Arc::new(sto_inn).export();
 
     eprintln!("    From: {}", raft_config.raft_dir);

src/meta/binaries/metactl/import.rs

@@ -29,10 +29,10 @@ use databend_common_meta_raft_store::config::RaftConfig;
 use databend_common_meta_raft_store::key_spaces::RaftStoreEntry;
 use databend_common_meta_raft_store::key_spaces::SMEntry;
 use databend_common_meta_raft_store::ondisk::DataVersion;
+use databend_common_meta_raft_store::raft_log_v004;
 use databend_common_meta_raft_store::sm_v003::adapter::SnapshotUpgradeV002ToV003;
 use databend_common_meta_raft_store::sm_v003::write_entry::WriteEntry;
 use databend_common_meta_raft_store::sm_v003::SnapshotStoreV003;
-use databend_common_meta_raft_store::state::RaftState;
 use databend_common_meta_raft_store::state_machine::MetaSnapshotId;
 use databend_common_meta_sled_store::get_sled_db;
 use databend_common_meta_sled_store::init_sled_db;
@@ -51,6 +51,7 @@ use databend_common_meta_types::Endpoint;
 use databend_common_meta_types::LogEntry;
 use databend_common_meta_types::Node;
 use databend_meta::store::RaftStore;
+use raft_log::api::raft_log_writer::RaftLogWriter;
 use url::Url;
 
 use crate::reading;
@@ -108,6 +109,7 @@ async fn import_lines<B: BufRead + 'static>(
         }
         DataVersion::V002 => import_v002(raft_config, it).await?,
         DataVersion::V003 => import_v003(raft_config, it).await?,
+        DataVersion::V004 => crate::import_v004::import_v004(raft_config, it).await?,
     };
 
     Ok(max_log_id)
@@ -305,10 +307,9 @@ async fn init_new_cluster(
     eprintln!();
     eprintln!("Initialize Cluster with: {:?}", nodes);
 
-    let db = get_sled_db();
     let raft_config: RaftConfig = args.clone().into();
 
-    let mut sto = RaftStore::open_create(&raft_config, Some(()), None).await?;
+    let mut sto = RaftStore::open(&raft_config).await?;
 
     let last_applied = {
         let sm2 = sto.get_state_machine().await;
@@ -373,8 +374,13 @@ async fn init_new_cluster(
     }
 
     // Reset node id
-    let raft_state = RaftState::open_create(&db, &raft_config, Some(()), None).await?;
-    raft_state.set_node_id(args.id).await?;
+    {
+        let mut log = sto.log.write().await;
+        log.save_user_data(Some(raft_log_v004::LogStoreMeta {
+            node_id: Some(args.id),
+        }))?;
+        raft_log_v004::util::blocking_flush(&mut log).await?;
+    }
 
     Ok(())
 }