refactor: restructure leveled map data architecture (#18619)

- Move LeveledMapData to separate module with impl files
- Split ApplierData impl blocks into dedicated files
- Change immutable_levels.indexes() return type to Vec
- Consolidate mutex access patterns inside LeveledMapDataInner
- Remove redundant impl_leveled_map_data.rs module

This restructuring improves code organization by separating
concerns and makes the codebase more maintainable with
clearer module boundaries and consistent access patterns.

Author: drmingdrmer

src/meta/raft-store/src/applier/applier_data.rs

@@ -0,0 +1,33 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use map_api::mvcc;
+use seq_marked::SeqMarked;
+use state_machine_api::ExpireKey;
+
+use crate::applier::applier_data::ApplierData;
+use crate::leveled_store::types::Key;
+use crate::leveled_store::types::Namespace;
+use crate::leveled_store::types::Value;
+
+#[async_trait::async_trait]
+impl mvcc::ScopedView<ExpireKey, String> for ApplierData {
+    fn set(&mut self, key: ExpireKey, value: Option<String>) -> SeqMarked<()> {
+        self.view.set(
+            Namespace::Expire,
+            Key::Expire(key),
+            value.map(Value::Expire),
+        )
+    }
+}

src/meta/raft-store/src/applier/applier_data/impl_expire_scoped_view.rs

@@ -0,0 +1,61 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::io;
+use std::ops::RangeBounds;
+
+use futures_util::StreamExt;
+use futures_util::TryStreamExt;
+use map_api::mvcc;
+use map_api::mvcc::ViewReadonly;
+use map_api::IOResultStream;
+use seq_marked::InternalSeq;
+use seq_marked::SeqMarked;
+use state_machine_api::ExpireKey;
+
+use crate::applier::applier_data::ApplierData;
+use crate::leveled_store::types::Key;
+use crate::leveled_store::types::Namespace;
+
+#[async_trait::async_trait]
+impl mvcc::ScopedViewReadonly<ExpireKey, String> for ApplierData {
+    fn base_seq(&self) -> InternalSeq {
+        self.view.base_seq()
+    }
+
+    async fn get(&self, key: ExpireKey) -> Result<SeqMarked<String>, io::Error> {
+        let got = self.view.get(Namespace::Expire, Key::Expire(key)).await?;
+        Ok(got.map(|x| x.into_expire()))
+    }
+
+    async fn range<R>(
+        &self,
+        range: R,
+    ) -> Result<IOResultStream<(ExpireKey, SeqMarked<String>)>, io::Error>
+    where
+        R: RangeBounds<ExpireKey> + Send + Sync + Clone + 'static,
+    {
+        let start = range.start_bound().cloned();
+        let end = range.end_bound().cloned();
+
+        let start = start.map(Key::Expire);
+        let end = end.map(Key::Expire);
+
+        let strm = self.view.range(Namespace::Expire, (start, end)).await?;
+
+        Ok(strm
+            .map_ok(|(k, v)| (k.into_expire(), v.map(|x| x.into_expire())))
+            .boxed())
+    }
+}

src/meta/raft-store/src/applier/applier_data/impl_expire_scoped_view_readonly.rs

@@ -0,0 +1,31 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use map_api::mvcc;
+use seq_marked::SeqMarked;
+use state_machine_api::MetaValue;
+use state_machine_api::UserKey;
+
+use crate::applier::applier_data::ApplierData;
+use crate::leveled_store::types::Key;
+use crate::leveled_store::types::Namespace;
+use crate::leveled_store::types::Value;
+
+#[async_trait::async_trait]
+impl mvcc::ScopedView<UserKey, MetaValue> for ApplierData {
+    fn set(&mut self, key: UserKey, value: Option<MetaValue>) -> SeqMarked<()> {
+        self.view
+            .set(Namespace::User, Key::User(key), value.map(Value::User))
+    }
+}

src/meta/raft-store/src/applier/applier_data/impl_user_scoped_view.rs

@@ -0,0 +1,62 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::io;
+use std::ops::RangeBounds;
+
+use futures_util::StreamExt;
+use futures_util::TryStreamExt;
+use map_api::mvcc;
+use map_api::mvcc::ViewReadonly;
+use map_api::IOResultStream;
+use seq_marked::InternalSeq;
+use seq_marked::SeqMarked;
+use state_machine_api::MetaValue;
+use state_machine_api::UserKey;
+
+use crate::applier::applier_data::ApplierData;
+use crate::leveled_store::types::Key;
+use crate::leveled_store::types::Namespace;
+
+#[async_trait::async_trait]
+impl mvcc::ScopedViewReadonly<UserKey, MetaValue> for ApplierData {
+    fn base_seq(&self) -> InternalSeq {
+        self.view.base_seq()
+    }
+
+    async fn get(&self, key: UserKey) -> Result<SeqMarked<MetaValue>, io::Error> {
+        let got = self.view.get(Namespace::User, Key::User(key)).await?;
+        Ok(got.map(|x| x.into_user()))
+    }
+
+    async fn range<R>(
+        &self,
+        range: R,
+    ) -> Result<IOResultStream<(UserKey, SeqMarked<MetaValue>)>, io::Error>
+    where
+        R: RangeBounds<UserKey> + Send + Sync + Clone + 'static,
+    {
+        let start = range.start_bound().cloned();
+        let end = range.end_bound().cloned();
+
+        let start = start.map(Key::User);
+        let end = end.map(Key::User);
+
+        let strm = self.view.range(Namespace::User, (start, end)).await?;
+
+        Ok(strm
+            .map_ok(|(k, v)| (k.into_user(), v.map(|x| x.into_user())))
+            .boxed())
+    }
+}

src/meta/raft-store/src/applier/applier_data/impl_user_scoped_view_readonly.rs

@@ -0,0 +1,45 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use map_api::mvcc;
+
+use crate::leveled_store::leveled_map::applier_acquirer::WriterPermit;
+use crate::leveled_store::leveled_map::leveled_map_data::LeveledMapData;
+use crate::leveled_store::types::Key;
+use crate::leveled_store::types::Namespace;
+use crate::leveled_store::types::Value;
+use crate::sm_v003::OnChange;
+
+mod impl_expire_scoped_view;
+mod impl_expire_scoped_view_readonly;
+mod impl_user_scoped_view;
+mod impl_user_scoped_view_readonly;
+
+pub(crate) type StateMachineView = mvcc::View<Namespace, Key, Value, Arc<LeveledMapData>>;
+
+pub(crate) struct ApplierData {
+    /// Hold a unique permit to serialize all apply operations to the state machine.
+    pub(crate) _permit: WriterPermit,
+
+    pub(crate) view: StateMachineView,
+
+    /// Since when to start cleaning expired keys.
+    pub(crate) cleanup_start_time: Arc<Mutex<Duration>>,
+
+    pub(crate) on_change_applied: Arc<Option<OnChange>>,
+}

src/meta/raft-store/src/applier/applier_data/mod.rs

@@ -65,7 +65,6 @@ use state_machine_api::StateMachineApi;
 use crate::state_machine_api_ext::StateMachineApiExt;
 
 pub(crate) mod applier_data;
-mod impl_leveled_map_data;
 
 /// A helper that applies raft log `Entry` to the state machine.
 pub struct Applier<SM>