feat: pass Kafka pruned entry id when creating regions (#8282)

* refactor: use typed region wal options

Signed-off-by: WenyXu <wenymedia@gmail.com>

* feat: pass kafka pruned entry id to regions

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: remove unused error

Signed-off-by: WenyXu <wenymedia@gmail.com>

* refactor: clarify wal options serialization errors

Signed-off-by: WenyXu <wenymedia@gmail.com>

* test: cover legacy region wal options encoding

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix: recover legacy create table wal options

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix: lock remote wal during table creation

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix: fix unit tests-s

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix: refresh remote wal prune hints

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>

src/cli/src/bench/metadata.rs

@@ -49,12 +49,7 @@ impl TableMetadataBencher {
 
                 let regions: Vec<_> = (0..64).collect();
                 let region_routes = create_region_routes(regions.clone());
-                let region_wal_options = create_region_wal_options(regions)
-                    .into_iter()
-                    .map(|(region_id, wal_options)| {
-                        (region_id, serde_json::to_string(&wal_options).unwrap())
-                    })
-                    .collect();
+                let region_wal_options = create_region_wal_options(regions);
 
                 let start = Instant::now();
 

src/cli/src/error.rs

@@ -20,7 +20,7 @@ use common_macro::stack_trace_debug;
 use common_meta::peer::Peer;
 use object_store::Error as ObjectStoreError;
 use snafu::{Location, Snafu};
-use store_api::storage::TableId;
+use store_api::storage::{RegionId, TableId};
 
 #[derive(Snafu)]
 #[snafu(visibility(pub))]
@@ -82,6 +82,14 @@ pub enum Error {
         source: common_meta::error::Error,
     },
 
+    #[snafu(display("Failed to build create region request for region: {region_id}"))]
+    BuildCreateRegionRequest {
+        region_id: RegionId,
+        #[snafu(implicit)]
+        location: Location,
+        source: common_meta::error::Error,
+    },
+
     #[snafu(display("Unexpected error: {}", msg))]
     Unexpected {
         msg: String,
@@ -324,7 +332,8 @@ impl ErrorExt for Error {
         match self {
             Error::InitMetadata { source, .. }
             | Error::InitDdlManager { source, .. }
-            | Error::TableMetadata { source, .. } => source.status_code(),
+            | Error::TableMetadata { source, .. }
+            | Error::BuildCreateRegionRequest { source, .. } => source.status_code(),
 
             Error::MissingConfig { .. }
             | Error::LoadLayeredConfig { .. }

src/cli/src/metadata/repair/create_table.rs

@@ -25,7 +25,7 @@ use snafu::ResultExt;
 use store_api::storage::{RegionId, TableId};
 use table::metadata::TableInfo;
 
-use crate::error::{CovertColumnSchemasToDefsSnafu, Result};
+use crate::error::{BuildCreateRegionRequestSnafu, CovertColumnSchemasToDefsSnafu, Result};
 
 /// Generates a `CreateTableExpr` from a `TableInfo`.
 pub fn generate_create_table_expr(table_info: &TableInfo) -> Result<CreateTableExpr> {
@@ -82,12 +82,14 @@ pub fn make_create_region_request_for_peer(
 
     for region_number in &regions_on_this_peer {
         let region_id = RegionId::new(logical_table_id, *region_number);
-        let region_request = request_builder.build_one(
-            region_id,
-            storage_path.clone(),
-            &HashMap::new(),
-            &partition_exprs,
-        );
+        let region_request = request_builder
+            .build_one(
+                region_id,
+                storage_path.clone(),
+                &HashMap::new(),
+                &partition_exprs,
+            )
+            .context(BuildCreateRegionRequestSnafu { region_id })?;
         requests.push(region_request);
     }
 

src/common/meta/src/ddl.rs

@@ -12,10 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
 use std::sync::Arc;
 
-use store_api::storage::{RegionId, RegionNumber, TableId};
+use store_api::storage::{RegionId, TableId};
 
 use crate::DatanodeId;
 use crate::cache_invalidator::CacheInvalidatorRef;
@@ -27,6 +26,7 @@ use crate::key::table_route::PhysicalTableRouteValue;
 use crate::node_manager::NodeManagerRef;
 use crate::region_keeper::MemoryRegionKeeperRef;
 use crate::region_registry::LeaderRegionRegistryRef;
+use crate::wal_provider::RegionWalOptions;
 
 pub mod allocator;
 pub mod alter_database;
@@ -59,9 +59,9 @@ pub struct TableMetadata {
     pub table_id: TableId,
     /// Route information for each region of the table.
     pub table_route: PhysicalTableRouteValue,
-    /// The encoded wal options for regions of the table.
+    /// The 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>,
+    pub region_wal_options: RegionWalOptions,
 }
 
 pub type RegionFailureDetectorControllerRef = Arc<dyn RegionFailureDetectorController>;

src/common/meta/src/ddl/allocator/wal_options.rs

@@ -12,12 +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 store_api::storage::RegionNumber;
 
 use crate::error::Result;
+use crate::wal_provider::RegionWalOptions;
 
 pub type WalOptionsAllocatorRef = Arc<dyn WalOptionsAllocator>;
 
@@ -27,5 +27,5 @@ pub trait WalOptionsAllocator: Send + Sync {
         &self,
         region_numbers: &[RegionNumber],
         skip_wal: bool,
-    ) -> Result<HashMap<RegionNumber, String>>;
+    ) -> Result<RegionWalOptions>;
 }

src/common/meta/src/ddl/create_logical_tables/region_request.rs

@@ -66,7 +66,7 @@ impl CreateLogicalTablesProcedure {
                     storage_path.clone(),
                     &HashMap::new(),
                     &partition_exprs,
-                );
+                )?;
                 requests.push(one_region_request);
             }
         }

src/common/meta/src/ddl/create_table.rs

@@ -15,20 +15,18 @@
 pub mod executor;
 pub mod template;
 
-use std::collections::HashMap;
-
 use api::v1::CreateTableExpr;
 use async_trait::async_trait;
 use common_error::ext::BoxedError;
 use common_procedure::error::{
     ExternalSnafu, FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu,
 };
+use common_procedure::local::DynamicKeyLockGuard;
 use common_procedure::{Context as ProcedureContext, LockKey, Procedure, ProcedureId, Status};
 use common_telemetry::info;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt};
 use store_api::metadata::ColumnMetadata;
-use store_api::storage::RegionNumber;
 use strum::AsRefStr;
 use table::metadata::{TableId, TableInfo};
 use table::table_name::TableName;
@@ -47,6 +45,10 @@ use crate::peer::PeerAllocContext;
 use crate::region_keeper::OperatingRegionGuard;
 use crate::rpc::ddl::{CreateTableTask, QueryContext};
 use crate::rpc::router::{RegionRoute, operating_leader_region_roles};
+use crate::wal_provider::{
+    RegionWalOptions, acquire_remote_wal_read_locks, optional_region_wal_options_serde,
+    refresh_initial_pruned_entry_ids,
+};
 
 pub struct CreateTableProcedure {
     pub context: DdlContext,
@@ -56,6 +58,8 @@ pub struct CreateTableProcedure {
     pub opening_regions: Vec<OperatingRegionGuard>,
     /// The executor of the procedure.
     pub executor: CreateTableExecutor,
+    /// The guards of remote WAL topic locks.
+    remote_wal_lock_guards: Vec<DynamicKeyLockGuard>,
 }
 
 fn build_executor_from_create_table_data(
@@ -92,6 +96,7 @@ impl CreateTableProcedure {
             data: CreateTableData::new(task, query_context),
             opening_regions: vec![],
             executor,
+            remote_wal_lock_guards: vec![],
         })
     }
 
@@ -109,6 +114,7 @@ impl CreateTableProcedure {
             data,
             opening_regions: vec![],
             executor,
+            remote_wal_lock_guards: vec![],
         })
     }
 
@@ -120,7 +126,7 @@ impl CreateTableProcedure {
         self.table_info().ident.table_id
     }
 
-    fn region_wal_options(&self) -> Result<&HashMap<RegionNumber, String>> {
+    fn region_wal_options(&self) -> Result<&RegionWalOptions> {
         self.data
             .region_wal_options
             .as_ref()
@@ -175,6 +181,29 @@ impl CreateTableProcedure {
         Ok(Status::executing(true))
     }
 
+    async fn ensure_remote_wal_read_locks(&mut self, ctx: &ProcedureContext) -> Result<()> {
+        if !self.remote_wal_lock_guards.is_empty() {
+            return Ok(());
+        }
+
+        self.remote_wal_lock_guards =
+            acquire_remote_wal_read_locks(ctx, self.region_wal_options()?).await;
+
+        Ok(())
+    }
+
+    async fn refresh_initial_pruned_entry_ids(&mut self) -> Result<()> {
+        let region_wal_options =
+            self.data
+                .region_wal_options
+                .as_mut()
+                .context(error::UnexpectedSnafu {
+                    err_msg: "region_wal_options is not allocated",
+                })?;
+        refresh_initial_pruned_entry_ids(&self.context.table_metadata_manager, region_wal_options)
+            .await
+    }
+
     /// Creates regions on datanodes
     ///
     /// Abort(non-retry):
@@ -261,6 +290,7 @@ impl CreateTableProcedure {
         );
 
         self.opening_regions.clear();
+        self.remote_wal_lock_guards.clear();
         Ok(Status::done_with_output(table_id))
     }
 
@@ -294,7 +324,7 @@ impl CreateTableProcedure {
         &mut self,
         table_id: TableId,
         table_route: PhysicalTableRouteValue,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
     ) {
         self.data.task.table_info.ident.table_id = table_id;
         self.data.table_route = Some(table_route);
@@ -332,10 +362,21 @@ impl Procedure for CreateTableProcedure {
         match state {
             CreateTableState::Prepare => self.on_prepare().await,
             CreateTableState::DatanodeCreateRegions => {
-                let retrying = ctx.is_retrying().await.unwrap_or(false);
-                self.on_datanode_create_regions(retrying).await
+                async {
+                    self.ensure_remote_wal_read_locks(ctx).await?;
+                    self.refresh_initial_pruned_entry_ids().await?;
+                    let retrying = ctx.is_retrying().await.unwrap_or(false);
+                    self.on_datanode_create_regions(retrying).await
+                }
+                .await
+            }
+            CreateTableState::CreateMetadata => {
+                async {
+                    self.ensure_remote_wal_read_locks(ctx).await?;
+                    self.on_create_metadata(ctx.procedure_id).await
+                }
+                .await
             }
-            CreateTableState::CreateMetadata => self.on_create_metadata(ctx.procedure_id).await,
         }
         .map_err(map_to_procedure_error)
     }
@@ -376,7 +417,9 @@ pub struct CreateTableData {
     /// None stands for not allocated yet.
     pub(crate) table_route: Option<PhysicalTableRouteValue>,
     /// None stands for not allocated yet.
-    pub region_wal_options: Option<HashMap<RegionNumber, String>>,
+    #[serde(default)]
+    #[serde(with = "optional_region_wal_options_serde")]
+    pub region_wal_options: Option<RegionWalOptions>,
 }
 
 impl CreateTableData {

src/common/meta/src/ddl/create_table/executor.rs

@@ -22,7 +22,7 @@ use futures::future::join_all;
 use snafu::ensure;
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::TABLE_COLUMN_METADATA_EXTENSION_KEY;
-use store_api::storage::{RegionId, RegionNumber};
+use store_api::storage::RegionId;
 use table::metadata::{TableId, TableInfo};
 use table::table_name::TableName;
 
@@ -38,6 +38,7 @@ use crate::key::table_name::TableNameKey;
 use crate::key::table_route::{PhysicalTableRouteValue, TableRouteValue};
 use crate::node_manager::NodeManagerRef;
 use crate::rpc::router::{RegionRoute, find_leader_regions, find_leaders};
+use crate::wal_provider::RegionWalOptions;
 
 /// [CreateTableExecutor] performs:
 /// - Creates the metadata of the table.
@@ -101,7 +102,7 @@ impl CreateTableExecutor {
         node_manager: &NodeManagerRef,
         table_id: TableId,
         region_routes: &[RegionRoute],
-        region_wal_options: &HashMap<RegionNumber, String>,
+        region_wal_options: &RegionWalOptions,
     ) -> Result<Vec<ColumnMetadata>> {
         let storage_path =
             region_storage_path(&self.table_name.catalog_name, &self.table_name.schema_name);
@@ -124,7 +125,7 @@ impl CreateTableExecutor {
                     storage_path.clone(),
                     region_wal_options,
                     &partition_exprs,
-                );
+                )?;
                 requests.push(PbRegionRequest::Create(create_region_request));
             }
 
@@ -178,7 +179,7 @@ impl CreateTableExecutor {
         mut table_info: TableInfo,
         column_metadatas: &[ColumnMetadata],
         table_route: PhysicalTableRouteValue,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
     ) -> Result<()> {
         if !column_metadatas.is_empty() {
             update_table_info_column_ids(&mut table_info, column_metadatas);

src/common/meta/src/ddl/create_table/template.rs

@@ -25,9 +25,9 @@ use store_api::region_request::RegionRequirements;
 use store_api::storage::{RegionId, RegionNumber};
 use table::metadata::{TableId, TableInfo};
 
-use crate::error::{self, Result};
+use crate::error::{self, Result, SerializeWalOptionsSnafu};
 use crate::reconciliation::utils::build_column_metadata_from_table_info;
-use crate::wal_provider::prepare_wal_options;
+use crate::wal_provider::{RegionWalOptions, serialize_wal_options};
 
 /// Constructs a [CreateRequest] based on the provided [TableInfo].
 ///
@@ -227,16 +227,17 @@ impl CreateRequestBuilder {
         &self,
         region_id: RegionId,
         storage_path: String,
-        region_wal_options: &HashMap<RegionNumber, String>,
+        region_wal_options: &RegionWalOptions,
         partition_exprs: &HashMap<RegionNumber, String>,
-    ) -> CreateRequest {
+    ) -> Result<CreateRequest> {
         let mut request = self.template.clone();
 
         request.region_id = region_id.as_u64();
         request.path = storage_path;
         request.requirements = Some(self.requirements.into());
         // Stores the encoded wal options into the request options.
-        prepare_wal_options(&mut request.options, region_id, region_wal_options);
+        serialize_wal_options(&mut request.options, region_id, region_wal_options)
+            .context(SerializeWalOptionsSnafu { region_id })?;
         request.partition = Some(prepare_partition_expr(region_id, partition_exprs));
 
         if let Some(physical_table_id) = self.physical_table_id {
@@ -251,7 +252,7 @@ impl CreateRequestBuilder {
             );
         }
 
-        request
+        Ok(request)
     }
 }
 
@@ -299,12 +300,14 @@ mod tests {
             r#"{"Expr":{"lhs":{"Column":"a"},"op":"Eq","rhs":{"Value":{"UInt32":1}}}}"#.to_string();
         partition_exprs.insert(0, expr_a.clone());
 
-        let r0 = builder.build_one(
-            RegionId::new(42, 0),
-            "/p".to_string(),
-            &Default::default(),
-            &partition_exprs,
-        );
+        let r0 = builder
+            .build_one(
+                RegionId::new(42, 0),
+                "/p".to_string(),
+                &Default::default(),
+                &partition_exprs,
+            )
+            .unwrap();
         assert_eq!(r0.partition.as_ref().unwrap().expression, expr_a);
         assert_eq!(
             r0.requirements.map(RegionRequirements::from),
@@ -327,12 +330,14 @@ mod tests {
         let builder = CreateRequestBuilder::new(template, None)
             .with_requirements(RegionRequirements::object_storage());
 
-        let request = builder.build_one(
-            RegionId::new(42, 0),
-            "/p".to_string(),
-            &Default::default(),
-            &Default::default(),
-        );
+        let request = builder
+            .build_one(
+                RegionId::new(42, 0),
+                "/p".to_string(),
+                &Default::default(),
+                &Default::default(),
+            )
+            .unwrap();
 
         assert_eq!(
             request.requirements.map(RegionRequirements::from),

src/common/meta/src/ddl/table_meta.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
 use std::sync::Arc;
 
 use common_telemetry::{debug, info};
@@ -27,6 +26,7 @@ use crate::error::{Result, UnsupportedSnafu};
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::peer::{NoopPeerAllocator, PeerAllocContext, PeerAllocatorRef};
 use crate::rpc::ddl::CreateTableTask;
+use crate::wal_provider::RegionWalOptions;
 
 pub type TableMetadataAllocatorRef = Arc<TableMetadataAllocator>;
 
@@ -98,7 +98,7 @@ impl TableMetadataAllocator {
         &self,
         region_numbers: &[RegionNumber],
         skip_wal: bool,
-    ) -> Result<HashMap<RegionNumber, String>> {
+    ) -> Result<RegionWalOptions> {
         self.wal_options_allocator
             .allocate(region_numbers, skip_wal)
             .await

src/common/meta/src/ddl/tests/create_table.rs

@@ -26,6 +26,7 @@ use common_procedure::{Context as ProcedureContext, Procedure, ProcedureId, Stat
 use common_procedure_test::{
     MockContextProvider, execute_procedure_until, execute_procedure_until_done,
 };
+use common_wal::options::{KafkaWalOptions, WalOptions};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::ColumnSchema;
 use store_api::metadata::ColumnMetadata;
@@ -34,7 +35,7 @@ use store_api::region_engine::RegionRole;
 use store_api::storage::RegionId;
 use tokio::sync::mpsc;
 
-use crate::ddl::create_table::{CreateTableProcedure, CreateTableState};
+use crate::ddl::create_table::{CreateTableData, CreateTableProcedure, CreateTableState};
 use crate::ddl::test_util::columns::TestColumnDefBuilder;
 use crate::ddl::test_util::create_table::{
     TestCreateTableExprBuilder, build_raw_table_info_from_expr,
@@ -107,6 +108,31 @@ fn assert_create_request(
     assert_eq!(req.region_id, expected_region_id);
 }
 
+#[test]
+fn test_deserialize_legacy_create_table_data_region_wal_options() {
+    let mut json = serde_json::to_value(CreateTableData::new(
+        test_create_table_task("foo"),
+        Default::default(),
+    ))
+    .unwrap();
+    json["region_wal_options"] = serde_json::json!({
+        "0": serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(
+            "topic_a".to_string(),
+        )))
+        .unwrap(),
+    });
+
+    let data: CreateTableData = serde_json::from_value(json).unwrap();
+
+    assert_eq!(
+        data.region_wal_options.unwrap(),
+        HashMap::from([(
+            0,
+            WalOptions::Kafka(KafkaWalOptions::new("topic_a".to_string())),
+        )])
+    );
+}
+
 pub(crate) fn test_create_table_task(name: &str) -> CreateTableTask {
     let create_table = TestCreateTableExprBuilder::default()
         .column_defs([

src/common/meta/src/ddl/utils.rs

@@ -32,20 +32,19 @@ use common_error::ext::BoxedError;
 use common_procedure::error::Error as ProcedureError;
 use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::{error, info, warn};
-use common_wal::options::WalOptions;
 use futures::future::join_all;
 use snafu::{OptionExt, ResultExt, ensure};
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, MANIFEST_INFO_EXTENSION_KEY};
 use store_api::region_engine::RegionManifestInfo;
-use store_api::storage::{RegionId, RegionNumber};
+use store_api::storage::RegionId;
 use table::metadata::TableId;
 use table::table_reference::TableReference;
 
 use crate::ddl::{DdlContext, DetectingRegion};
 use crate::error::{
-    self, DecodeJsonSnafu, Error, MetadataCorruptionSnafu, OperateDatanodeSnafu,
-    ParseWalOptionsSnafu, Result, TableNotFoundSnafu, UnsupportedSnafu,
+    self, DecodeJsonSnafu, Error, MetadataCorruptionSnafu, OperateDatanodeSnafu, Result,
+    TableNotFoundSnafu, UnsupportedSnafu,
 };
 use crate::key::datanode_table::DatanodeTableValue;
 use crate::key::table_name::TableNameKey;
@@ -54,6 +53,7 @@ use crate::key::{TableMetadataManager, TableMetadataManagerRef};
 use crate::peer::Peer;
 use crate::rpc::ddl::CreateTableTask;
 use crate::rpc::router::{RegionRoute, find_follower_regions, find_followers};
+use crate::wal_provider::RegionWalOptions;
 
 /// Adds [Peer] context if the error is unretryable.
 pub fn add_peer_context_if_needed(datanode: Peer) -> impl FnOnce(Error) -> Error {
@@ -182,25 +182,12 @@ pub fn convert_region_routes_to_detecting_regions(
         .collect::<Vec<_>>()
 }
 
-/// Parses [WalOptions] from serialized strings in hashmap.
-pub fn parse_region_wal_options(
-    serialized_options: &HashMap<RegionNumber, String>,
-) -> Result<HashMap<RegionNumber, WalOptions>> {
-    let mut region_wal_options = HashMap::with_capacity(serialized_options.len());
-    for (region_number, wal_options) in serialized_options {
-        let wal_option = serde_json::from_str::<WalOptions>(wal_options)
-            .context(ParseWalOptionsSnafu { wal_options })?;
-        region_wal_options.insert(*region_number, wal_option);
-    }
-    Ok(region_wal_options)
-}
-
 /// Gets the wal options for a table.
 pub async fn get_region_wal_options(
     table_metadata_manager: &TableMetadataManager,
     table_route_value: &TableRouteValue,
     physical_table_id: TableId,
-) -> Result<HashMap<RegionNumber, WalOptions>> {
+) -> Result<RegionWalOptions> {
     let region_wal_options =
         if let TableRouteValue::Physical(table_route_value) = &table_route_value {
             let datanode_table_values = table_metadata_manager
@@ -217,12 +204,10 @@ pub async fn get_region_wal_options(
 /// Extracts region wal options from [DatanodeTableValue]s.
 pub fn extract_region_wal_options(
     datanode_table_values: &Vec<DatanodeTableValue>,
-) -> Result<HashMap<RegionNumber, WalOptions>> {
-    let mut region_wal_options = HashMap::new();
+) -> Result<RegionWalOptions> {
+    let mut region_wal_options = RegionWalOptions::new();
     for value in datanode_table_values {
-        let serialized_options = &value.region_info.region_wal_options;
-        let parsed_options = parse_region_wal_options(serialized_options)?;
-        region_wal_options.extend(parsed_options);
+        region_wal_options.extend(value.region_info.region_wal_options.clone());
     }
     Ok(region_wal_options)
 }

src/common/meta/src/error.rs

@@ -19,7 +19,6 @@ use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
 use common_procedure::ProcedureId;
-use common_wal::options::WalOptions;
 use serde_json::error::Error as JsonError;
 use snafu::{Location, Snafu};
 use store_api::storage::RegionId;
@@ -529,12 +528,9 @@ pub enum Error {
         clean_poisons: bool,
     },
 
-    #[snafu(display(
-        "Failed to encode a wal options to json string, wal_options: {:?}",
-        wal_options
-    ))]
-    EncodeWalOptions {
-        wal_options: WalOptions,
+    #[snafu(display("Failed to serialize WAL options for region: {region_id}"))]
+    SerializeWalOptions {
+        region_id: RegionId,
         #[snafu(source)]
         error: serde_json::Error,
         #[snafu(implicit)]
@@ -902,15 +898,6 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Failed to parse wal options: {}", wal_options))]
-    ParseWalOptions {
-        wal_options: String,
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: serde_json::Error,
-    },
-
     #[snafu(display("No leader found for table_id: {}", table_id))]
     NoLeader {
         table_id: TableId,
@@ -1175,7 +1162,7 @@ impl ErrorExt for Error {
             | TableRouteNotFound { .. }
             | TableRepartNotFound { .. }
             | RegionOperatingRace { .. }
-            | EncodeWalOptions { .. }
+            | SerializeWalOptions { .. }
             | BuildKafkaClient { .. }
             | BuildKafkaCtrlClient { .. }
             | KafkaPartitionClient { .. }
@@ -1186,7 +1173,6 @@ impl ErrorExt for Error {
             | ProcedureOutput { .. }
             | FromUtf8 { .. }
             | MetadataCorruption { .. }
-            | ParseWalOptions { .. }
             | KafkaGetOffset { .. }
             | ReadFlexbuffers { .. }
             | SerializeFlexbuffers { .. }

src/common/meta/src/instruction.rs

@@ -28,6 +28,7 @@ use crate::flow_name::FlowName;
 use crate::key::schema_name::SchemaName;
 use crate::key::{FlowId, FlowPartitionId};
 use crate::peer::Peer;
+use crate::wal_provider::{RegionWalOptions, region_wal_options_serde};
 use crate::{DatanodeId, FlownodeId};
 
 #[derive(Eq, Hash, PartialEq, Clone, Debug, Serialize, Deserialize)]
@@ -204,8 +205,8 @@ pub struct OpenRegion {
     pub region_storage_path: String,
     pub region_options: HashMap<String, String>,
     #[serde(default)]
-    #[serde_as(as = "HashMap<serde_with::DisplayFromStr, _>")]
-    pub region_wal_options: HashMap<RegionNumber, String>,
+    #[serde(with = "region_wal_options_serde")]
+    pub region_wal_options: RegionWalOptions,
     #[serde(default)]
     pub skip_wal_replay: bool,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -219,7 +220,7 @@ impl OpenRegion {
         region_ident: RegionIdent,
         path: &str,
         region_options: HashMap<String, String>,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
         skip_wal_replay: bool,
         reason: Option<OpenRegionReason>,
         requirements: RegionRequirements,
@@ -1129,6 +1130,7 @@ impl InstructionReply {
 mod tests {
     use std::collections::HashSet;
 
+    use common_wal::options::WalOptions;
     use store_api::storage::{FileId, FileRef};
 
     use super::*;
@@ -1398,6 +1400,18 @@ mod tests {
         assert_eq!(expected, deserialized);
     }
 
+    #[test]
+    fn test_deserialize_open_region_with_legacy_region_wal_options() {
+        let open_region = r#"{"region_ident":{"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"},"region_storage_path":"test/foo","region_options":{},"region_wal_options":{"1":"{\"wal.provider\":\"raft_engine\"}"},"skip_wal_replay":false}"#;
+
+        let open_region: OpenRegion = serde_json::from_str(open_region).unwrap();
+
+        assert_eq!(
+            open_region.region_wal_options,
+            HashMap::from([(1, WalOptions::RaftEngine)])
+        );
+    }
+
     #[test]
     fn test_serialize_open_region_with_reason_and_requirements() {
         let open_region = OpenRegion::new(

src/common/meta/src/key.rs

@@ -167,6 +167,7 @@ use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::rpc::router::{LeaderState, RegionRoute, region_distribution};
 use crate::rpc::store::BatchDeleteRequest;
 use crate::state_store::PoisonValue;
+use crate::wal_provider::RegionWalOptions;
 
 pub const TOPIC_NAME_PATTERN: &str = r"[a-zA-Z0-9_:-][a-zA-Z0-9_:\-\.@#]*";
 pub const LEGACY_MAINTENANCE_KEY: &str = "__maintenance";
@@ -778,7 +779,7 @@ impl TableMetadataManager {
         &self,
         table_info: TableInfo,
         table_route_value: TableRouteValue,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
     ) -> Result<()> {
         let table_id = table_info.ident.table_id;
         let engine = table_info.meta.engine.clone();
@@ -1175,10 +1176,7 @@ impl TableMetadataManager {
             let datanode_table_value = DatanodeTableValue::try_from_raw_value(&kv.value)?;
             for (region_number, wal_options) in &datanode_table_value.region_info.region_wal_options
             {
-                region_wal_options.insert(
-                    *region_number,
-                    serde_json::from_str(wal_options).context(error::SerdeJsonSnafu)?,
-                );
+                region_wal_options.insert(*region_number, wal_options.clone());
             }
         }
 
@@ -1435,7 +1433,7 @@ impl TableMetadataManager {
         current_table_route_value: &DeserializedValueWithBytes<TableRouteValue>,
         new_region_routes: Vec<RegionRoute>,
         new_region_options: &HashMap<String, String>,
-        new_region_wal_options: &HashMap<RegionNumber, String>,
+        new_region_wal_options: &RegionWalOptions,
     ) -> Result<()> {
         // Updates the datanode table key value pairs.
         let current_region_distribution =
@@ -1664,7 +1662,7 @@ mod tests {
     use crate::peer::Peer;
     use crate::rpc::router::{LeaderState, Region, RegionRoute, region_distribution};
     use crate::rpc::store::{PutRequest, RangeRequest};
-    use crate::wal_provider::WalProvider;
+    use crate::wal_provider::{RegionWalOptions, WalProvider};
 
     #[test]
     fn test_deserialized_value_with_bytes() {
@@ -1737,7 +1735,7 @@ mod tests {
         table_metadata_manager: &TableMetadataManager,
         table_info: TableInfo,
         region_routes: Vec<RegionRoute>,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
     ) -> Result<()> {
         table_metadata_manager
             .create_table_metadata(
@@ -1754,11 +1752,7 @@ mod tests {
             .collect::<Vec<_>>();
         let wal_options = topics
             .iter()
-            .map(|topic| {
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: topic.clone(),
-                })
-            })
+            .map(|topic| WalOptions::Kafka(KafkaWalOptions::new(topic.clone())))
             .collect::<Vec<_>>();
 
         (0..16)
@@ -1771,17 +1765,13 @@ mod tests {
         HashMap::from([
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "greptimedb_topic0".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("greptimedb_topic0".to_string())),
             ),
             (1, WalOptions::RaftEngine),
             (2, WalOptions::Noop),
             (
                 3,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "greptimedb_topic1".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("greptimedb_topic1".to_string())),
             ),
         ])
     }
@@ -1819,10 +1809,7 @@ mod tests {
         let region_route = new_test_region_route();
         let region_routes = &vec![region_route.clone()];
         let table_info = new_test_table_info();
-        let region_wal_options = create_mock_region_wal_options()
-            .into_iter()
-            .map(|(k, v)| (k, serde_json::to_string(&v).unwrap()))
-            .collect::<HashMap<_, _>>();
+        let region_wal_options = create_mock_region_wal_options();
 
         // creates metadata.
         create_physical_table_metadata(
@@ -2083,10 +2070,7 @@ mod tests {
         let table_id = table_info.ident.table_id;
         let datanode_id = 2;
         let region_wal_options = create_mock_region_wal_options();
-        let serialized_region_wal_options = region_wal_options
-            .iter()
-            .map(|(k, v)| (*k, serde_json::to_string(v).unwrap()))
-            .collect::<HashMap<_, _>>();
+        let serialized_region_wal_options = region_wal_options.clone();
 
         // creates metadata.
         create_physical_table_metadata(
@@ -2561,20 +2545,14 @@ mod tests {
             region_storage_path(&table_info.catalog_name, &table_info.schema_name);
 
         // Create initial metadata with Kafka WAL options
-        let old_region_wal_options: HashMap<RegionNumber, String> = vec![
+        let old_region_wal_options: RegionWalOptions = vec![
             (
                 1,
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(),
-                }))
-                .unwrap(),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())),
             ),
             (
                 2,
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_2".to_string(),
-                }))
-                .unwrap(),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_2".to_string())),
             ),
         ]
         .into_iter()
@@ -2621,27 +2599,18 @@ mod tests {
             new_region_route(2, 2),
             new_region_route(3, 3), // New region
         ];
-        let new_region_wal_options: HashMap<RegionNumber, String> = vec![
+        let new_region_wal_options: RegionWalOptions = vec![
             (
                 1,
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(), // Unchanged
-                }))
-                .unwrap(),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())), // Unchanged
             ),
             (
                 2,
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_2".to_string(), // Unchanged
-                }))
-                .unwrap(),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_2".to_string())), // Unchanged
             ),
             (
                 3,
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_3".to_string(), // New topic
-                }))
-                .unwrap(),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_3".to_string())), // New topic
             ),
         ]
         .into_iter()
@@ -2685,20 +2654,14 @@ mod tests {
             // Region 2 removed
             // Region 3 now has different topic
         ];
-        let newer_region_wal_options: HashMap<RegionNumber, String> = vec![
+        let newer_region_wal_options: RegionWalOptions = vec![
             (
                 1,
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(), // Unchanged
-                }))
-                .unwrap(),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())), // Unchanged
             ),
             (
                 3,
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_3_new".to_string(), // Changed topic
-                }))
-                .unwrap(),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_3_new".to_string())), // Changed topic
             ),
         ]
         .into_iter()
@@ -2768,10 +2731,7 @@ mod tests {
         let table_name = "foo";
         let task = test_create_table_task(table_name, table_id);
         let options = create_mixed_region_wal_options();
-        let serialized_options = options
-            .iter()
-            .map(|(k, v)| (*k, serde_json::to_string(v).unwrap()))
-            .collect::<HashMap<_, _>>();
+        let serialized_options = options.clone();
         table_metadata_manager
             .create_table_metadata(
                 task.table_info,
@@ -2828,10 +2788,7 @@ mod tests {
         let table_name = "foo";
         let task = test_create_table_task(table_name, table_id);
         let options = create_mixed_region_wal_options();
-        let serialized_options = options
-            .iter()
-            .map(|(k, v)| (*k, serde_json::to_string(v).unwrap()))
-            .collect::<HashMap<_, _>>();
+        let serialized_options = options.clone();
         table_metadata_manager
             .create_table_metadata(
                 task.table_info,
@@ -2904,10 +2861,7 @@ mod tests {
         let task = test_create_table_task(table_name, table_id);
         let table_info = task.table_info.clone();
         let options = create_mixed_region_wal_options();
-        let serialized_options = options
-            .iter()
-            .map(|(k, v)| (*k, serde_json::to_string(v).unwrap()))
-            .collect::<HashMap<_, _>>();
+        let serialized_options = options.clone();
         table_metadata_manager
             .create_table_metadata(
                 table_info.clone(),
@@ -3004,10 +2958,7 @@ mod tests {
         let dropped_task = test_create_table_task(table_name, dropped_table_id);
         let dropped_table_info = dropped_task.table_info.clone();
         let options = create_mock_region_wal_options();
-        let serialized_options = options
-            .iter()
-            .map(|(k, v)| (*k, serde_json::to_string(v).unwrap()))
-            .collect::<HashMap<_, _>>();
+        let serialized_options = options.clone();
         table_metadata_manager
             .create_table_metadata(
                 dropped_table_info.clone(),
@@ -3110,10 +3061,7 @@ mod tests {
         let task = test_create_table_task(table_name, table_id);
         let table_info = task.table_info.clone();
         let options = create_mixed_region_wal_options();
-        let serialized_options = options
-            .iter()
-            .map(|(k, v)| (*k, serde_json::to_string(v).unwrap()))
-            .collect::<HashMap<_, _>>();
+        let serialized_options = options.clone();
         table_metadata_manager
             .create_table_metadata(
                 table_info.clone(),

src/common/meta/src/key/datanode_table.rs

@@ -34,6 +34,7 @@ use crate::range_stream::{DEFAULT_PAGE_SIZE, PaginationStream};
 use crate::rpc::KeyValue;
 use crate::rpc::router::region_distribution;
 use crate::rpc::store::{BatchGetRequest, RangeRequest};
+use crate::wal_provider::{RegionWalOptions, region_wal_options_serde};
 
 #[serde_with::serde_as]
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
@@ -50,10 +51,10 @@ pub struct RegionInfo {
     #[serde(default)]
     pub region_options: HashMap<String, String>,
     /// The per-region wal options.
-    /// Key: region number. Value: the encoded wal options of the region.
+    /// Key: region number. Value: the wal options of the region.
     #[serde(default)]
-    #[serde_as(as = "HashMap<serde_with::DisplayFromStr, _>")]
-    pub region_wal_options: HashMap<RegionNumber, String>,
+    #[serde(with = "region_wal_options_serde")]
+    pub region_wal_options: RegionWalOptions,
 }
 
 /// The key mapping {datanode_id} to {table_id}
@@ -228,7 +229,7 @@ impl DatanodeTableManager {
         engine: &str,
         region_storage_path: &str,
         region_options: HashMap<String, String>,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
         distribution: RegionDistribution,
     ) -> Result<Txn> {
         let txns = distribution
@@ -313,7 +314,7 @@ impl DatanodeTableManager {
         current_region_distribution: RegionDistribution,
         new_region_distribution: RegionDistribution,
         new_region_options: &HashMap<String, String>,
-        new_region_wal_options: &HashMap<RegionNumber, String>,
+        new_region_wal_options: &RegionWalOptions,
     ) -> Result<Txn> {
         let mut opts = Vec::new();
 
@@ -386,6 +387,8 @@ impl DatanodeTableManager {
 
 #[cfg(test)]
 mod tests {
+    use common_wal::options::WalOptions;
+
     use super::*;
 
     #[test]
@@ -487,9 +490,9 @@ mod tests {
                 ("c".to_string(), "cc".to_string()),
             ]),
             region_wal_options: HashMap::from([
-                (1, "aaa".to_string()),
-                (2, "bbb".to_string()),
-                (3, "ccc".to_string()),
+                (1, WalOptions::RaftEngine),
+                (2, WalOptions::Noop),
+                (3, WalOptions::RaftEngine),
             ]),
         };
         let table_value = DatanodeTableValue {
@@ -509,6 +512,18 @@ mod tests {
         assert_eq!(table_value, decoded);
     }
 
+    #[test]
+    fn test_deserialize_legacy_region_wal_options() {
+        let literal = br#"{"table_id":42,"regions":[1],"follower_regions":[],"engine":"","region_storage_path":"","region_options":{},"region_wal_options":{"1":"{\"wal.provider\":\"raft_engine\"}"},"version":1}"#;
+
+        let actual = DatanodeTableValue::try_from_raw_value(literal).unwrap();
+
+        assert_eq!(
+            actual.region_info.region_wal_options,
+            HashMap::from([(1, WalOptions::RaftEngine)])
+        );
+    }
+
     #[test]
     fn test_deserialization() {
         fn test_err(raw_key: &[u8]) {

src/common/meta/src/key/topic_region.rs

@@ -18,10 +18,9 @@ use std::fmt::{self, Display};
 use common_wal::options::WalOptions;
 use serde::{Deserialize, Serialize};
 use snafu::OptionExt;
-use store_api::storage::{RegionId, RegionNumber};
+use store_api::storage::RegionId;
 use table::metadata::TableId;
 
-use crate::ddl::utils::parse_region_wal_options;
 use crate::error::{Error, InvalidMetadataSnafu, Result};
 use crate::key::{MetadataKey, MetadataValue, TOPIC_REGION_PATTERN, TOPIC_REGION_PREFIX};
 use crate::kv_backend::KvBackendRef;
@@ -30,6 +29,7 @@ use crate::rpc::KeyValue;
 use crate::rpc::store::{
     BatchDeleteRequest, BatchGetRequest, BatchPutRequest, PutRequest, RangeRequest,
 };
+use crate::wal_provider::RegionWalOptions;
 
 // The TopicRegionKey is a key for the topic-region mapping in the kvbackend.
 // The layout of the key is `__topic_region/{topic_name}/{region_id}`.
@@ -265,10 +265,9 @@ impl TopicRegionManager {
     pub fn build_create_txn(
         &self,
         table_id: TableId,
-        region_wal_options: &HashMap<RegionNumber, String>,
+        region_wal_options: &RegionWalOptions,
     ) -> Result<Txn> {
-        let region_wal_options = parse_region_wal_options(region_wal_options)?;
-        let topic_region_mapping = self.get_topic_region_mapping(table_id, &region_wal_options);
+        let topic_region_mapping = self.get_topic_region_mapping(table_id, region_wal_options);
         let topic_region_keys = topic_region_mapping
             .iter()
             .map(|(region_id, topic)| TopicRegionKey::new(*region_id, topic))
@@ -284,13 +283,11 @@ impl TopicRegionManager {
     pub fn build_update_txn(
         &self,
         table_id: TableId,
-        old_region_wal_options: &HashMap<RegionNumber, String>,
-        new_region_wal_options: &HashMap<RegionNumber, String>,
+        old_region_wal_options: &RegionWalOptions,
+        new_region_wal_options: &RegionWalOptions,
     ) -> Result<Txn> {
-        let old_wal_options_parsed = parse_region_wal_options(old_region_wal_options)?;
-        let new_wal_options_parsed = parse_region_wal_options(new_region_wal_options)?;
-        let old_mapping = self.get_topic_region_mapping(table_id, &old_wal_options_parsed);
-        let new_mapping = self.get_topic_region_mapping(table_id, &new_wal_options_parsed);
+        let old_mapping = self.get_topic_region_mapping(table_id, old_region_wal_options);
+        let new_mapping = self.get_topic_region_mapping(table_id, new_region_wal_options);
 
         // Convert to HashMap for easier lookup: RegionId -> Topic
         let old_map: HashMap<RegionId, &str> = old_mapping.into_iter().collect();
@@ -369,7 +366,7 @@ impl TopicRegionManager {
     pub fn get_topic_region_mapping<'a>(
         &self,
         table_id: TableId,
-        region_wal_options: &'a HashMap<RegionNumber, WalOptions>,
+        region_wal_options: &'a RegionWalOptions,
     ) -> Vec<(RegionId, &'a str)> {
         region_wal_options
             .keys()
@@ -516,17 +513,14 @@ mod tests {
             .map(|i| {
                 let region_number = i;
                 let wal_options = if i % 2 == 0 {
-                    WalOptions::Kafka(KafkaWalOptions {
-                        topic: format!("topic_{}", i),
-                    })
+                    WalOptions::Kafka(KafkaWalOptions::new(format!("topic_{}", i)))
                 } else {
                     WalOptions::RaftEngine
                 };
-                (region_number, serde_json::to_string(&wal_options).unwrap())
+                (region_number, wal_options)
             })
             .collect::<HashMap<_, _>>();
 
-        let region_wal_options = parse_region_wal_options(&region_wal_options).unwrap();
         let mut topic_region_mapping =
             manager.get_topic_region_mapping(table_id, &region_wal_options);
         let mut expected = (0..64)
@@ -569,20 +563,15 @@ mod tests {
         let region_wal_options = vec![
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_0".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
             ),
             (
                 1,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())),
             ),
             (2, WalOptions::RaftEngine), // Should be ignored
         ]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
 
         let txn = manager
@@ -628,29 +617,21 @@ mod tests {
         let table_id = 1;
         let old_region_wal_options = vec![(
             0,
-            WalOptions::Kafka(KafkaWalOptions {
-                topic: "topic_0".to_string(),
-            }),
+            WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
         )]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let new_region_wal_options = vec![
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_0".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
             ),
             (
                 1,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())),
             ),
         ]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let txn = manager
             .build_update_txn(table_id, &old_region_wal_options, &new_region_wal_options)
@@ -675,28 +656,20 @@ mod tests {
         let old_region_wal_options = vec![
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_0".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
             ),
             (
                 1,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())),
             ),
         ]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let new_region_wal_options = vec![(
             0,
-            WalOptions::Kafka(KafkaWalOptions {
-                topic: "topic_0".to_string(),
-            }),
+            WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
         )]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let txn = manager
             .build_update_txn(table_id, &old_region_wal_options, &new_region_wal_options)
@@ -723,21 +696,15 @@ mod tests {
         let table_id = 1;
         let old_region_wal_options = vec![(
             0,
-            WalOptions::Kafka(KafkaWalOptions {
-                topic: "topic_0".to_string(),
-            }),
+            WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
         )]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let new_region_wal_options = vec![(
             0,
-            WalOptions::Kafka(KafkaWalOptions {
-                topic: "topic_0_new".to_string(),
-            }),
+            WalOptions::Kafka(KafkaWalOptions::new("topic_0_new".to_string())),
         )]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let txn = manager
             .build_update_txn(table_id, &old_region_wal_options, &new_region_wal_options)
@@ -780,19 +747,14 @@ mod tests {
         let region_wal_options = vec![
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_0".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
             ),
             (
                 1,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())),
             ),
         ]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let txn = manager
             .build_update_txn(table_id, &region_wal_options, &region_wal_options)
@@ -810,49 +772,35 @@ mod tests {
         let old_region_wal_options = vec![
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_0".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
             ),
             (
                 1,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())),
             ),
             (
                 2,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_2".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_2".to_string())),
             ),
         ]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let new_region_wal_options = vec![
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_0".to_string(), // Unchanged
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())), // Unchanged
             ),
             (
                 1,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1_new".to_string(), // Topic changed
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1_new".to_string())), // Topic changed
             ),
             // Region 2 removed
             (
                 3,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_3".to_string(), // New region
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_3".to_string())), // New region
             ),
         ]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let txn = manager
             .build_update_txn(table_id, &old_region_wal_options, &new_region_wal_options)
@@ -931,31 +879,23 @@ mod tests {
         let old_region_wal_options = vec![
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_0".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
             ),
             (1, WalOptions::RaftEngine), // Should be ignored
         ]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let new_region_wal_options = vec![
             (
                 0,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_0".to_string(),
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_0".to_string())),
             ),
             (
                 1,
-                WalOptions::Kafka(KafkaWalOptions {
-                    topic: "topic_1".to_string(), // Changed from RaftEngine to Kafka
-                }),
+                WalOptions::Kafka(KafkaWalOptions::new("topic_1".to_string())), // Changed from RaftEngine to Kafka
             ),
         ]
         .into_iter()
-        .map(|(num, opts)| (num, serde_json::to_string(&opts).unwrap()))
         .collect::<HashMap<_, _>>();
         let txn = manager
             .build_update_txn(table_id, &old_region_wal_options, &new_region_wal_options)

src/common/meta/src/reconciliation/reconcile_logical_tables/reconcile_regions.rs

@@ -128,7 +128,7 @@ impl ReconcileRegions {
                     storage_path.clone(),
                     &HashMap::new(),
                     &partition_exprs,
-                );
+                )?;
                 requests.push(one_region_request);
             }
         }

src/common/meta/src/wal_provider.rs

@@ -17,23 +17,193 @@ pub(crate) mod topic_creator;
 mod topic_manager;
 pub(crate) mod topic_pool;
 
-use std::collections::HashMap;
+use std::collections::{BTreeSet, HashMap};
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use common_procedure::Context as ProcedureContext;
+use common_procedure::local::DynamicKeyLockGuard;
 use common_wal::config::MetasrvWalConfig;
 use common_wal::options::{KafkaWalOptions, WAL_OPTIONS_KEY, WalOptions};
-use snafu::{ResultExt, ensure};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use snafu::ensure;
 use store_api::storage::{RegionId, RegionNumber};
 
 use crate::ddl::allocator::wal_options::WalOptionsAllocator;
-use crate::error::{EncodeWalOptionsSnafu, InvalidTopicNamePrefixSnafu, Result};
-use crate::key::TOPIC_NAME_PATTERN_REGEX;
+use crate::error::{InvalidTopicNamePrefixSnafu, Result};
+use crate::key::topic_name::TopicNameKey;
+use crate::key::{TOPIC_NAME_PATTERN_REGEX, TableMetadataManagerRef};
 use crate::kv_backend::KvBackendRef;
 use crate::leadership_notifier::LeadershipChangeListener;
+use crate::lock_key::RemoteWalLock;
 pub use crate::wal_provider::topic_creator::{build_kafka_client, build_kafka_topic_creator};
 use crate::wal_provider::topic_pool::KafkaTopicPool;
 
+/// WAL options allocated for each region.
+pub type RegionWalOptions = HashMap<RegionNumber, WalOptions>;
+
+/// Returns remote WAL topics referenced by region WAL options.
+pub fn remote_wal_topics(region_wal_options: &RegionWalOptions) -> Vec<&str> {
+    region_wal_options
+        .values()
+        .filter_map(|wal_options| match wal_options {
+            WalOptions::Kafka(kafka_options) => Some(kafka_options.topic.as_str()),
+            _ => None,
+        })
+        .collect::<BTreeSet<_>>()
+        .into_iter()
+        .collect()
+}
+
+/// Acquires per-topic read locks for remote WAL topics.
+pub async fn acquire_remote_wal_read_locks(
+    ctx: &ProcedureContext,
+    region_wal_options: &RegionWalOptions,
+) -> Vec<DynamicKeyLockGuard> {
+    let topics = remote_wal_topics(region_wal_options)
+        .into_iter()
+        .map(str::to_string)
+        .collect::<Vec<_>>();
+    let mut guards = Vec::with_capacity(topics.len());
+    for topic in topics {
+        let guard = ctx
+            .provider
+            .acquire_lock(&(RemoteWalLock::Read(topic).into()))
+            .await;
+        guards.push(guard);
+    }
+    guards
+}
+
+/// Refreshes initial pruned entry ids for Kafka WAL options.
+pub async fn refresh_initial_pruned_entry_ids(
+    table_metadata_manager: &TableMetadataManagerRef,
+    region_wal_options: &mut RegionWalOptions,
+) -> Result<()> {
+    let topics = remote_wal_topics(region_wal_options);
+    if topics.is_empty() {
+        return Ok(());
+    }
+
+    let topic_values = table_metadata_manager
+        .topic_name_manager()
+        .batch_get(
+            topics
+                .iter()
+                .map(|topic| TopicNameKey::new(topic))
+                .collect(),
+        )
+        .await?;
+
+    for wal_options in region_wal_options.values_mut() {
+        let WalOptions::Kafka(kafka_options) = wal_options else {
+            continue;
+        };
+        kafka_options.initial_pruned_entry_id = Some(
+            topic_values
+                .get(&kafka_options.topic)
+                .map(|value| value.pruned_entry_id)
+                .unwrap_or_default(),
+        );
+    }
+
+    Ok(())
+}
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum WalOptionsCompat {
+    Encoded(String),
+    Structured(WalOptions),
+}
+
+fn deserialize_region_wal_options<E>(
+    values: HashMap<String, WalOptionsCompat>,
+) -> std::result::Result<RegionWalOptions, E>
+where
+    E: serde::de::Error,
+{
+    values
+        .into_iter()
+        .map(|(region_number, wal_options)| {
+            let region_number = region_number.parse::<RegionNumber>().map_err(|err| {
+                E::custom(format!(
+                    "invalid region number in region_wal_options: {region_number}, err: {err}"
+                ))
+            })?;
+            let wal_options = match wal_options {
+                WalOptionsCompat::Encoded(encoded) => serde_json::from_str(&encoded).map_err(|err| {
+                    E::custom(format!(
+                        "failed to decode legacy wal options for region {region_number}: {encoded}, err: {err}"
+                    ))
+                })?,
+                WalOptionsCompat::Structured(wal_options) => wal_options,
+            };
+            Ok((region_number, wal_options))
+        })
+        .collect()
+}
+
+/// Serde helpers for [`RegionWalOptions`] persisted in metadata.
+///
+/// New metadata stores WAL options as structured JSON objects. The deserializer
+/// also accepts the legacy format whose map values are JSON strings encoded from
+/// [`WalOptions`].
+pub mod region_wal_options_serde {
+    use super::*;
+
+    /// Serializes region WAL options in structured form.
+    pub fn serialize<S>(
+        value: &RegionWalOptions,
+        serializer: S,
+    ) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        value.serialize(serializer)
+    }
+
+    /// Deserializes region WAL options from either structured or legacy encoded form.
+    pub fn deserialize<'de, D>(deserializer: D) -> std::result::Result<RegionWalOptions, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let values = HashMap::<String, WalOptionsCompat>::deserialize(deserializer)?;
+        deserialize_region_wal_options(values)
+    }
+}
+
+/// Serde helpers for optional [`RegionWalOptions`] persisted in procedure state.
+pub mod optional_region_wal_options_serde {
+    use super::*;
+
+    /// Serializes optional region WAL options in structured form.
+    pub fn serialize<S>(
+        value: &Option<RegionWalOptions>,
+        serializer: S,
+    ) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        value.serialize(serializer)
+    }
+
+    /// Deserializes optional region WAL options from structured or legacy encoded form.
+    pub fn deserialize<'de, D>(
+        deserializer: D,
+    ) -> std::result::Result<Option<RegionWalOptions>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let Some(values) = Option::<HashMap<String, WalOptionsCompat>>::deserialize(deserializer)?
+        else {
+            return Ok(None);
+        };
+
+        deserialize_region_wal_options(values).map(Some)
+    }
+}
+
 /// Provides wal options in region granularity.
 #[derive(Default, Debug)]
 pub enum WalProvider {
@@ -51,14 +221,8 @@ impl WalOptionsAllocator for WalProvider {
         &self,
         region_numbers: &[RegionNumber],
         skip_wal: bool,
-    ) -> Result<HashMap<RegionNumber, String>> {
-        let wal_options = self
-            .alloc_batch(region_numbers.len(), skip_wal)?
-            .into_iter()
-            .map(|wal_options| {
-                serde_json::to_string(&wal_options).context(EncodeWalOptionsSnafu { wal_options })
-            })
-            .collect::<Result<Vec<_>>>()?;
+    ) -> Result<RegionWalOptions> {
+        let wal_options = self.alloc_batch(region_numbers.len(), skip_wal).await?;
 
         Ok(region_numbers.iter().copied().zip(wal_options).collect())
     }
@@ -75,7 +239,7 @@ impl WalProvider {
 
     /// Allocates a batch of wal options where each wal options goes to a region.
     /// If skip_wal is true, the wal options will be set to Noop regardless of the provider type.
-    pub fn alloc_batch(&self, num_regions: usize, skip_wal: bool) -> Result<Vec<WalOptions>> {
+    pub async fn alloc_batch(&self, num_regions: usize, skip_wal: bool) -> Result<Vec<WalOptions>> {
         if skip_wal {
             return Ok(vec![WalOptions::Noop; num_regions]);
         }
@@ -85,11 +249,7 @@ impl WalProvider {
                 let options_batch = topic_manager
                     .select_batch(num_regions)?
                     .into_iter()
-                    .map(|topic| {
-                        WalOptions::Kafka(KafkaWalOptions {
-                            topic: topic.clone(),
-                        })
-                    })
+                    .map(|topic| WalOptions::Kafka(KafkaWalOptions::new(topic.clone())))
                     .collect();
                 Ok(options_batch)
             }
@@ -139,34 +299,29 @@ pub async fn build_wal_provider(
     }
 }
 
-/// Inserts wal options into options.
-pub fn prepare_wal_options(
+/// Serializes and inserts WAL options into the region options.
+pub fn serialize_wal_options(
     options: &mut HashMap<String, String>,
     region_id: RegionId,
-    region_wal_options: &HashMap<RegionNumber, String>,
-) {
+    region_wal_options: &RegionWalOptions,
+) -> std::result::Result<(), serde_json::Error> {
     if let Some(wal_options) = region_wal_options.get(&region_id.region_number()) {
-        options.insert(WAL_OPTIONS_KEY.to_string(), wal_options.clone());
+        let encoded = serde_json::to_string(wal_options)?;
+        options.insert(WAL_OPTIONS_KEY.to_string(), encoded);
     }
+    Ok(())
 }
 
 /// Extracts the topic from the wal options.
 pub fn extract_topic_from_wal_options(
     region_id: RegionId,
-    region_options: &HashMap<RegionNumber, String>,
+    region_options: &RegionWalOptions,
 ) -> Option<String> {
     region_options
         .get(&region_id.region_number())
-        .and_then(|wal_options| {
-            serde_json::from_str::<WalOptions>(wal_options)
-                .ok()
-                .and_then(|wal_options| {
-                    if let WalOptions::Kafka(kafka_wal_option) = wal_options {
-                        Some(kafka_wal_option.topic)
-                    } else {
-                        None
-                    }
-                })
+        .and_then(|wal_options| match wal_options {
+            WalOptions::Kafka(kafka_wal_option) => Some(kafka_wal_option.topic.clone()),
+            _ => None,
         })
 }
 
@@ -197,10 +352,9 @@ mod tests {
         let regions = (0..num_regions).collect::<Vec<_>>();
         let got = provider.allocate(&regions, false).await.unwrap();
 
-        let encoded_wal_options = serde_json::to_string(&WalOptions::RaftEngine).unwrap();
         let expected = regions
             .into_iter()
-            .zip(vec![encoded_wal_options; num_regions as usize])
+            .zip(vec![WalOptions::RaftEngine; num_regions as usize])
             .collect();
         assert_eq!(got, expected);
     }
@@ -250,15 +404,85 @@ mod tests {
         // Check the allocated wal options contain the expected topics.
         let expected = (0..num_regions)
             .map(|i| {
-                let options = WalOptions::Kafka(KafkaWalOptions {
-                    topic: topics[i as usize].clone(),
-                });
-                (i, serde_json::to_string(&options).unwrap())
+                let options = WalOptions::Kafka(KafkaWalOptions::new(topics[i as usize].clone()));
+                (i, options)
             })
             .collect::<HashMap<_, _>>();
         assert_eq!(got, expected);
     }
 
+    #[derive(Debug, PartialEq, Serialize, Deserialize)]
+    struct RegionWalOptionsWrapper {
+        #[serde(with = "region_wal_options_serde")]
+        region_wal_options: RegionWalOptions,
+    }
+
+    #[test]
+    fn test_deserialize_legacy_region_wal_options_from_encoded_map() {
+        let legacy_region_wal_options = HashMap::from([
+            (1, serde_json::to_string(&WalOptions::RaftEngine).unwrap()),
+            (
+                2,
+                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(
+                    "topic_a".to_string(),
+                )))
+                .unwrap(),
+            ),
+        ]);
+        let legacy_json = serde_json::json!({
+            "region_wal_options": legacy_region_wal_options,
+        });
+
+        assert_eq!(
+            legacy_json.to_string(),
+            r#"{"region_wal_options":{"1":"{\"wal.provider\":\"raft_engine\"}","2":"{\"wal.provider\":\"kafka\",\"wal.kafka.topic\":\"topic_a\"}"}}"#
+        );
+
+        let decoded: RegionWalOptionsWrapper = serde_json::from_value(legacy_json).unwrap();
+
+        assert_eq!(
+            decoded.region_wal_options,
+            HashMap::from([
+                (1, WalOptions::RaftEngine),
+                (
+                    2,
+                    WalOptions::Kafka(KafkaWalOptions::new("topic_a".to_string())),
+                ),
+            ])
+        );
+    }
+
+    #[test]
+    fn test_deserialize_structured_region_wal_options() {
+        let json = r#"{
+            "region_wal_options": {
+                "1": {"wal.provider":"raft_engine"},
+                "2": {"wal.provider":"noop"}
+            }
+        }"#;
+
+        let decoded: RegionWalOptionsWrapper = serde_json::from_str(json).unwrap();
+
+        assert_eq!(
+            decoded.region_wal_options,
+            HashMap::from([(1, WalOptions::RaftEngine), (2, WalOptions::Noop)])
+        );
+    }
+
+    #[test]
+    fn test_serialize_structured_region_wal_options() {
+        let wrapper = RegionWalOptionsWrapper {
+            region_wal_options: HashMap::from([(1, WalOptions::RaftEngine)]),
+        };
+
+        let encoded = serde_json::to_string(&wrapper).unwrap();
+
+        assert_eq!(
+            encoded,
+            r#"{"region_wal_options":{"1":{"wal.provider":"raft_engine"}}}"#
+        );
+    }
+
     #[tokio::test]
     async fn test_provider_with_skip_wal() {
         let provider = WalProvider::RaftEngine;
@@ -269,7 +493,7 @@ mod tests {
         let got = provider.allocate(&regions, true).await.unwrap();
         assert_eq!(got.len(), num_regions as usize);
         for wal_options in got.values() {
-            assert_eq!(wal_options, &"{\"wal.provider\":\"noop\"}");
+            assert_eq!(wal_options, &WalOptions::Noop);
         }
     }
 }

src/common/wal/src/options.rs

@@ -54,9 +54,7 @@ mod tests {
         assert_eq!(decoded, wal_options);
 
         // Test serde kafka wal options.
-        let wal_options = WalOptions::Kafka(KafkaWalOptions {
-            topic: "test_topic".to_string(),
-        });
+        let wal_options = WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string()));
         let encoded = serde_json::to_string(&wal_options).unwrap();
         let expected = r#"{"wal.provider":"kafka","wal.kafka.topic":"test_topic"}"#;
         assert_eq!(&encoded, expected);
@@ -64,6 +62,25 @@ mod tests {
         let decoded: WalOptions = serde_json::from_str(&encoded).unwrap();
         assert_eq!(decoded, wal_options);
 
+        let wal_options = WalOptions::Kafka(KafkaWalOptions {
+            topic: "test_topic".to_string(),
+            initial_pruned_entry_id: Some(42),
+        });
+        let encoded = serde_json::to_string(&wal_options).unwrap();
+        let expected = r#"{"wal.provider":"kafka","wal.kafka.topic":"test_topic","wal.kafka.initial_pruned_entry_id":42}"#;
+        assert_eq!(&encoded, expected);
+
+        let decoded: WalOptions = serde_json::from_str(&encoded).unwrap();
+        assert_eq!(decoded, wal_options);
+
+        let decoded: WalOptions =
+            serde_json::from_str(r#"{"wal.provider":"kafka","wal.kafka.topic":"test_topic"}"#)
+                .unwrap();
+        assert_eq!(
+            decoded,
+            WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string()))
+        );
+
         // Test serde noop wal options.
         let wal_options = WalOptions::Noop;
         let encoded = serde_json::to_string(&wal_options).unwrap();

src/common/wal/src/options/kafka.rs

@@ -19,4 +19,20 @@ use serde::{Deserialize, Serialize};
 pub struct KafkaWalOptions {
     /// Kafka wal topic.
     pub topic: String,
+    /// Initial pruned entry id of the topic when this option is allocated.
+    ///
+    /// This is a create-time hint for initializing a new region's flushed entry id,
+    /// not the authoritative latest pruned entry id of the topic.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub initial_pruned_entry_id: Option<u64>,
+}
+
+impl KafkaWalOptions {
+    /// Creates kafka WAL options with the topic only.
+    pub fn new(topic: String) -> Self {
+        Self {
+            topic,
+            initial_pruned_entry_id: None,
+        }
+    }
 }

src/datanode/src/error.rs

@@ -301,6 +301,15 @@ pub enum Error {
         source: store_api::metadata::MetadataError,
     },
 
+    #[snafu(display("Failed to serialize WAL options for region {}", region_id))]
+    SerializeWalOptions {
+        region_id: RegionId,
+        #[snafu(source)]
+        error: serde_json::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Failed to stop region engine {}", name))]
     StopRegionEngine {
         name: String,
@@ -453,6 +462,7 @@ impl ErrorExt for Error {
 
             PayloadNotExist { .. }
             | Unexpected { .. }
+            | SerializeWalOptions { .. }
             | WatchAsyncTaskChange { .. }
             | BuildHttpClient { .. } => StatusCode::Unexpected,
 

src/datanode/src/heartbeat/handler/open_region.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use common_meta::instruction::{InstructionReply, OpenRegion, SimpleReply};
-use common_meta::wal_provider::prepare_wal_options;
+use common_meta::wal_provider::serialize_wal_options;
 use common_telemetry::info;
 use store_api::path_utils::table_dir;
 use store_api::region_request::{PathType, RegionOpenRequest};
@@ -49,7 +49,13 @@ impl InstructionHandler for OpenRegionsHandler {
                 info!(
                     "Received open region instruction, region_id: {region_id}, reason: {reason:?}"
                 );
-                prepare_wal_options(&mut region_options, region_id, &region_wal_options);
+                if let Err(err) =
+                    serialize_wal_options(&mut region_options, region_id, &region_wal_options)
+                {
+                    return Err(format!(
+                        "Failed to serialize WAL options for region {region_id}: {err:?}"
+                    ));
+                }
                 let request = RegionOpenRequest {
                     engine: region_ident.engine,
                     table_dir: table_dir(&region_storage_path, region_id.table_id()),
@@ -59,9 +65,18 @@ impl InstructionHandler for OpenRegionsHandler {
                     checkpoint: None,
                     requirements,
                 };
-                (region_id, request)
+                Ok((region_id, request))
             })
-            .collect::<Vec<_>>();
+            .collect::<Result<Vec<_>, _>>();
+        let requests = match requests {
+            Ok(requests) => requests,
+            Err(error) => {
+                return Some(InstructionReply::OpenRegions(SimpleReply {
+                    result: false,
+                    error: Some(error),
+                }));
+            }
+        };
 
         let result = ctx
             .region_server

src/datanode/src/utils.rs

@@ -22,16 +22,18 @@ use common_meta::key::topic_region::{
     TopicRegionValue,
 };
 use common_meta::kv_backend::KvBackendRef;
-use common_meta::wal_provider::{extract_topic_from_wal_options, prepare_wal_options};
+use common_meta::wal_provider::{
+    RegionWalOptions, extract_topic_from_wal_options, serialize_wal_options,
+};
 use futures::TryStreamExt;
 use snafu::ResultExt;
 use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
 use store_api::path_utils::table_dir;
 use store_api::region_request::{PathType, RegionOpenRequest, ReplayCheckpoint};
-use store_api::storage::{RegionId, RegionNumber};
+use store_api::storage::RegionId;
 use tracing::info;
 
-use crate::error::{GetMetadataSnafu, Result};
+use crate::error::{GetMetadataSnafu, Result, SerializeWalOptionsSnafu};
 
 /// The requests to open regions.
 pub struct RegionOpenRequests {
@@ -60,7 +62,7 @@ impl RegionOpenRequests {
 
 fn group_region_by_topic(
     region_id: RegionId,
-    region_options: &HashMap<RegionNumber, String>,
+    region_options: &RegionWalOptions,
     topic_regions: &mut HashMap<String, Vec<RegionId>>,
 ) {
     if let Some(topic) = extract_topic_from_wal_options(region_id, region_options) {
@@ -136,11 +138,12 @@ pub async fn build_region_open_requests(
             let region_id = RegionId::new(table_value.table_id, region_number);
             // Augments region options with wal options if a wal options is provided.
             let mut region_options = table_value.region_info.region_options.clone();
-            prepare_wal_options(
+            serialize_wal_options(
                 &mut region_options,
                 region_id,
                 &table_value.region_info.region_wal_options,
-            );
+            )
+            .context(SerializeWalOptionsSnafu { region_id })?;
             group_region_by_topic(
                 region_id,
                 &table_value.region_info.region_wal_options,
@@ -160,11 +163,12 @@ pub async fn build_region_open_requests(
             let region_id = RegionId::new(table_value.table_id, region_number);
             // Augments region options with wal options if a wal options is provided.
             let mut region_options = table_value.region_info.region_options.clone();
-            prepare_wal_options(
+            serialize_wal_options(
                 &mut region_options,
                 RegionId::new(table_value.table_id, region_number),
                 &table_value.region_info.region_wal_options,
-            );
+            )
+            .context(SerializeWalOptionsSnafu { region_id })?;
             group_region_by_topic(
                 region_id,
                 &table_value.region_info.region_wal_options,

src/meta-srv/src/error.rs

@@ -951,13 +951,6 @@ pub enum Error {
         source: common_meta::error::Error,
     },
 
-    #[snafu(display("Failed to parse wal options"))]
-    ParseWalOptions {
-        #[snafu(implicit)]
-        location: Location,
-        source: common_meta::error::Error,
-    },
-
     #[snafu(display("Failed to build kafka client."))]
     BuildKafkaClient {
         #[snafu(implicit)]
@@ -1278,8 +1271,7 @@ impl ErrorExt for Error {
             | Error::RuntimeSwitchManager { source, .. }
             | Error::KvBackend { source, .. }
             | Error::UnexpectedLogicalRouteTable { source, .. }
-            | Error::UpdateTopicNameValue { source, .. }
-            | Error::ParseWalOptions { source, .. } => source.status_code(),
+            | Error::UpdateTopicNameValue { source, .. } => source.status_code(),
             Error::ListActiveFrontends { source, .. }
             | Error::ListActiveDatanodes { source, .. }
             | Error::ListActiveFlownodes { source, .. } => source.status_code(),

src/meta-srv/src/procedure/region_migration/downgrade_leader_region.rs

@@ -379,6 +379,7 @@ mod tests {
     use common_meta::key::test_utils::new_test_table_info;
     use common_meta::peer::Peer;
     use common_meta::rpc::router::{Region, RegionRoute};
+    use common_meta::wal_provider::RegionWalOptions;
     use store_api::storage::RegionId;
     use tokio::time::Instant;
 
@@ -402,7 +403,7 @@ mod tests {
         )
     }
 
-    async fn prepare_table_metadata(ctx: &Context, wal_options: HashMap<u32, String>) {
+    async fn prepare_table_metadata(ctx: &Context, wal_options: RegionWalOptions) {
         let region_id = ctx.persistent_ctx.region_ids[0];
         let table_info = new_test_table_info(region_id.table_id());
         let region_routes = vec![RegionRoute {

src/meta-srv/src/procedure/region_migration/upgrade_candidate_region.rs

@@ -17,7 +17,6 @@ use std::collections::HashSet;
 use std::time::Duration;
 
 use api::v1::meta::MailboxMessage;
-use common_meta::ddl::utils::parse_region_wal_options;
 use common_meta::instruction::{
     Instruction, InstructionReply, UpgradeRegion, UpgradeRegionReply, UpgradeRegionsReply,
 };
@@ -97,9 +96,7 @@ impl UpgradeCandidateRegion {
                 continue;
             };
 
-            let region_wal_options =
-                parse_region_wal_options(&datanode_table_value.region_info.region_wal_options)
-                    .context(error::ParseWalOptionsSnafu)?;
+            let region_wal_options = &datanode_table_value.region_info.region_wal_options;
 
             for region_id in regions {
                 let Some(WalOptions::Kafka(kafka_wal_options)) =
@@ -359,6 +356,7 @@ mod tests {
     use common_meta::key::topic_region::{ReplayCheckpoint, TopicRegionKey, TopicRegionValue};
     use common_meta::peer::Peer;
     use common_meta::rpc::router::{Region, RegionRoute};
+    use common_meta::wal_provider::RegionWalOptions;
     use common_wal::options::KafkaWalOptions;
     use store_api::storage::RegionId;
 
@@ -382,23 +380,20 @@ mod tests {
         )
     }
 
-    fn kafka_wal_options(topic: &str) -> HashMap<u32, String> {
-        HashMap::from([(
+    fn kafka_wal_options(topic: &str) -> RegionWalOptions {
+        RegionWalOptions::from([(
             1,
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.to_string(),
-            }))
-            .unwrap(),
+            WalOptions::Kafka(KafkaWalOptions::new(topic.to_string())),
         )])
     }
 
-    async fn prepare_table_metadata(ctx: &Context, wal_options: HashMap<u32, String>) {
+    async fn prepare_table_metadata(ctx: &Context, wal_options: RegionWalOptions) {
         prepare_table_metadata_with_engine(ctx, wal_options, "engine").await;
     }
 
     async fn prepare_table_metadata_with_engine(
         ctx: &Context,
-        wal_options: HashMap<u32, String>,
+        wal_options: RegionWalOptions,
         engine: &str,
     ) {
         let region_id = ctx.persistent_ctx.region_ids[0];

src/meta-srv/src/procedure/repartition.rs

@@ -44,6 +44,7 @@ use common_meta::node_manager::NodeManagerRef;
 use common_meta::region_keeper::{MemoryRegionKeeperRef, OperatingRegionGuard};
 use common_meta::region_registry::LeaderRegionRegistryRef;
 use common_meta::rpc::router::{RegionRoute, operating_leader_region_roles};
+use common_meta::wal_provider::RegionWalOptions;
 use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
     BoxedProcedure, Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure,
@@ -53,7 +54,7 @@ use common_telemetry::{error, info, warn};
 use partition::expr::PartitionExpr;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt};
-use store_api::storage::{RegionNumber, TableId};
+use store_api::storage::TableId;
 use table::table_name::TableName;
 
 use crate::error::{self, Result};
@@ -378,7 +379,7 @@ impl Context {
         &self,
         current_table_route_value: &DeserializedValueWithBytes<TableRouteValue>,
         new_region_routes: Vec<RegionRoute>,
-        new_region_wal_options: HashMap<RegionNumber, String>,
+        new_region_wal_options: RegionWalOptions,
     ) -> Result<()> {
         let table_id = self.persistent_ctx.table_id;
         if new_region_routes.is_empty() {

src/meta-srv/src/procedure/repartition/allocate_region.rs

@@ -23,6 +23,9 @@ use common_meta::lock_key::TableLock;
 use common_meta::node_manager::NodeManagerRef;
 use common_meta::peer::PeerAllocContext;
 use common_meta::rpc::router::RegionRoute;
+use common_meta::wal_provider::{
+    RegionWalOptions, acquire_remote_wal_read_locks, refresh_initial_pruned_entry_ids,
+};
 use common_procedure::{Context as ProcedureContext, Status};
 use common_telemetry::{debug, info};
 use serde::{Deserialize, Deserializer, Serialize};
@@ -162,7 +165,7 @@ impl ExecutePlan {
             )
             .await
             .context(error::AllocateRegionRoutesSnafu { table_id })?;
-        let wal_options = ctx
+        let mut wal_options = ctx
             .wal_options_allocator
             .allocate(
                 &allocate_regions
@@ -173,6 +176,11 @@ impl ExecutePlan {
             )
             .await
             .context(error::AllocateWalOptionsSnafu { table_id })?;
+        let _remote_wal_lock_guards =
+            acquire_remote_wal_read_locks(procedure_ctx, &wal_options).await;
+        refresh_initial_pruned_entry_ids(&ctx.table_metadata_manager, &mut wal_options)
+            .await
+            .context(error::AllocateWalOptionsSnafu { table_id })?;
 
         let new_region_count = new_allocated_region_routes.len();
         let new_regions_brief: Vec<_> = new_allocated_region_routes
@@ -366,7 +374,7 @@ impl AllocateRegion {
         node_manager: &NodeManagerRef,
         raw_table_info: &TableInfo,
         region_routes: &[RegionRoute],
-        wal_options: &HashMap<RegionNumber, String>,
+        wal_options: &RegionWalOptions,
     ) -> Result<()> {
         let table_ref = TableReference::full(
             &raw_table_info.catalog_name,
@@ -403,7 +411,6 @@ impl AllocateRegion {
 
 #[cfg(test)]
 mod tests {
-    use std::collections::HashMap;
     use std::sync::Arc;
 
     use api::v1::region::region_request::Body;
@@ -500,7 +507,7 @@ mod tests {
         table_metadata_manager: TableMetadataManagerRef,
         table_id: TableId,
         concurrent_region_route: RegionRoute,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
     }
 
     #[async_trait::async_trait]

src/meta-srv/src/procedure/repartition/test_util.rs

@@ -27,6 +27,7 @@ use common_meta::peer::Peer;
 use common_meta::rpc::router::{Region, RegionRoute};
 use common_meta::sequence::SequenceBuilder;
 use common_meta::test_util::new_ddl_context_with_kv_backend;
+use common_meta::wal_provider::RegionWalOptions;
 use common_procedure::{
     Context as ProcedureContext, ContextProvider, ProcedureId, ProcedureState, Status,
 };
@@ -121,7 +122,7 @@ impl TestingEnv {
         &self,
         table_id: TableId,
         region_routes: Vec<RegionRoute>,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
     ) {
         self.table_metadata_manager
             .create_table_metadata(
@@ -137,7 +138,7 @@ impl TestingEnv {
         &self,
         table_id: TableId,
         region_routes: Vec<RegionRoute>,
-        region_wal_options: HashMap<RegionNumber, String>,
+        region_wal_options: RegionWalOptions,
     ) {
         let mut table_info = new_test_table_info_with_name(table_id, "test_table");
         table_info.meta.column_ids = vec![0, 1, 2];
@@ -163,11 +164,8 @@ pub fn range_expr(col_name: &str, start: i64, end: i64) -> PartitionExpr {
         .and(col(col_name).lt(Value::Int64(end)))
 }
 
-pub fn test_region_wal_options(region_numbers: &[RegionNumber]) -> HashMap<RegionNumber, String> {
-    let wal_options = serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-        topic: "test_topic".to_string(),
-    }))
-    .unwrap();
+pub fn test_region_wal_options(region_numbers: &[RegionNumber]) -> RegionWalOptions {
+    let wal_options = WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string()));
 
     region_numbers
         .iter()

src/meta-srv/src/procedure/repartition/utils.rs

@@ -18,6 +18,7 @@ use common_error::ext::BoxedError;
 use common_meta::key::TableMetadataManagerRef;
 use common_meta::key::datanode_table::{DatanodeTableKey, DatanodeTableValue};
 use common_meta::rpc::router::RegionRoute;
+use common_meta::wal_provider::RegionWalOptions;
 use snafu::{OptionExt, ResultExt, ensure};
 use store_api::storage::{RegionId, RegionNumber, TableId};
 
@@ -75,11 +76,11 @@ pub async fn get_datanode_table_value(
 /// - New WAL options try to overwrite existing ones for the same region
 /// - Any region in `new_region_routes` is missing a WAL option
 pub fn merge_and_validate_region_wal_options(
-    region_wal_options: &HashMap<RegionNumber, String>,
-    mut new_region_wal_options: HashMap<RegionNumber, String>,
+    region_wal_options: &RegionWalOptions,
+    mut new_region_wal_options: RegionWalOptions,
     new_region_routes: &[RegionRoute],
     table_id: TableId,
-) -> Result<HashMap<RegionNumber, String>> {
+) -> Result<RegionWalOptions> {
     // Doesn't allow overwriting existing WAL options.
     for (region_number, _) in new_region_wal_options.iter() {
         if region_wal_options.contains_key(region_number) {
@@ -196,12 +197,9 @@ mod tests {
     use crate::procedure::repartition::plan::{SourceRegionDescriptor, TargetRegionDescriptor};
     use crate::procedure::repartition::test_util::range_expr;
 
-    /// Helper function to create a Kafka WAL option string from a topic name.
-    fn kafka_wal_option(topic: &str) -> String {
-        serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-            topic: topic.to_string(),
-        }))
-        .unwrap()
+    /// Helper function to create a Kafka WAL option from a topic name.
+    fn kafka_wal_option(topic: &str) -> WalOptions {
+        WalOptions::Kafka(KafkaWalOptions::new(topic.to_string()))
     }
 
     fn new_region_route(region_id: u64, datanode_id: u64) -> RegionRoute {
@@ -260,13 +258,13 @@ mod tests {
     #[test]
     fn test_merge_and_validate_region_wal_options_success() {
         let table_id = 1;
-        let existing_wal_options: HashMap<RegionNumber, String> = vec![
+        let existing_wal_options: RegionWalOptions = vec![
             (1, kafka_wal_option("topic_1")),
             (2, kafka_wal_option("topic_2")),
         ]
         .into_iter()
         .collect();
-        let new_wal_options: HashMap<RegionNumber, String> =
+        let new_wal_options: RegionWalOptions =
             vec![(3, kafka_wal_option("topic_3"))].into_iter().collect();
         let new_region_routes = vec![
             new_region_route(1, 1),
@@ -296,14 +294,12 @@ mod tests {
     #[test]
     fn test_merge_and_validate_region_wal_options_new_overrides_existing() {
         let table_id = 1;
-        let existing_wal_options: HashMap<RegionNumber, String> =
-            vec![(1, kafka_wal_option("topic_1_old"))]
-                .into_iter()
-                .collect();
-        let new_wal_options: HashMap<RegionNumber, String> =
-            vec![(1, kafka_wal_option("topic_1_new"))]
-                .into_iter()
-                .collect();
+        let existing_wal_options: RegionWalOptions = vec![(1, kafka_wal_option("topic_1_old"))]
+            .into_iter()
+            .collect();
+        let new_wal_options: RegionWalOptions = vec![(1, kafka_wal_option("topic_1_new"))]
+            .into_iter()
+            .collect();
         let new_region_routes = vec![new_region_route(1, 1)];
         merge_and_validate_region_wal_options(
             &existing_wal_options,
@@ -317,7 +313,7 @@ mod tests {
     #[test]
     fn test_merge_and_validate_region_wal_options_filters_removed_regions() {
         let table_id = 1;
-        let existing_wal_options: HashMap<RegionNumber, String> = vec![
+        let existing_wal_options: RegionWalOptions = vec![
             (1, kafka_wal_option("topic_1")),
             (2, kafka_wal_option("topic_2")),
             (3, kafka_wal_option("topic_3")),
@@ -345,7 +341,7 @@ mod tests {
     #[test]
     fn test_merge_and_validate_region_wal_options_missing_option() {
         let table_id = 1;
-        let existing_wal_options: HashMap<RegionNumber, String> =
+        let existing_wal_options: RegionWalOptions =
             vec![(1, kafka_wal_option("topic_1"))].into_iter().collect();
         let new_wal_options = HashMap::new();
         // Region 2 is in routes but has no WAL option

src/meta-srv/src/procedure/test_util.rs

@@ -31,6 +31,7 @@ use common_meta::region_registry::{
 };
 use common_meta::rpc::router::{Region, RegionRoute};
 use common_meta::sequence::Sequence;
+use common_meta::wal_provider::RegionWalOptions;
 use common_time::util::current_time_millis;
 use common_wal::options::{KafkaWalOptions, WalOptions};
 use store_api::logstore::EntryId;
@@ -317,11 +318,8 @@ pub async fn new_wal_prune_metadata(
                 ..Default::default()
             })
             .collect::<Vec<_>>();
-        let wal_options = WalOptions::Kafka(KafkaWalOptions {
-            topic: topic.clone(),
-        });
-        let wal_options = serde_json::to_string(&wal_options).unwrap();
-        let region_wal_options: HashMap<u32, String> = (0..n_region)
+        let wal_options = WalOptions::Kafka(KafkaWalOptions::new(topic.clone()));
+        let region_wal_options: RegionWalOptions = (0..n_region)
             .map(|region_number| (region_number, wal_options.clone()))
             .collect();
 

src/meta-srv/src/procedure/wal_prune.rs

@@ -172,7 +172,12 @@ impl Procedure for WalPruneProcedure {
         false
     }
 
-    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
+    async fn execute(&mut self, ctx: &ProcedureContext) -> ProcedureResult<Status> {
+        let _guard = ctx
+            .provider
+            .acquire_lock(&(RemoteWalLock::Write(self.data.topic.clone()).into()))
+            .await;
+
         self.on_prune().await.map_err(|e| {
             if e.is_retryable() {
                 ProcedureError::retry_later(e)

src/metric-engine/src/engine.rs

@@ -805,9 +805,7 @@ mod test {
             let physical_region_id = RegionId::new(1, i);
             physical_region_ids.push(physical_region_id);
 
-            let wal_options = WalOptions::Kafka(KafkaWalOptions {
-                topic: topics[topic_idx(i)].clone(),
-            });
+            let wal_options = WalOptions::Kafka(KafkaWalOptions::new(topics[topic_idx(i)].clone()));
             env.create_physical_region(
                 physical_region_id,
                 &table_dir(physical_region_id),
@@ -841,9 +839,8 @@ mod test {
             .enumerate()
             .map(|(idx, region_id)| {
                 let mut options = HashMap::new();
-                let wal_options = WalOptions::Kafka(KafkaWalOptions {
-                    topic: topics[topic_idx(idx as u32)].clone(),
-                });
+                let wal_options =
+                    WalOptions::Kafka(KafkaWalOptions::new(topics[topic_idx(idx as u32)].clone()));
                 options.insert(PHYSICAL_TABLE_METADATA_KEY.to_string(), String::new());
                 options.insert(
                     WAL_OPTIONS_KEY.to_string(),

src/mito2/src/engine/basic_test.rs

@@ -197,10 +197,7 @@ async fn test_region_replay_with_format(factory: Option<LogStoreFactory>, flat_f
     if let Some(topic) = &topic {
         options.insert(
             WAL_OPTIONS_KEY.to_string(),
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            }))
-            .unwrap(),
+            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone()))).unwrap(),
         );
     };
 

src/mito2/src/engine/batch_catchup_test.rs

@@ -149,7 +149,7 @@ async fn test_batch_catchup_with_format(factory: Option<LogStoreFactory>, flat_f
             let mut options = HashMap::new();
             options.insert(
                 WAL_OPTIONS_KEY.to_string(),
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions { topic })).unwrap(),
+                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic))).unwrap(),
             );
             (
                 region_id,

src/mito2/src/engine/batch_open_test.rs

@@ -118,10 +118,7 @@ async fn test_batch_open_with_format(factory: Option<LogStoreFactory>, flat_form
     if let Some(topic) = &topic {
         options.insert(
             WAL_OPTIONS_KEY.to_string(),
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            }))
-            .unwrap(),
+            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone()))).unwrap(),
         );
     };
     let mut requests = (1..=num_regions)
@@ -204,10 +201,7 @@ async fn test_batch_open_err_with_format(factory: Option<LogStoreFactory>, flat_
     if let Some(topic) = &topic {
         options.insert(
             WAL_OPTIONS_KEY.to_string(),
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            }))
-            .unwrap(),
+            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone()))).unwrap(),
         );
     };
     let num_regions = 3u32;

src/mito2/src/engine/catchup_test.rs

@@ -81,10 +81,7 @@ async fn test_catchup_with_last_entry_id(factory: Option<LogStoreFactory>) {
     if let Some(topic) = &topic {
         options.insert(
             WAL_OPTIONS_KEY.to_string(),
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            }))
-            .unwrap(),
+            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone()))).unwrap(),
         );
     };
     follower_engine
@@ -203,10 +200,7 @@ async fn test_catchup_with_incorrect_last_entry_id(factory: Option<LogStoreFacto
     if let Some(topic) = &topic {
         options.insert(
             WAL_OPTIONS_KEY.to_string(),
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            }))
-            .unwrap(),
+            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone()))).unwrap(),
         );
     };
     follower_engine
@@ -307,10 +301,7 @@ async fn test_catchup_without_last_entry_id(factory: Option<LogStoreFactory>) {
     if let Some(topic) = &topic {
         options.insert(
             WAL_OPTIONS_KEY.to_string(),
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            }))
-            .unwrap(),
+            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone()))).unwrap(),
         );
     };
     follower_engine
@@ -410,10 +401,7 @@ async fn test_catchup_with_manifest_update(factory: Option<LogStoreFactory>) {
     if let Some(topic) = &topic {
         options.insert(
             WAL_OPTIONS_KEY.to_string(),
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            }))
-            .unwrap(),
+            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone()))).unwrap(),
         );
     };
     follower_engine

src/mito2/src/engine/flush_test.rs

@@ -382,10 +382,7 @@ async fn test_flush_reopen_region(factory: Option<LogStoreFactory>) {
     if let Some(topic) = &topic {
         options.insert(
             WAL_OPTIONS_KEY.to_string(),
-            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            }))
-            .unwrap(),
+            serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone()))).unwrap(),
         );
     };
     reopen_region(&engine, region_id, table_dir, true, options).await;
@@ -623,10 +620,8 @@ fn kafka_wal_options(topic: &Option<String>) -> HashMap<String, String> {
         .map(|topic| {
             HashMap::from([(
                 WAL_OPTIONS_KEY.to_string(),
-                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions {
-                    topic: topic.clone(),
-                }))
-                .unwrap(),
+                serde_json::to_string(&WalOptions::Kafka(KafkaWalOptions::new(topic.clone())))
+                    .unwrap(),
             )])
         })
         .unwrap_or_default()

src/mito2/src/region/opener.rs

@@ -84,6 +84,13 @@ const PARQUET_META_PRELOAD_CONCURRENCY: usize = 8;
 static PARQUET_META_PRELOAD_SEMAPHORE: LazyLock<Semaphore> =
     LazyLock::new(|| Semaphore::new(PARQUET_META_PRELOAD_CONCURRENCY));
 
+fn initial_pruned_entry_id(wal_options: &WalOptions) -> EntryId {
+    match wal_options {
+        WalOptions::Kafka(options) => options.initial_pruned_entry_id.unwrap_or(0),
+        WalOptions::RaftEngine | WalOptions::Noop => 0,
+    }
+}
+
 /// A fetcher to retrieve partition expr for a region.
 ///
 /// Compatibility: older regions didn't persist `partition_expr` in engine metadata,
@@ -325,7 +332,10 @@ impl RegionOpener {
             .and_then(|cm| cm.write_cache())
             .and_then(|wc| wc.manifest_cache());
         // For remote WAL, we need to set flushed_entry_id to current topic's latest entry id.
-        let flushed_entry_id = provider.initial_flushed_entry_id::<S>(wal.store());
+        // Kafka WAL allocation also carries the topic's pruned entry id as a create-time hint.
+        let flushed_entry_id = provider
+            .initial_flushed_entry_id::<S>(wal.store())
+            .max(initial_pruned_entry_id(&options.wal_options));
         let manifest_manager = RegionManifestManager::new(
             metadata.clone(),
             flushed_entry_id,
@@ -1233,6 +1243,7 @@ mod tests {
     use common_base::readable_size::ReadableSize;
     use common_test_util::temp_dir::create_temp_dir;
     use common_time::Timestamp;
+    use common_wal::options::{KafkaWalOptions, WalOptions};
     use datatypes::arrow::array::{ArrayRef, BinaryArray, Int64Array};
     use datatypes::arrow::record_batch::RecordBatch;
     use object_store::ObjectStore;
@@ -1244,7 +1255,7 @@ mod tests {
     use store_api::storage::{FileId, RegionId};
 
     use super::{
-        preload_parquet_meta_cache_for_files, sanitize_region_options,
+        initial_pruned_entry_id, preload_parquet_meta_cache_for_files, sanitize_region_options,
         supports_open_region_object_storage_requirement,
     };
     use crate::cache::CacheManager;
@@ -1280,6 +1291,25 @@ mod tests {
             .finish()
     }
 
+    #[test]
+    fn test_initial_pruned_entry_id() {
+        assert_eq!(0, initial_pruned_entry_id(&WalOptions::RaftEngine));
+        assert_eq!(0, initial_pruned_entry_id(&WalOptions::Noop));
+        assert_eq!(
+            0,
+            initial_pruned_entry_id(&WalOptions::Kafka(KafkaWalOptions::new(
+                "test_topic".to_string()
+            )))
+        );
+        assert_eq!(
+            42,
+            initial_pruned_entry_id(&WalOptions::Kafka(KafkaWalOptions {
+                topic: "test_topic".to_string(),
+                initial_pruned_entry_id: Some(42),
+            }))
+        );
+    }
+
     #[test]
     #[cfg(not(feature = "test-shared-fs-region-migration"))]
     fn test_open_requirement_rejects_fs_object_store() {

src/mito2/src/region/options.rs

@@ -631,9 +631,7 @@ mod tests {
     fn test_with_any_wal_options() {
         let all_wal_options = [
             WalOptions::RaftEngine,
-            WalOptions::Kafka(KafkaWalOptions {
-                topic: "test_topic".to_string(),
-            }),
+            WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string())),
         ];
         all_wal_options.iter().all(test_with_wal_options);
     }
@@ -804,9 +802,7 @@ mod tests {
 
     #[test]
     fn test_with_all() {
-        let wal_options = WalOptions::Kafka(KafkaWalOptions {
-            topic: "test_topic".to_string(),
-        });
+        let wal_options = WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string()));
         let map = make_map(&[
             ("ttl", "7d"),
             ("compaction.twcs.trigger_file_num", "8"),
@@ -877,9 +873,7 @@ mod tests {
             compaction_override: false,
             storage: Some("S3".to_string()),
             append_mode: false,
-            wal_options: WalOptions::Kafka(KafkaWalOptions {
-                topic: "test_topic".to_string(),
-            }),
+            wal_options: WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string())),
             index_options: IndexOptions {
                 inverted_index: InvertedIndexOptions {
                     ignore_column_ids: vec![1, 2, 3],
@@ -935,9 +929,7 @@ mod tests {
             compaction_override: false,
             storage: Some("S3".to_string()),
             append_mode: false,
-            wal_options: WalOptions::Kafka(KafkaWalOptions {
-                topic: "test_topic".to_string(),
-            }),
+            wal_options: WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string())),
             index_options: IndexOptions {
                 inverted_index: InvertedIndexOptions {
                     ignore_column_ids: vec![],

src/mito2/src/test_util.rs

@@ -891,9 +891,7 @@ impl CreateRequestBuilder {
         });
         let mut options = self.options.clone();
         if let Some(topic) = &self.kafka_topic {
-            let wal_options = WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            });
+            let wal_options = WalOptions::Kafka(KafkaWalOptions::new(topic.clone()));
             options.insert(
                 WAL_OPTIONS_KEY.to_string(),
                 serde_json::to_string(&wal_options).unwrap(),
@@ -958,9 +956,7 @@ impl CreateRequestBuilder {
         });
         let mut options = self.options.clone();
         if let Some(topic) = &self.kafka_topic {
-            let wal_options = WalOptions::Kafka(KafkaWalOptions {
-                topic: topic.clone(),
-            });
+            let wal_options = WalOptions::Kafka(KafkaWalOptions::new(topic.clone()));
             options.insert(
                 WAL_OPTIONS_KEY.to_string(),
                 serde_json::to_string(&wal_options).unwrap(),

src/operator/src/tests/partition_manager.rs

@@ -22,6 +22,7 @@ use common_meta::key::table_route::TableRouteValue;
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::peer::Peer;
 use common_meta::rpc::router::{Region, RegionRoute};
+use common_meta::wal_provider::RegionWalOptions;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{ColumnSchema, SchemaBuilder};
 use moka::future::CacheBuilder;
@@ -69,7 +70,7 @@ pub fn new_test_table_info(
         .unwrap()
 }
 
-fn new_test_region_wal_options(regions: Vec<RegionNumber>) -> HashMap<RegionNumber, String> {
+fn new_test_region_wal_options(regions: Vec<RegionNumber>) -> RegionWalOptions {
     // TODO(niebayes): construct region wal options for test.
     let _ = regions;
     HashMap::default()