fix(object_store): fs and hdfs object store should set atomic_write_d…

…ir (#15155)

risedev.yml

@@ -164,6 +164,17 @@ profile:
       - use: compactor
       # - use: prometheus
       # - use: grafana
+  fs:
+    steps:
+      # - use: etcd
+      - use: meta-node
+      - use: compute-node
+      - use: frontend
+      - use: opendal
+        engine: fs
+      - use: compactor
+      # - use: prometheus
+      # - use: grafana
   webhdfs:
     steps:
       # - use: etcd

src/object_store/src/object/opendal_engine/fs.rs

@@ -17,15 +17,17 @@ use opendal::services::Fs;
 use opendal::Operator;
 
 use super::{EngineType, OpendalObjectStore};
+use crate::object::opendal_engine::ATOMIC_WRITE_DIR;
 use crate::object::ObjectResult;
+
 impl OpendalObjectStore {
     /// create opendal fs engine.
     pub fn new_fs_engine(root: String) -> ObjectResult<Self> {
         // Create fs backend builder.
         let mut builder = Fs::default();
-
         builder.root(&root);
-
+        let atomic_write_dir = format!("{}/{}", root, ATOMIC_WRITE_DIR);
+        builder.atomic_write_dir(&atomic_write_dir);
         let op: Operator = Operator::new(builder)?
             .layer(RetryLayer::default())
             .finish();

src/object_store/src/object/opendal_engine/hdfs.rs

@@ -17,7 +17,9 @@ use opendal::services::Hdfs;
 use opendal::Operator;
 
 use super::{EngineType, OpendalObjectStore};
+use crate::object::opendal_engine::ATOMIC_WRITE_DIR;
 use crate::object::ObjectResult;
+
 impl OpendalObjectStore {
     /// create opendal hdfs engine.
     pub fn new_hdfs_engine(namenode: String, root: String) -> ObjectResult<Self> {
@@ -26,7 +28,8 @@ impl OpendalObjectStore {
         // Set the name node for hdfs.
         builder.name_node(&namenode);
         builder.root(&root);
-
+        let atomic_write_dir = format!("{}/{}", root, ATOMIC_WRITE_DIR);
+        builder.atomic_write_dir(&atomic_write_dir);
         let op: Operator = Operator::new(builder)?
             .layer(LoggingLayer::default())
             .layer(RetryLayer::default())

src/object_store/src/object/opendal_engine/mod.rs

@@ -31,3 +31,6 @@ pub mod opendal_s3;
 pub mod oss;
 
 pub mod fs;
+
+// To make sure the the operation is consistent, we should specially set `atomic_write_dir` for fs, hdfs and webhdfs services.
+const ATOMIC_WRITE_DIR: &str = "atomic_write_dir/";

src/object_store/src/object/opendal_engine/webhdfs.rs

@@ -17,6 +17,7 @@ use opendal::services::Webhdfs;
 use opendal::Operator;
 
 use super::{EngineType, OpendalObjectStore};
+use crate::object::opendal_engine::ATOMIC_WRITE_DIR;
 use crate::object::ObjectResult;
 
 impl OpendalObjectStore {
@@ -30,6 +31,8 @@ impl OpendalObjectStore {
         // NOTE: the root must be absolute path.
         builder.root(&root);
 
+        let atomic_write_dir = format!("{}/{}", root, ATOMIC_WRITE_DIR);
+        builder.atomic_write_dir(&atomic_write_dir);
         let op: Operator = Operator::new(builder)?
             .layer(LoggingLayer::default())
             .layer(RetryLayer::default())