feat: enable atomic write for file object storage (#643)

* fix: remove opendal from catalog dependencies

* feat: enable atomic writing for fs service

Cargo.lock

@@ -845,7 +845,6 @@ dependencies = [
  "meta-client",
  "mito",
  "object-store",
- "opendal",
  "regex",
  "serde",
  "serde_json",

src/catalog/Cargo.toml

@@ -27,7 +27,6 @@ futures = "0.3"
 futures-util = "0.3"
 lazy_static = "1.4"
 meta-client = { path = "../meta-client" }
-opendal = "0.21"
 regex = "1.6"
 serde = "1.0"
 serde_json = "1.0"
@@ -40,7 +39,6 @@ tokio = { version = "1.18", features = ["full"] }
 chrono = "0.4"
 log-store = { path = "../log-store" }
 object-store = { path = "../object-store" }
-opendal = "0.21"
 storage = { path = "../storage" }
 mito = { path = "../mito", features = ["test"] }
 tempdir = "0.3"

src/datanode/src/instance.rs

@@ -197,8 +197,11 @@ pub(crate) async fn new_object_store(store_config: &ObjectStoreConfig) -> Result
 
     info!("The storage directory is: {}", &data_dir);
 
+    let atomic_write_dir = format!("{}/.tmp/", data_dir);
+
     let accessor = Builder::default()
         .root(&data_dir)
+        .atomic_write_dir(&atomic_write_dir)
         .build()
         .context(error::InitBackendSnafu { dir: &data_dir })?;
 

src/object-store/tests/object_store_test.rs

@@ -88,10 +88,12 @@ async fn test_object_list(store: &ObjectStore) -> Result<()> {
 
 #[tokio::test]
 async fn test_fs_backend() -> Result<()> {
+    let data_dir = TempDir::new("test_fs_backend")?;
     let tmp_dir = TempDir::new("test_fs_backend")?;
     let store = ObjectStore::new(
         fs::Builder::default()
-            .root(&tmp_dir.path().to_string_lossy())
+            .root(&data_dir.path().to_string_lossy())
+            .atomic_write_dir(&tmp_dir.path().to_string_lossy())
             .build()?,
     );