feat(mito): Init the write cache in datanode

src/datanode/src/datanode.rs

@@ -42,7 +42,7 @@ use metric_engine::engine::MetricEngine;
 use mito2::config::MitoConfig;
 use mito2::engine::MitoEngine;
 use object_store::manager::{ObjectStoreManager, ObjectStoreManagerRef};
-use object_store::util::normalize_dir;
+use object_store::util::{join_dir, normalize_dir};
 use query::QueryEngineFactory;
 use servers::export_metrics::ExportMetricsTask;
 use servers::grpc::{GrpcServer, GrpcServerConfig};
@@ -60,9 +60,9 @@ use tokio::sync::Notify;
 
 use crate::config::{DatanodeOptions, RegionEngineConfig};
 use crate::error::{
-    CreateDirSnafu, GetMetadataSnafu, MissingKvBackendSnafu, MissingNodeIdSnafu, OpenLogStoreSnafu,
-    ParseAddrSnafu, Result, RuntimeResourceSnafu, ShutdownInstanceSnafu, ShutdownServerSnafu,
-    StartServerSnafu,
+    BuildMitoEngineSnafu, CreateDirSnafu, GetMetadataSnafu, MissingKvBackendSnafu,
+    MissingNodeIdSnafu, OpenLogStoreSnafu, ParseAddrSnafu, Result, RuntimeResourceSnafu,
+    ShutdownInstanceSnafu, ShutdownServerSnafu, StartServerSnafu,
 };
 use crate::event_listener::{
     new_region_server_event_channel, NoopRegionServerEventListener, RegionServerEventListenerRef,
@@ -458,20 +458,33 @@ impl DatanodeBuilder {
     async fn build_mito_engine(
         opts: &DatanodeOptions,
         object_store_manager: ObjectStoreManagerRef,
-        config: MitoConfig,
+        mut config: MitoConfig,
     ) -> Result<MitoEngine> {
+        // Sets write cache path if it is empty.
+        if config.experimental_write_cache_path.is_empty() {
+            config.experimental_write_cache_path = join_dir(&opts.storage.data_home, "write_cache");
+            info!(
+                "Sets write cache path to {}",
+                config.experimental_write_cache_path
+            );
+        }
+
         let mito_engine = match &opts.wal {
             WalConfig::RaftEngine(raft_engine_config) => MitoEngine::new(
                 config,
                 Self::build_raft_engine_log_store(&opts.storage.data_home, raft_engine_config)
                     .await?,
                 object_store_manager,
-            ),
+            )
+            .await
+            .context(BuildMitoEngineSnafu)?,
             WalConfig::Kafka(kafka_config) => MitoEngine::new(
                 config,
                 Self::build_kafka_log_store(kafka_config).await?,
                 object_store_manager,
-            ),
+            )
+            .await
+            .context(BuildMitoEngineSnafu)?,
         };
         Ok(mito_engine)
     }

src/datanode/src/error.rs

@@ -282,6 +282,12 @@ pub enum Error {
         source: metric_engine::error::Error,
         location: Location,
     },
+
+    #[snafu(display("Failed to build mito engine"))]
+    BuildMitoEngine {
+        source: mito2::error::Error,
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -352,6 +358,7 @@ impl ErrorExt for Error {
             StopRegionEngine { source, .. } => source.status_code(),
 
             FindLogicalRegions { source, .. } => source.status_code(),
+            BuildMitoEngine { source, .. } => source.status_code(),
         }
     }
 

src/datanode/src/store.rs

@@ -26,10 +26,10 @@ use std::{env, path};
 
 use common_base::readable_size::ReadableSize;
 use common_telemetry::logging::info;
-use object_store::layers::{LoggingLayer, LruCacheLayer, RetryLayer, TracingLayer};
-use object_store::services::Fs as FsBuilder;
-use object_store::util::normalize_dir;
-use object_store::{util, HttpClient, ObjectStore, ObjectStoreBuilder};
+use object_store::layers::{LruCacheLayer, RetryLayer};
+use object_store::services::Fs;
+use object_store::util::{join_dir, normalize_dir, with_instrument_layers};
+use object_store::{HttpClient, ObjectStore, ObjectStoreBuilder};
 use snafu::prelude::*;
 
 use crate::config::{ObjectStoreConfig, DEFAULT_OBJECT_STORE_CACHE_SIZE};
@@ -60,16 +60,7 @@ pub(crate) async fn new_object_store(
         object_store
     };
 
-    let store = object_store
-        .layer(
-            LoggingLayer::default()
-                // Print the expected error only in DEBUG level.
-                // See https://docs.rs/opendal/latest/opendal/layers/struct.LoggingLayer.html#method.with_error_level
-                .with_error_level(Some("debug"))
-                .expect("input error level must be valid"),
-        )
-        .layer(TracingLayer)
-        .layer(object_store::layers::PrometheusMetricsLayer);
+    let store = with_instrument_layers(object_store);
     Ok(store)
 }
 
@@ -114,11 +105,10 @@ async fn create_object_store_with_cache(
     };
 
     if let Some(path) = cache_path {
-        let path = util::normalize_dir(path);
-        let atomic_temp_dir = format!("{path}.tmp/");
+        let atomic_temp_dir = join_dir(path, ".tmp/");
         clean_temp_dir(&atomic_temp_dir)?;
-        let cache_store = FsBuilder::default()
-            .root(&path)
+        let cache_store = Fs::default()
+            .root(path)
             .atomic_write_dir(&atomic_temp_dir)
             .build()
             .context(error::InitBackendSnafu)?;

src/datanode/src/store/azblob.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use common_telemetry::logging::info;
-use object_store::services::Azblob as AzureBuilder;
+use object_store::services::Azblob;
 use object_store::{util, ObjectStore};
 use secrecy::ExposeSecret;
 use snafu::prelude::*;
@@ -30,7 +30,7 @@ pub(crate) async fn new_azblob_object_store(azblob_config: &AzblobConfig) -> Res
         azblob_config.container, &root
     );
 
-    let mut builder = AzureBuilder::default();
+    let mut builder = Azblob::default();
     let _ = builder
         .root(&root)
         .container(&azblob_config.container)

src/datanode/src/store/fs.rs

@@ -15,7 +15,8 @@
 use std::{fs, path};
 
 use common_telemetry::logging::info;
-use object_store::services::Fs as FsBuilder;
+use object_store::services::Fs;
+use object_store::util::join_dir;
 use object_store::ObjectStore;
 use snafu::prelude::*;
 
@@ -31,10 +32,10 @@ pub(crate) async fn new_fs_object_store(
         .context(error::CreateDirSnafu { dir: data_home })?;
     info!("The file storage home is: {}", data_home);
 
-    let atomic_write_dir = format!("{data_home}.tmp/");
+    let atomic_write_dir = join_dir(data_home, ".tmp/");
     store::clean_temp_dir(&atomic_write_dir)?;
 
-    let mut builder = FsBuilder::default();
+    let mut builder = Fs::default();
     let _ = builder.root(data_home).atomic_write_dir(&atomic_write_dir);
 
     let object_store = ObjectStore::new(builder)

src/datanode/src/store/gcs.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use common_telemetry::logging::info;
-use object_store::services::Gcs as GCSBuilder;
+use object_store::services::Gcs;
 use object_store::{util, ObjectStore};
 use secrecy::ExposeSecret;
 use snafu::prelude::*;
@@ -29,7 +29,7 @@ pub(crate) async fn new_gcs_object_store(gcs_config: &GcsConfig) -> Result<Objec
         gcs_config.bucket, &root
     );
 
-    let mut builder = GCSBuilder::default();
+    let mut builder = Gcs::default();
     builder
         .root(&root)
         .bucket(&gcs_config.bucket)

src/datanode/src/store/oss.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use common_telemetry::logging::info;
-use object_store::services::Oss as OSSBuilder;
+use object_store::services::Oss;
 use object_store::{util, ObjectStore};
 use secrecy::ExposeSecret;
 use snafu::prelude::*;
@@ -29,7 +29,7 @@ pub(crate) async fn new_oss_object_store(oss_config: &OssConfig) -> Result<Objec
         oss_config.bucket, &root
     );
 
-    let mut builder = OSSBuilder::default();
+    let mut builder = Oss::default();
     let _ = builder
         .root(&root)
         .bucket(&oss_config.bucket)

src/datanode/src/store/s3.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use common_telemetry::logging::info;
-use object_store::services::S3 as S3Builder;
+use object_store::services::S3;
 use object_store::{util, ObjectStore};
 use secrecy::ExposeSecret;
 use snafu::prelude::*;
@@ -30,7 +30,7 @@ pub(crate) async fn new_s3_object_store(s3_config: &S3Config) -> Result<ObjectSt
         s3_config.bucket, &root
     );
 
-    let mut builder = S3Builder::default();
+    let mut builder = S3::default();
     let _ = builder
         .root(&root)
         .bucket(&s3_config.bucket)

src/mito2/src/access_layer.rs

@@ -14,13 +14,15 @@
 
 use std::sync::Arc;
 
+use object_store::services::Fs;
+use object_store::util::{join_dir, with_instrument_layers};
 use object_store::ObjectStore;
 use snafu::ResultExt;
 use store_api::metadata::RegionMetadataRef;
 
 use crate::cache::write_cache::SstUploadRequest;
 use crate::cache::CacheManagerRef;
-use crate::error::{DeleteSstSnafu, Result};
+use crate::error::{CleanDirSnafu, DeleteSstSnafu, OpenDalSnafu, Result};
 use crate::read::Source;
 use crate::sst::file::{FileHandle, FileId};
 use crate::sst::location;
@@ -119,3 +121,31 @@ pub(crate) struct SstWriteRequest {
     pub(crate) cache_manager: CacheManagerRef,
     pub(crate) storage: Option<String>,
 }
+
+/// Creates a fs object store with atomic write dir.
+pub(crate) async fn new_fs_object_store(root: &str) -> Result<ObjectStore> {
+    let atomic_write_dir = join_dir(root, ".tmp/");
+    clean_dir(&atomic_write_dir).await?;
+
+    let mut builder = Fs::default();
+    builder.root(root).atomic_write_dir(&atomic_write_dir);
+    let object_store = ObjectStore::new(builder).context(OpenDalSnafu)?.finish();
+
+    // Add layers.
+    let object_store = with_instrument_layers(object_store);
+    Ok(object_store)
+}
+
+/// Clean the directory.
+async fn clean_dir(dir: &str) -> Result<()> {
+    if tokio::fs::try_exists(dir)
+        .await
+        .context(CleanDirSnafu { dir })?
+    {
+        tokio::fs::remove_dir_all(dir)
+            .await
+            .context(CleanDirSnafu { dir })?;
+    }
+
+    Ok(())
+}