[WIP] opendal writes

Author: comphead

native/Cargo.lock

@@ -340,7 +340,7 @@ impl ParquetWriterExec {
                     .unwrap_or(output_file_path);
 
                 // Create output directory
-                std::fs::create_dir_all(&local_path).map_err(|e| {
+                std::fs::create_dir_all(local_path).map_err(|e| {
                     DataFusionError::Execution(format!(
                         "Failed to create output directory '{}': {}",
                         local_path, e

native/core/src/execution/operators/parquet_writer.rs

@@ -39,10 +39,17 @@ use log4rs::{
 };
 use once_cell::sync::OnceCell;
 
-#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
+#[cfg(all(
+    not(target_env = "msvc"),
+    feature = "jemalloc",
+    not(feature = "mimalloc")
+))]
 use tikv_jemallocator::Jemalloc;
 
-#[cfg(feature = "mimalloc")]
+#[cfg(all(
+    feature = "mimalloc",
+    not(all(not(target_env = "msvc"), feature = "jemalloc"))
+))]
 use mimalloc::MiMalloc;
 
 use errors::{try_unwrap_or_throw, CometError, CometResult};
@@ -55,11 +62,18 @@ pub mod execution;
 mod jvm_bridge;
 pub mod parquet;
 
-#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
+#[cfg(all(
+    not(target_env = "msvc"),
+    feature = "jemalloc",
+    not(feature = "mimalloc")
+))]
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
 
-#[cfg(feature = "mimalloc")]
+#[cfg(all(
+    feature = "mimalloc",
+    not(all(not(target_env = "msvc"), feature = "jemalloc"))
+))]
 #[global_allocator]
 static GLOBAL: MiMalloc = MiMalloc;
 

native/core/src/lib.rs

@@ -370,7 +370,7 @@ fn is_hdfs_scheme(url: &Url, object_store_configs: &HashMap<String, String>) ->
 }
 
 // Creates an HDFS object store from a URL using the native HDFS implementation
-#[cfg(feature = "hdfs")]
+#[cfg(all(feature = "hdfs", not(feature = "hdfs-opendal")))]
 fn create_hdfs_object_store(
     url: &Url,
 ) -> Result<(Box<dyn ObjectStore>, Path), object_store::Error> {

native/core/src/parquet/parquet_support.rs

@@ -32,7 +32,7 @@ publish = false
 readme = "README.md"
 
 [lib]
-name = "hdfs"
+name = "fs_hdfs"
 path = "src/lib.rs"
 
 [features]

native/fs-hdfs/Cargo.toml

@@ -33,7 +33,7 @@ edition = { workspace = true }
 
 [features]
 default = ["hdfs", "try_spawn_blocking"]
-hdfs = ["fs-hdfs"]
+hdfs = ["fs_hdfs"]
 hdfs3 = ["fs-hdfs3"]
 # Used for trying to spawn a blocking thread for implementing each object store interface when running in a tokio runtime
 try_spawn_blocking = []
@@ -42,7 +42,7 @@ try_spawn_blocking = []
 async-trait = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true }
-fs-hdfs = { version = "^0.1.12", optional = true }
+fs_hdfs = { package = "datafusion-comet-fs-hdfs3", path = "../fs-hdfs", optional = true }
 fs-hdfs3 = { version = "^0.1.12", optional = true }
 futures = { workspace = true }
 object_store = { workspace = true }

native/hdfs/Cargo.toml

@@ -26,9 +26,9 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use bytes::Bytes;
 use chrono::{DateTime, Utc};
+use fs_hdfs::hdfs::{get_hdfs_by_full_path, FileStatus, HdfsErr, HdfsFile, HdfsFs};
+use fs_hdfs::walkdir::HdfsWalkDir;
 use futures::{stream::BoxStream, StreamExt, TryStreamExt};
-use hdfs::hdfs::{get_hdfs_by_full_path, FileStatus, HdfsErr, HdfsFile, HdfsFs};
-use hdfs::walkdir::HdfsWalkDir;
 use object_store::{
     path::{self, Path},
     Error, GetOptions, GetRange, GetResult, GetResultPayload, ListResult, MultipartUpload,
@@ -422,7 +422,7 @@ impl ObjectStore for HadoopFileSystem {
                 hdfs.delete(&to, false).map_err(to_error)?;
             }
 
-            hdfs::util::HdfsUtil::copy(hdfs.as_ref(), &from, hdfs.as_ref(), &to)
+            fs_hdfs::util::HdfsUtil::copy(hdfs.as_ref(), &from, hdfs.as_ref(), &to)
                 .map_err(to_error)?;
 
             Ok(())
@@ -437,7 +437,7 @@ impl ObjectStore for HadoopFileSystem {
         let to = HadoopFileSystem::path_to_filesystem(to);
 
         maybe_spawn_blocking(move || {
-            hdfs.rename(&from, &to).map_err(to_error)?;
+            hdfs.rename(&from, &to, true).map_err(to_error)?;
 
             Ok(())
         })
@@ -459,7 +459,7 @@ impl ObjectStore for HadoopFileSystem {
                 });
             }
 
-            hdfs::util::HdfsUtil::copy(hdfs.as_ref(), &from, hdfs.as_ref(), &to)
+            fs_hdfs::util::HdfsUtil::copy(hdfs.as_ref(), &from, hdfs.as_ref(), &to)
                 .map_err(to_error)?;
 
             Ok(())