feat(remote_wal): add skeleton for remote wal related to meta srv (#2933

)

* feat: introduce wal config and kafka config

* feat: introduce kafka topic manager and selector

* feat: introduce region wal options

* chore: build region wal options upon starting meta srv

* feat: integrate region wal options allocator into table meta allocator

* chore: add wal config to metasrv.example.toml

* chore: add region wal options map to create table procedure

* feat: augment region create request with wal options

* feat: augment DatanodeTableValue with region wal options map

* chore: encode region wal options upon constructing table creator

* feat: persist region wal options when creating table meta

* fix: sqlness test

* chore: set default wal provider to raft-engine

* refactor: refactor wal options

* chore: update wal options allocator

* refactor: rename region wal options to wal options

* chore: update usages of region wal options

* chore: add some comments to kafka

* chore: fill in kafka config

* test: add tests for serde wal config

* test: add tests for wal options

* refactor: refactor wal options allocator to enum

* refactor: store wal options into the request options instead

* fix: typo

* fix: typo

* refactor: move wal options map to region info

* refactor: refacto serialization and deserialization of wal options

* refactor: use serde_json to encode wal options

* chore: rename wal_options_map to region_wal_options

* chore: resolve some review comments

* fix: typo

* refactor: replace kecab-case with snake_case

* fix: sqlness and converage tests

* fix: typo

* fix: coverage test

* fix: coverage test

* chore: resolve some review conversations

* fix: resolve some review conversations

* chore: format comments in metasrv.example.toml

* chore: update import style

* feat: integrate wal options allocator to standalone mode

* test: add compatible test for OpenRegion

* test: add compatible test for UpdateRegionMetadata

* chore: remove based suffix from topic selector type

config/metasrv.example.toml

@@ -42,3 +42,27 @@ first_heartbeat_estimate = "1000ms"
 # timeout = "10s"
 # connect_timeout = "10s"
 # tcp_nodelay = true
+
+[wal]
+# Available wal providers:
+# - "raft_engine" (default)
+# - "kafka"
+provider = "raft_engine"
+
+# There're none raft-engine wal config since meta srv only involves in remote wal currently.
+
+# Kafka wal config.
+# The broker endpoints of the Kafka cluster. ["127.0.0.1:9090"] by default.
+# broker_endpoints = ["127.0.0.1:9090"]
+# Number of topics to be created upon start.
+# num_topics = 64
+# Topic selector type.
+# Available selector types: 
+# - "round_robin" (default)
+# selector_type = "round_robin"
+# A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.
+# topic_name_prefix = "greptimedb_kafka_wal"
+# Number of partitions per topic.
+# num_partitions = 1
+# Expected number of replicas of each partition.
+# replication_factor = 3

src/cmd/Cargo.toml

@@ -58,6 +58,7 @@ serde_json.workspace = true
 servers.workspace = true
 session.workspace = true
 snafu.workspace = true
+store-api.workspace = true
 substrait.workspace = true
 table.workspace = true
 tokio.workspace = true

src/cmd/src/cli/bench.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
@@ -28,6 +28,7 @@ use common_telemetry::info;
 use datatypes::data_type::ConcreteDataType;
 use datatypes::schema::{ColumnSchema, RawSchema};
 use rand::Rng;
+use store_api::storage::RegionNumber;
 use table::metadata::{RawTableInfo, RawTableMeta, TableId, TableIdent, TableType};
 
 use self::metadata::TableMetadataBencher;
@@ -137,12 +138,12 @@ fn create_table_info(table_id: TableId, table_name: TableName) -> RawTableInfo {
     }
 }
 
-fn create_region_routes() -> Vec<RegionRoute> {
-    let mut regions = Vec::with_capacity(100);
+fn create_region_routes(regions: Vec<RegionNumber>) -> Vec<RegionRoute> {
+    let mut region_routes = Vec::with_capacity(100);
     let mut rng = rand::thread_rng();
 
-    for region_id in 0..64u64 {
-        regions.push(RegionRoute {
+    for region_id in regions.into_iter().map(u64::from) {
+        region_routes.push(RegionRoute {
             region: Region {
                 id: region_id.into(),
                 name: String::new(),
@@ -158,5 +159,11 @@ fn create_region_routes() -> Vec<RegionRoute> {
         });
     }
 
-    regions
+    region_routes
+}
+
+fn create_region_wal_options(regions: Vec<RegionNumber>) -> HashMap<RegionNumber, String> {
+    // TODO(niebayes): construct region wal options for benchmark.
+    let _ = regions;
+    HashMap::default()
 }

src/cmd/src/cli/bench/metadata.rs

@@ -17,7 +17,9 @@ use std::time::Instant;
 use common_meta::key::TableMetadataManagerRef;
 use common_meta::table_name::TableName;
 
-use super::{bench_self_recorded, create_region_routes, create_table_info};
+use crate::cli::bench::{
+    bench_self_recorded, create_region_routes, create_region_wal_options, create_table_info,
+};
 
 pub struct TableMetadataBencher {
     table_metadata_manager: TableMetadataManagerRef,
@@ -43,12 +45,15 @@ impl TableMetadataBencher {
                 let table_name = format!("bench_table_name_{}", i);
                 let table_name = TableName::new("bench_catalog", "bench_schema", table_name);
                 let table_info = create_table_info(i, table_name);
-                let region_routes = create_region_routes();
+
+                let regions: Vec<_> = (0..64).collect();
+                let region_routes = create_region_routes(regions.clone());
+                let region_wal_options = create_region_wal_options(regions);
 
                 let start = Instant::now();
 
                 self.table_metadata_manager
-                    .create_table_metadata(table_info, region_routes)
+                    .create_table_metadata(table_info, region_routes, region_wal_options)
                     .await
                     .unwrap();
 

src/cmd/src/cli/upgrade.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use async_trait::async_trait;
@@ -395,6 +396,9 @@ impl MigrateTableMetadata {
         let region_distribution: RegionDistribution =
             value.regions_id_map.clone().into_iter().collect();
 
+        // TODO(niebayes): properly fetch or construct wal options.
+        let region_wal_options = HashMap::default();
+
         let datanode_table_kvs = region_distribution
             .into_iter()
             .map(|(datanode_id, regions)| {
@@ -409,6 +413,7 @@ impl MigrateTableMetadata {
                             engine: engine.to_string(),
                             region_storage_path: region_storage_path.clone(),
                             region_options: (&value.table_info.meta.options).into(),
+                            region_wal_options: region_wal_options.clone(),
                         },
                     ),
                 )

src/cmd/src/error.rs

@@ -14,7 +14,7 @@
 
 use std::any::Any;
 
-use common_error::ext::ErrorExt;
+use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
 use config::ConfigError;
@@ -231,6 +231,12 @@ pub enum Error {
         #[snafu(source)]
         error: std::io::Error,
     },
+
+    #[snafu(display("Other error"))]
+    Other {
+        source: BoxedError,
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -276,6 +282,8 @@ impl ErrorExt for Error {
             Error::StartCatalogManager { source, .. } => source.status_code(),
 
             Error::SerdeJson { .. } | Error::FileIo { .. } => StatusCode::Unexpected,
+
+            Error::Other { source, .. } => source.status_code(),
         }
     }
 

src/cmd/src/standalone.rs

@@ -19,6 +19,7 @@ use async_trait::async_trait;
 use clap::Parser;
 use common_catalog::consts::MIN_USER_TABLE_ID;
 use common_config::{metadata_store_dir, KvBackendConfig, WalConfig};
+use common_error::ext::BoxedError;
 use common_meta::cache_invalidator::DummyCacheInvalidator;
 use common_meta::datanode_manager::DatanodeManagerRef;
 use common_meta::ddl::{DdlTaskExecutorRef, TableMetadataAllocatorRef};
@@ -27,6 +28,7 @@ use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::sequence::SequenceBuilder;
+use common_meta::wal::build_wal_options_allocator;
 use common_procedure::ProcedureManagerRef;
 use common_telemetry::info;
 use common_telemetry::logging::LoggingOptions;
@@ -35,7 +37,7 @@ use datanode::datanode::{Datanode, DatanodeBuilder};
 use file_engine::config::EngineConfig as FileEngineConfig;
 use frontend::frontend::FrontendOptions;
 use frontend::instance::builder::FrontendBuilder;
-use frontend::instance::standalone::StandaloneTableMetadataCreator;
+use frontend::instance::standalone::StandaloneTableMetadataAllocator;
 use frontend::instance::{FrontendInstance, Instance as FeInstance, StandaloneDatanodeManager};
 use frontend::service_config::{
     GrpcOptions, InfluxdbOptions, MysqlOptions, OpentsdbOptions, PostgresOptions, PromStoreOptions,
@@ -48,7 +50,7 @@ use servers::Mode;
 use snafu::ResultExt;
 
 use crate::error::{
-    CreateDirSnafu, IllegalConfigSnafu, InitDdlManagerSnafu, InitMetadataSnafu, Result,
+    CreateDirSnafu, IllegalConfigSnafu, InitDdlManagerSnafu, InitMetadataSnafu, OtherSnafu, Result,
     ShutdownDatanodeSnafu, ShutdownFrontendSnafu, StartDatanodeSnafu, StartFrontendSnafu,
     StartProcedureManagerSnafu, StopProcedureManagerSnafu,
 };
@@ -372,7 +374,16 @@ impl StartCommand {
                 .step(10)
                 .build(),
         );
-        let table_meta_allocator = Arc::new(StandaloneTableMetadataCreator::new(table_id_sequence));
+        // TODO(niebayes): add a wal config into the MixOptions and pass it to the allocator builder.
+        let wal_options_allocator =
+            build_wal_options_allocator(&common_meta::wal::WalConfig::default(), &kv_backend)
+                .await
+                .map_err(BoxedError::new)
+                .context(OtherSnafu)?;
+        let table_meta_allocator = Arc::new(StandaloneTableMetadataAllocator::new(
+            table_id_sequence,
+            wal_options_allocator,
+        ));
 
         let ddl_task_executor = Self::create_ddl_task_executor(
             kv_backend.clone(),

src/common/meta/Cargo.toml

@@ -34,11 +34,13 @@ prost.workspace = true
 regex.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+serde_with = "3"
 snafu.workspace = true
 store-api.workspace = true
 strum.workspace = true
 table.workspace = true
 tokio.workspace = true
+toml.workspace = true
 tonic.workspace = true
 
 [dev-dependencies]

src/common/meta/src/ddl.rs

@@ -12,11 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use api::v1::meta::Partition;
 use common_telemetry::tracing_context::W3cTrace;
-use store_api::storage::TableId;
+use store_api::storage::{RegionNumber, TableId};
 use table::metadata::RawTableInfo;
 
 use crate::cache_invalidator::CacheInvalidatorRef;
@@ -54,14 +55,25 @@ pub struct TableMetadataAllocatorContext {
     pub cluster_id: u64,
 }
 
+/// Metadata allocated to a table.
+pub struct TableMetadata {
+    /// Table id.
+    pub table_id: TableId,
+    /// Route information for each region of the table.
+    pub region_routes: Vec<RegionRoute>,
+    /// The encoded wal options for regions of the table.
+    // If a region does not have an associated wal options, no key for the region would be found in the map.
+    pub region_wal_options: HashMap<RegionNumber, String>,
+}
+
 #[async_trait::async_trait]
 pub trait TableMetadataAllocator: Send + Sync {
     async fn create(
         &self,
         ctx: &TableMetadataAllocatorContext,
         table_info: &mut RawTableInfo,
         partitions: &[Partition],
-    ) -> Result<(TableId, Vec<RegionRoute>)>;
+    ) -> Result<TableMetadata>;
 }
 
 pub type TableMetadataAllocatorRef = Arc<dyn TableMetadataAllocator>;