Repository: Kimahriman/hdfs-native (language: python)

Issue:
Migrate object_store implementation to new repo
Migrate the object_store implementation to https://github.com/datafusion-contrib/hdfs-native-object-store


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index a59de21..61443f6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -70,17 +70,6 @@ version = "1.0.80"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5ad32ce52e4161730f7098c077cd2ed6229b5804ccf99e5366be1ab72a98b4e1"
 
-[[package]]
-name = "async-trait"
-version = "0.1.77"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c980ee35e870bd1a4d2c8294d4c04d0499e67bca1e4b5cefcc693c2fa00caea9"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.50",
-]
-
 [[package]]
 name = "autocfg"
 version = "1.1.0"
@@ -481,12 +470,6 @@ dependencies = [
  "subtle",
 ]
 
-[[package]]
-name = "doc-comment"
-version = "0.3.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fea41bba32d969b513997752735605054bc0dfa92b4c56bf1189f2e174be7a10"
-
 [[package]]
 name = "either"
 version = "1.10.0"
@@ -767,25 +750,6 @@ dependencies = [
  "which",
 ]
 
-[[package]]
-name = "hdfs-native-object-store"
-version = "0.9.1"
-dependencies = [
- "async-trait",
- "bytes",
- "chrono",
- "criterion",
- "env_logger",
- "futures",
- "hdfs-native",
- "object_store",
- "serial_test",
- "tempfile",
- "thiserror",
- "tokio",
- "which",
-]
-
 [[package]]
 name = "hdfs-native-python"
 version = "0.9.1"
@@ -1118,27 +1082,6 @@ dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "object_store"
-version = "0.10.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9cda97d37376c28c508b0abc56b855c5e700324e9b753226681e658c833675c5"
-dependencies = [
- "async-trait",
- "bytes",
- "chrono",
- "futures",
- "humantime",
- "itertools 0.12.1",
- "parking_lot",
- "percent-encoding",
- "snafu",
- "tokio",
- "tracing",
- "url",
- "walkdir",
-]
-
 [[package]]
 name = "once_cell"
 version = "1.19.0"
@@ -1620,28 +1563,6 @@ version = "1.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e6ecd384b10a64542d77071bd64bd7b231f4ed5940fba55e98c3de13824cf3d7"
 
-[[package]]
-name = "snafu"
-version = "0.7.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e4de37ad025c587a29e8f3f5605c00f70b98715ef90b9061a815b9e59e9042d6"
-dependencies = [
- "doc-comment",
- "snafu-derive",
-]
-
-[[package]]
-name = "snafu-derive"
-version = "0.7.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "990079665f075b699031e9c08fd3ab99be5029b96f3b78dc0709e8f77e4efebf"
-dependencies = [
- "heck",
- "proc-macro2",
- "quote",
- "syn 1.0.109",
-]
-
 [[package]]
 name = "socket2"
 version = "0.5.5"
@@ -1780,37 +1701,6 @@ dependencies = [
  "syn 2.0.50",
 ]
 
-[[package]]
-name = "tracing"
-version = "0.1.40"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3523ab5a71916ccf420eebdf5521fcef02141234bbc0b8a49f2fdc4544364ef"
-dependencies = [
- "pin-project-lite",
- "tracing-attributes",
- "tracing-core",
-]
-
-[[package]]
-name = "tracing-attributes"
-version = "0.1.27"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.50",
-]
-
-[[package]]
-name = "tracing-core"
-version = "0.1.32"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c06d3da6113f116aaee68e4d601191614c9053067f9ab7f6edbcb161237daa54"
-dependencies = [
- "once_cell",
-]
-
 [[package]]
 name = "typenum"
 version = "1.17.0"
diff --git a/Cargo.toml b/Cargo.toml
index 77f796b..fb0340a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,14 +1,12 @@
 [workspace]
 members = [
-    "crates/*",
+    "rust",
     "python",
 ]
 resolver = "2"
 
 [workspace.dependencies]
 bytes = "1"
-chrono = "0.4"
-futures = "0.3"
 log = "0.4"
 thiserror = "1"
 tokio = "1"
diff --git a/crates/hdfs-native-object-store/Cargo.toml b/crates/hdfs-native-object-store/Cargo.toml
deleted file mode 100644
index de76d8d..0000000
--- a/crates/hdfs-native-object-store/Cargo.toml
+++ /dev/null
@@ -1,33 +0,0 @@
-[package]
-name = "hdfs-native-object-store"
-version = "0.9.1"
-edition = "2021"
-authors = ["Adam Binford <adamq43@gmail.com>"]
-homepage = "https://github.com/Kimahriman/hdfs-native"
-repository = "https://github.com/Kimahriman/hdfs-native"
-keywords = ["hadoop", "hdfs", "object_store"]
-description = "object_store implementation for Native HDFS client"
-readme = "../../README.md"
-license = "Apache-2.0"
-
-[dependencies]
-async-trait = { version = "0.1" }
-bytes = { workspace = true }
-chrono = { workspace = true }
-futures = { workspace = true }
-hdfs-native = { path = "../hdfs-native", version = "0.9" }
-object_store = { version = "0.10" }
-thiserror = "1"
-tokio = { workspace = true, features = ["rt", "net", "io-util", "macros", "sync", "time"] }
-
-[dev-dependencies]
-criterion = "0.5"
-env_logger = "0.10"
-serial_test = "2.0.0"
-tempfile = "3"
-which = "4"
-
-[features]
-kerberos = ["hdfs-native/kerberos"]
-
-integration-test = ["hdfs-native/integration-test"]
\ No newline at end of file
diff --git a/crates/hdfs-native-object-store/README.md b/crates/hdfs-native-object-store/README.md
deleted file mode 100644
index 960b437..0000000
--- a/crates/hdfs-native-object-store/README.md
+++ /dev/null
@@ -1,14 +0,0 @@
-# Object store implementation for the Native Rust HDFS client
-
-## Usage
-
-```rust
-use hdfs_native::Client;
-use hdfs_native_object_store::HdfsObjectStore;
-use hdfs_native::Result;
-fn main() -> Result<()> {
-    let client = Client::new("hdfs://localhost:9000")?;
-    let store = HdfsObjectStore::new(client);
-Ok(())
-}
-```
\ No newline at end of file
diff --git a/crates/hdfs-native-object-store/src/lib.rs b/crates/hdfs-native-object-store/src/lib.rs
deleted file mode 100644
index e52132a..0000000
--- a/crates/hdfs-native-object-store/src/lib.rs
+++ /dev/null
@@ -1,581 +0,0 @@
-// #![warn(missing_docs)]
-//! Object store implementation for the Native Rust HDFS client
-//!
-//! # Usage
-//!
-//! ```rust
-//! use hdfs_native_object_store::HdfsObjectStore;
-//! # use object_store::Result;
-//! # fn main() -> Result<()> {
-//! let store = HdfsObjectStore::with_url("hdfs://localhost:9000")?;
-//! # Ok(())
-//! # }
-//! ```
-use std::{
-    fmt::{Display, Formatter},
-    future,
-    path::PathBuf,
-    sync::Arc,
-};
-
-use async_trait::async_trait;
-use chrono::{NaiveDateTime, TimeZone, Utc};
-use futures::{
-    stream::{BoxStream, StreamExt},
-    FutureExt,
-};
-use hdfs_native::{client::FileStatus, file::FileWriter, Client, HdfsError, WriteOptions};
-use object_store::{
-    path::Path, GetOptions, GetRange, GetResult, GetResultPayload, ListResult, MultipartUpload,
-    ObjectMeta, ObjectStore, PutMode, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
-    UploadPart,
-};
-use tokio::{
-    sync::{mpsc, oneshot},
-    task::{self, JoinHandle},
-};
-
-#[derive(Debug)]
-pub struct HdfsObjectStore {
-    client: Arc<Client>,
-}
-
-impl HdfsObjectStore {
-    pub fn new(client: Arc<Client>) -> Self {
-        Self { client }
-    }
-
-    pub fn with_url(url: &str) -> Result<Self> {
-        Ok(Self {
-            client: Arc::new(Client::new(url).to_object_store_err()?),
-        })
-    }
-
-    async fn internal_copy(&self, from: &Path, to: &Path, overwrite: bool) -> Result<()> {
-        let overwrite = match self.client.get_file_info(&make_absolute_file(to)).await {
-            Ok(_) if overwrite => true,
-            Ok(_) => Err(HdfsError::AlreadyExists(make_absolute_file(to))).to_object_store_err()?,
-            Err(HdfsError::FileNotFound(_)) => false,
-            Err(e) => Err(e).to_object_store_err()?,
-        };
-
-        let write_options = WriteOptions {
-            overwrite,
-            ..Default::default()
-        };
-
-        let file = self
-            .client
-            .read(&make_absolute_file(from))
-            .await
-            .to_object_store_err()?;
-        let mut stream = file.read_range_stream(0, file.file_length()).boxed();
-
-        let mut new_file = self
-            .client
-            .create(&make_absolute_file(to), write_options)
-            .await
-            .to_object_store_err()?;
-
-        while let Some(bytes) = stream.next().await.transpose().to_object_store_err()? {
-            new_file.write(bytes).await.to_object_store_err()?;
-        }
-        new_file.close().await.to_object_store_err()?;
-
-        Ok(())
-    }
-
-    async fn open_tmp_file(&self, file_path: &str) -> Result<(FileWriter, String)> {
-        let path_buf = PathBuf::from(file_path);
-
-        let file_name = path_buf
-            .file_name()
-            .ok_or(HdfsError::InvalidPath("path missing filename".to_string()))
-            .to_object_store_err()?
-            .to_str()
-            .ok_or(HdfsError::InvalidPath("path not valid unicode".to_string()))
-            .to_object_store_err()?
-            .to_string();
-
-        let tmp_file_path = path_buf
-            .with_file_name(format!(".{}.tmp", file_name))
-            .to_str()
-            .ok_or(HdfsError::InvalidPath("path not valid unicode".to_string()))
-            .to_object_store_err()?
-            .to_string();
-
-        // Try to create a file with an incrementing index until we find one that doesn't exist yet
-        let mut index = 1;
-        loop {
-            let path = format!("{}.{}", tmp_file_path, index);
-            match self.client.create(&path, WriteOptions::default()).await {
-                Ok(writer) => break Ok((writer, path)),
-                Err(HdfsError::AlreadyExists(_)) => index += 1,
-                Err(e) => break Err(e).to_object_store_err(),
-            }
-        }
-    }
-}
-
-impl Display for HdfsObjectStore {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "HdfsObjectStore")
-    }
-}
-
-impl From<Client> for HdfsObjectStore {
-    fn from(value: Client) -> Self {
-        Self::new(Arc::new(value))
-    }
-}
-
-#[async_trait]
-impl ObjectStore for HdfsObjectStore {
-    /// Save the provided bytes to the specified location
-    ///
-    /// To make the operation atomic, we write to a temporary file ".{filename}.tmp.{i}" and rename
-    /// on a successful write.
-    async fn put_opts(
-        &self,
-        location: &Path,
-        payload: PutPayload,
-        opts: PutOptions,
-    ) -> Result<PutResult> {
-        let overwrite = match opts.mode {
-            PutMode::Create => false,
-            PutMode::Overwrite => true,
-            PutMode::Update(_) => {
-                return Err(object_store::Error::NotSupported {
-                    source: "Update mode not supported".to_string().into(),
-                })
-            }
-        };
-
-        let final_file_path = make_absolute_file(location);
-
-        // If we're not overwriting, do an upfront check to see if the file already
-        // exists. Otherwise we have to write the whole file and try to rename before
-        // finding out.
-        if !overwrite && self.client.get_file_info(&final_file_path).await.is_ok() {
-            return Err(HdfsError::AlreadyExists(final_file_path)).to_object_store_err();
-        }
-
-        let (mut tmp_file, tmp_file_path) = self.open_tmp_file(&final_file_path).await?;
-
-        for bytes in payload {
-            tmp_file.write(bytes).await.to_object_store_err()?;
-        }
-        tmp_file.close().await.to_object_store_err()?;
-
-        self.client
-            .rename(&tmp_file_path, &final_file_path, overwrite)
-            .await
-            .to_object_store_err()?;
-
-        Ok(PutResult {
-            e_tag: None,
-            version: None,
-        })
-    }
-
-    /// Create a multipart writer that writes to a temporary file in a background task, and renames
-    /// to the final destination on complete.
-    async fn put_multipart_opts(
-        &self,
-        location: &Path,
-        _opts: PutMultipartOpts,
-    ) -> Result<Box<dyn MultipartUpload>> {
-        let final_file_path = make_absolute_file(location);
-
-        let (tmp_file, tmp_file_path) = self.open_tmp_file(&final_file_path).await?;
-
-        Ok(Box::new(HdfsMultipartWriter::new(
-            Arc::clone(&self.client),
-            tmp_file,
-            &tmp_file_path,
-            &final_file_path,
-        )))
-    }
-
-    async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
-        if options.if_match.is_some()
-            || options.if_none_match.is_some()
-            || options.if_modified_since.is_some()
-            || options.if_unmodified_since.is_some()
-        {
-            return Err(object_store::Error::NotImplemented);
-        }
-
-        let meta = self.head(location).await?;
-
-        let range = options
-            .range
-            .map(|r| match r {
-                GetRange::Bounded(range) => range,
-                GetRange::Offset(offset) => offset..meta.size,
-                GetRange::Suffix(suffix) => meta.size.saturating_sub(suffix)..meta.size,
-            })
-            .unwrap_or(0..meta.size);
-
-        let reader = self
-            .client
-            .read(&make_absolute_file(location))
-            .await
-            .to_object_store_err()?;
-        let stream = reader
-            .read_range_stream(range.start, range.end - range.start)
-            .map(|b| b.to_object_store_err())
-            .boxed();
-
-        let payload = GetResultPayload::Stream(stream);
-
-        Ok(GetResult {
-            payload,
-            meta,
-            range,
-            attributes: Default::default(),
-        })
-    }
-
-    /// Return the metadata for the specified location
-    async fn head(&self, location: &Path) -> Result<ObjectMeta> {
-        let status = self
-            .client
-            .get_file_info(&make_absolute_file(location))
-            .await
-            .to_object_store_err()?;
-
-        Ok(ObjectMeta {
-            location: location.clone(),
-            last_modified: Utc.from_utc_datetime(
-                &NaiveDateTime::from_timestamp_opt(status.modification_time as i64, 0).unwrap(),
-            ),
-            size: status.length,
-            e_tag: None,
-            version: None,
-        })
-    }
-
-    /// Delete the object at the specified location.
-    async fn delete(&self, location: &Path) -> Result<()> {
-        let result = self
-            .client
-            .delete(&make_absolute_file(location), false)
-            .await
-            .to_object_store_err()?;
-
-        if !result {
-            Err(HdfsError::OperationFailed(
-                "failed to delete object".to_string(),
-            ))
-            .to_object_store_err()?
-        }
-
-        Ok(())
-    }
-
-    /// List all the objects with the given prefix.
-    ///
-    /// Prefixes are evaluated on a path segment basis, i.e. `foo/bar/` is a prefix of `foo/bar/x` but not of
-    /// `foo/bar_baz/x`.
-    ///
-    /// Note: the order of returned [`ObjectMeta`] is not guaranteed
-    fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, Result<ObjectMeta>> {
-        let status_stream = self
-            .client
-            .list_status_iter(
-                &prefix.map(make_absolute_dir).unwrap_or("".to_string()),
-                true,
-            )
-            .into_stream()
-            .filter(|res| {
-                let result = match res {
-                    Ok(status) => !status.isdir,
-                    // Listing by prefix should just return an empty list if the prefix isn't found
-                    Err(HdfsError::FileNotFound(_)) => false,
-                    _ => true,
-                };
-                future::ready(result)
-            })
-            .map(|res| res.map_or_else(|e| Err(e).to_object_store_err(), |s| get_object_meta(&s)));
-
-        Box::pin(status_stream)
-    }
-
-    /// List objects with the given prefix and an implementation specific
-    /// delimiter. Returns common prefixes (directories) in addition to object
-    /// metadata.
-    ///
-    /// Prefixes are evaluated on a path segment basis, i.e. `foo/bar/` is a prefix of `foo/bar/x` but not of
-    /// `foo/bar_baz/x`.
-    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
-        let mut status_stream = self
-            .client
-            .list_status_iter(
-                &prefix.map(make_absolute_dir).unwrap_or("".to_string()),
-                false,
-            )
-            .into_stream()
-            .filter(|res| {
-                let result = match res {
-                    // Listing by prefix should just return an empty list if the prefix isn't found
-                    Err(HdfsError::FileNotFound(_)) => false,
-                    _ => true,
-                };
-                future::ready(result)
-            });
-
-        let mut statuses = Vec::<FileStatus>::new();
-        while let Some(status) = status_stream.next().await {
-            statuses.push(status.to_object_store_err()?);
-        }
-
-        let mut dirs: Vec<Path> = Vec::new();
-        for status in statuses.iter().filter(|s| s.isdir) {
-            dirs.push(Path::parse(&status.path)?)
-        }
-
-        let mut files: Vec<ObjectMeta> = Vec::new();
-        for status in statuses.iter().filter(|s| !s.isdir) {
-            files.push(get_object_meta(status)?)
-        }
-
-        Ok(ListResult {
-            common_prefixes: dirs,
-            objects: files,
-        })
-    }
-
-    async fn rename(&self, from: &Path, to: &Path) -> Result<()> {
-        Ok(self
-            .client
-            .rename(&make_absolute_file(from), &make_absolute_file(to), true)
-            .await
-            .to_object_store_err()?)
-    }
-
-    async fn rename_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
-        Ok(self
-            .client
-            .rename(&make_absolute_file(from), &make_absolute_file(to), false)
-            .await
-            .to_object_store_err()?)
-    }
-
-    /// Copy an object from one path to another in the same object store.
-    ///
-    /// If there exists an object at the destination, it will be overwritten.
-    async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
-        self.internal_copy(from, to, true).await
-    }
-
-    /// Copy an object from one path to another, only if destination is empty.
-    ///
-    /// Will return an error if the destination already has an object.
-    ///
-    /// Performs an atomic operation if the underlying object storage supports it.
-    /// If atomic operations are not supported by the underlying object storage (like S3)
-    /// it will return an error.
-    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
-        self.internal_copy(from, to, false).await
-    }
-}
-
-#[cfg(feature = "integration-test")]
-pub trait HdfsErrorConvert<T> {
-    fn to_object_store_err(self) -> Result<T>;
-}
-
-#[cfg(not(feature = "integration-test"))]
-trait HdfsErrorConvert<T> {
-    fn to_object_store_err(self) -> Result<T>;
-}
-
-impl<T> HdfsErrorConvert<T> for hdfs_native::Result<T> {
-    fn to_object_store_err(self) -> Result<T> {
-        self.map_err(|err| match err {
-            HdfsError::FileNotFound(path) => object_store::Error::NotFound {
-                path: path.clone(),
-                source: Box::new(HdfsError::FileNotFound(path)),
-            },
-            HdfsError::AlreadyExists(path) => object_store::Error::AlreadyExists {
-                path: path.clone(),
-                source: Box::new(HdfsError::AlreadyExists(path)),
-            },
-            _ => object_store::Error::Generic {
-                store: "HdfsObjectStore",
-                source: Box::new(err),
-            },
-        })
-    }
-}
-
-type PartSender = mpsc::UnboundedSender<(oneshot::Sender<Result<()>>, PutPayload)>;
-
-// Create a fake multipart writer the creates an uploader to a temp file as a background
-// task, and submits new parts to be uploaded to a queue for this task.
-// A once cell is used to track whether a part has finished writing or not.
-// On completing, rename the file to the actual target.
-struct HdfsMultipartWriter {
-    client: Arc<Client>,
-    sender: Option<(JoinHandle<Result<()>>, PartSender)>,
-    tmp_filename: String,
-    final_filename: String,
-}
-
-impl HdfsMultipartWriter {
-    fn new(
-        client: Arc<Client>,
-        writer: FileWriter,
-        tmp_filename: &str,
-        final_filename: &str,
-    ) -> Self {
-        let (sender, receiver) = mpsc::unbounded_channel();
-
-        Self {
-            client,
-            sender: Some((Self::start_writer_task(writer, receiver), sender)),
-            tmp_filename: tmp_filename.to_string(),
-            final_filename: final_filename.to_string(),
-        }
-    }
-
-    fn start_writer_task(
-        mut writer: FileWriter,
-        mut part_receiver: mpsc::UnboundedReceiver<(oneshot::Sender<Result<()>>, PutPayload)>,
-    ) -> JoinHandle<Result<()>> {
-        task::spawn(async move {
-            'outer: loop {
-                match part_receiver.recv().await {
-                    Some((sender, part)) => {
-                        for bytes in part {
-                            if let Err(e) = writer.write(bytes).await.to_object_store_err() {
-                                let _ = sender.send(Err(e));
-                                break 'outer;
-                            }
-                        }
-                        let _ = sender.send(Ok(()));
-                    }
-                    None => {
-                        return writer.close().await.to_object_store_err();
-                    }
-                }
-            }
-
-            // If we've reached here, a write task failed so just return Err's for all new parts that come in
-            while let Some((sender, _)) = part_receiver.recv().await {
-                let _ = sender.send(
-                    Err(HdfsError::OperationFailed(
-                        "Write failed during one of the parts".to_string(),
-                    ))
-                    .to_object_store_err(),
-                );
-            }
-            Err(HdfsError::OperationFailed(
-                "Write failed during one of the parts".to_string(),
-            ))
-            .to_object_store_err()
-        })
-    }
-}
-
-impl std::fmt::Debug for HdfsMultipartWriter {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("HdfsMultipartWriter")
-            .field("tmp_filename", &self.tmp_filename)
-            .field("final_filename", &self.final_filename)
-            .finish()
-    }
-}
-
-#[async_trait]
-impl MultipartUpload for HdfsMultipartWriter {
-    fn put_part(&mut self, payload: PutPayload) -> UploadPart {
-        let (result_sender, result_receiver) = oneshot::channel();
-
-        if let Some((_, payload_sender)) = self.sender.as_ref() {
-            payload_sender.send((result_sender, payload)).unwrap();
-        } else {
-            result_sender
-                .send(
-                    Err(HdfsError::OperationFailed(
-                        "Cannot put part after completing or aborting".to_string(),
-                    ))
-                    .to_object_store_err(),
-                )
-                .unwrap();
-        }
-
-        async { result_receiver.await.unwrap() }.boxed()
-    }
-
-    async fn complete(&mut self) -> Result<PutResult> {
-        // Drop the sender so the task knows no more data is coming
-        if let Some((handle, sender)) = self.sender.take() {
-            drop(sender);
-
-            // Wait for the writer task to finish
-            handle.await??;
-
-            self.client
-                .rename(&self.tmp_filename, &self.final_filename, true)
-                .await
-                .to_object_store_err()?;
-
-            Ok(PutResult {
-                e_tag: None,
-                version: None,
-            })
-        } else {
-            Err(object_store::Error::NotSupported {
-                source: "Cannot call abort or complete multiple times".into(),
-            })
-        }
-    }
-
-    async fn abort(&mut self) -> Result<()> {
-        // Drop the sender so the task knows no more data is coming
-        if let Some((handle, sender)) = self.sender.take() {
-            drop(sender);
-
-            // Wait for the writer task to finish
-            handle.abort();
-
-            self.client
-                .delete(&self.tmp_filename, false)
-                .await
-                .to_object_store_err()?;
-
-            Ok(())
-        } else {
-            Err(object_store::Error::NotSupported {
-                source: "Cannot call abort or complete multiple times".into(),
-            })
-        }
-    }
-}
-
-/// ObjectStore paths always remove the leading slash, so add it back
-fn make_absolute_file(path: &Path) -> String {
-    format!("/{}", path.as_ref())
-}
-
-fn make_absolute_dir(path: &Path) -> String {
-    if path.parts().count() > 0 {
-        format!("/{}/", path.as_ref())
-    } else {
-        "/".to_string()
-    }
-}
-
-fn get_object_meta(status: &FileStatus) -> Result<ObjectMeta> {
-    Ok(ObjectMeta {
-        location: Path::parse(&status.path)?,
-        last_modified: Utc.from_utc_datetime(
-            &NaiveDateTime::from_timestamp_opt(status.modification_time as i64, 0).unwrap(),
-        ),
-        size: status.length,
-        e_tag: None,
-        version: None,
-    })
-}
diff --git a/crates/hdfs-native/src/error.rs b/crates/hdfs-native/src/error.rs
deleted file mode 100644
index e2aff56..0000000
--- a/crates/hdfs-native/src/error.rs
+++ /dev/null
@@ -1,55 +0,0 @@
-use std::io;
-
-#[cfg(feature = "kerberos")]
-use libgssapi::error::Error as GssapiError;
-use prost::DecodeError;
-use thiserror::Error;
-
-#[derive(Error, Debug)]
-pub enum HdfsError {
-    #[error("IO error occurred while communicating with HDFS")]
-    IOError(#[from] io::Error),
-    #[error("data transfer error")]
-    DataTransferError(String),
-    #[error("checksums didn't match")]
-    ChecksumError,
-    #[error("invalid path")]
-    InvalidPath(String),
-    #[error("invalid argument")]
-    InvalidArgument(String),
-    #[error("failed to parse URL")]
-    UrlParseError(#[from] url::ParseError),
-    #[error("file already exists")]
-    AlreadyExists(String),
-    #[error("operation failed")]
-    OperationFailed(String),
-    #[error("file not found")]
-    FileNotFound(String),
-    #[error("blocks not found")]
-    BlocksNotFound(String),
-    #[error("path is a directory")]
-    IsADirectoryError(String),
-    #[error("unsupported erasure coding policy")]
-    UnsupportedErasureCodingPolicy(String),
-    #[error("erasure coding error")]
-    ErasureCodingError(String),
-    #[error("operation not supported")]
-    UnsupportedFeature(String),
-    #[error("interal error, this shouldn't happen")]
-    InternalError(String),
-    #[error("failed to decode RPC response")]
-    InvalidRPCResponse(#[from] DecodeError),
-    #[error("RPC error")]
-    RPCError(String, String),
-    #[error("fatal RPC error")]
-    FatalRPCError(String, String),
-    #[error("SASL error")]
-    SASLError(String),
-    #[cfg(feature = "kerberos")]
-    #[error("GSSAPI error")]
-    GSSAPIError(#[from] GssapiError),
-    #[error("No valid SASL mechanism found")]
-    NoSASLMechanism,
-}
-
-pub type Result<T> = std::result::Result<T, HdfsError>;
diff --git a/python/Cargo.toml b/python/Cargo.toml
index d69dc5b..6756625 100644
--- a/python/Cargo.toml
+++ b/python/Cargo.toml
@@ -27,8 +27,8 @@ name = "hdfs_native._internal"
 [dependencies]
 bytes = { workspace = true } 
 env_logger = "0.10"
-hdfs-native = { path = "../crates/hdfs-native" }
-log = "0.4"
+hdfs-native = { path = "../rust" }
+log = { workspace = true }
 pyo3 = { version = "0.20", features = ["extension-module", "abi3", "abi3-py38"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread"] }
diff --git a/crates/hdfs-native/Cargo.toml b/rust/Cargo.toml
similarity index 95%
rename from crates/hdfs-native/Cargo.toml
rename to rust/Cargo.toml
index f3ac4a7..acc9d36 100644
--- a/crates/hdfs-native/Cargo.toml
+++ b/rust/Cargo.toml
@@ -15,14 +15,14 @@ aes = "0.8"
 base64 = "0.21"
 bytes = { workspace = true }
 cbc = "0.1"
-chrono = { workspace = true }
+chrono = "0.4"
 cipher = "0.4"
 crc = "3.2"
 ctr = "0.9"
 des = "0.8"
 # Just used for benchmarks
 fs-hdfs3 = { version = "0.1.12", optional = true }
-futures = { workspace = true }
+futures = "0.3"
 g2p = "1"
 hex = "0.4"
 hmac = "0.12"
@@ -52,7 +52,7 @@ protobuf-src = { version = "1.1", optional = true }
 [dev-dependencies]
 criterion = { version = "0.5", features = ["async_tokio", "async_futures"] }
 env_logger = "0.10"
-serial_test = "2.0.0"
+serial_test = "2"
 tempfile = "3"
 which = "4"
 
diff --git a/crates/hdfs-native/benches/ec.rs b/rust/benches/ec.rs
similarity index 100%
rename from crates/hdfs-native/benches/ec.rs
rename to rust/benches/ec.rs
diff --git a/crates/hdfs-native/benches/io.rs b/rust/benches/io.rs
similarity index 100%
rename from crates/hdfs-native/benches/io.rs
rename to rust/benches/io.rs
diff --git a/crates/hdfs-native/benches/rpc.rs b/rust/benches/rpc.rs
similarity index 100%
rename from crates/hdfs-native/benches/rpc.rs
rename to rust/benches/rpc.rs
diff --git a/crates/hdfs-native/build.rs b/rust/build.rs
similarity index 100%
rename from crates/hdfs-native/build.rs
rename to rust/build.rs
diff --git a/crates/hdfs-native/examples/simple.rs b/rust/examples/simple.rs
similarity index 100%
rename from crates/hdfs-native/examples/simple.rs
rename to rust/examples/simple.rs
diff --git a/crates/hdfs-native/minidfs/pom.xml b/rust/minidfs/pom.xml
similarity index 100%
rename from crates/hdfs-native/minidfs/pom.xml
rename to rust/minidfs/pom.xml
diff --git a/crates/hdfs-native/minidfs/src/main/java/main/Main.java b/rust/minidfs/src/main/java/main/Main.java
similarity index 100%
rename from crates/hdfs-native/minidfs/src/main/java/main/Main.java
rename to rust/minidfs/src/main/java/main/Main.java
diff --git a/crates/hdfs-native/minidfs/src/main/resources/log4j.properties b/rust/minidfs/src/main/resources/log4j.properties
similarity index 100%
rename from crates/hdfs-native/minidfs/src/main/resources/log4j.properties
rename to rust/minidfs/src/main/resources/log4j.properties
diff --git a/crates/hdfs-native/src/client.rs b/rust/src/client.rs
similarity index 100%
rename from crates/hdfs-native/src/client.rs
rename to rust/src/client.rs
diff --git a/crates/hdfs-native/src/common/config.rs b/rust/src/common/config.rs
similarity index 100%
rename from crates/hdfs-native/src/common/config.rs
rename to rust/src/common/config.rs
diff --git a/crates/hdfs-native/src/common/mod.rs b/rust/src/common/mod.rs
similarity index 100%
rename from crates/hdfs-native/src/common/mod.rs
rename to rust/src/common/mod.rs
diff --git a/crates/hdfs-native/src/ec/gf256.rs b/rust/src/ec/gf256.rs
similarity index 100%
rename from crates/hdfs-native/src/ec/gf256.rs
rename to rust/src/ec/gf256.rs
diff --git a/crates/hdfs-native/src/ec/matrix.rs b/rust/src/ec/matrix.rs
similarity index 100%
rename from crates/hdfs-native/src/ec/matrix.rs
rename to rust/src/ec/matrix.rs
diff --git a/crates/hdfs-native/src/ec/mod.rs b/rust/src/ec/mod.rs
similarity index 100%
rename from crates/hdfs-native/src/ec/mod.rs
rename to rust/src/ec/mod.rs
diff --git a/crates/hdfs-native-object-store/src/error.rs b/rust/src/error.rs
similarity index 100%
rename from crates/hdfs-native-object-store/src/error.rs
rename to rust/src/error.rs
diff --git a/crates/hdfs-native/src/file.rs b/rust/src/file.rs
similarity index 100%
rename from crates/hdfs-native/src/file.rs
rename to rust/src/file.rs
diff --git a/crates/hdfs-native/src/hdfs/block_reader.rs b/rust/src/hdfs/block_reader.rs
similarity index 100%
rename from crates/hdfs-native/src/hdfs/block_reader.rs
rename to rust/src/hdfs/block_reader.rs
diff --git a/crates/hdfs-native/src/hdfs/block_writer.rs b/rust/src/hdfs/block_writer.rs
similarity index 100%
rename from crates/hdfs-native/src/hdfs/block_writer.rs
rename to rust/src/hdfs/block_writer.rs
diff --git a/crates/hdfs-native/src/hdfs/connection.rs b/rust/src/hdfs/connection.rs
similarity index 100%
rename from crates/hdfs-native/src/hdfs/connection.rs
rename to rust/src/hdfs/connection.rs
diff --git a/crates/hdfs-native/src/hdfs/mod.rs b/rust/src/hdfs/mod.rs
similarity index 100%
rename from crates/hdfs-native/src/hdfs/mod.rs
rename to rust/src/hdfs/mod.rs
diff --git a/crates/hdfs-native/src/hdfs/protocol.rs b/rust/src/hdfs/protocol.rs
similarity index 100%
rename from crates/hdfs-native/src/hdfs/protocol.rs
rename to rust/src/hdfs/protocol.rs
diff --git a/crates/hdfs-native/src/hdfs/proxy.rs b/rust/src/hdfs/proxy.rs
similarity index 100%
rename from crates/hdfs-native/src/hdfs/proxy.rs
rename to rust/src/hdfs/proxy.rs
diff --git a/crates/hdfs-native/src/lib.rs b/rust/src/lib.rs
similarity index 100%
rename from crates/hdfs-native/src/lib.rs
rename to rust/src/lib.rs
diff --git a/crates/hdfs-native/src/minidfs.rs b/rust/src/minidfs.rs
similarity index 100%
rename from crates/hdfs-native/src/minidfs.rs
rename to rust/src/minidfs.rs
diff --git a/crates/hdfs-native/src/proto/common/FSProtos.proto b/rust/src/proto/common/FSProtos.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/FSProtos.proto
rename to rust/src/proto/common/FSProtos.proto
diff --git a/crates/hdfs-native/src/proto/common/GenericRefreshProtocol.proto b/rust/src/proto/common/GenericRefreshProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/GenericRefreshProtocol.proto
rename to rust/src/proto/common/GenericRefreshProtocol.proto
diff --git a/crates/hdfs-native/src/proto/common/GetUserMappingsProtocol.proto b/rust/src/proto/common/GetUserMappingsProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/GetUserMappingsProtocol.proto
rename to rust/src/proto/common/GetUserMappingsProtocol.proto
diff --git a/crates/hdfs-native/src/proto/common/HAServiceProtocol.proto b/rust/src/proto/common/HAServiceProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/HAServiceProtocol.proto
rename to rust/src/proto/common/HAServiceProtocol.proto
diff --git a/crates/hdfs-native/src/proto/common/IpcConnectionContext.proto b/rust/src/proto/common/IpcConnectionContext.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/IpcConnectionContext.proto
rename to rust/src/proto/common/IpcConnectionContext.proto
diff --git a/crates/hdfs-native/src/proto/common/ProtobufRpcEngine.proto b/rust/src/proto/common/ProtobufRpcEngine.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/ProtobufRpcEngine.proto
rename to rust/src/proto/common/ProtobufRpcEngine.proto
diff --git a/crates/hdfs-native/src/proto/common/ProtobufRpcEngine2.proto b/rust/src/proto/common/ProtobufRpcEngine2.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/ProtobufRpcEngine2.proto
rename to rust/src/proto/common/ProtobufRpcEngine2.proto
diff --git a/crates/hdfs-native/src/proto/common/ProtocolInfo.proto b/rust/src/proto/common/ProtocolInfo.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/ProtocolInfo.proto
rename to rust/src/proto/common/ProtocolInfo.proto
diff --git a/crates/hdfs-native/src/proto/common/RefreshAuthorizationPolicyProtocol.proto b/rust/src/proto/common/RefreshAuthorizationPolicyProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/RefreshAuthorizationPolicyProtocol.proto
rename to rust/src/proto/common/RefreshAuthorizationPolicyProtocol.proto
diff --git a/crates/hdfs-native/src/proto/common/RefreshCallQueueProtocol.proto b/rust/src/proto/common/RefreshCallQueueProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/RefreshCallQueueProtocol.proto
rename to rust/src/proto/common/RefreshCallQueueProtocol.proto
diff --git a/crates/hdfs-native/src/proto/common/RefreshUserMappingsProtocol.proto b/rust/src/proto/common/RefreshUserMappingsProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/RefreshUserMappingsProtocol.proto
rename to rust/src/proto/common/RefreshUserMappingsProtocol.proto
diff --git a/crates/hdfs-native/src/proto/common/RpcHeader.proto b/rust/src/proto/common/RpcHeader.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/RpcHeader.proto
rename to rust/src/proto/common/RpcHeader.proto
diff --git a/crates/hdfs-native/src/proto/common/Security.proto b/rust/src/proto/common/Security.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/Security.proto
rename to rust/src/proto/common/Security.proto
diff --git a/crates/hdfs-native/src/proto/common/TraceAdmin.proto b/rust/src/proto/common/TraceAdmin.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/TraceAdmin.proto
rename to rust/src/proto/common/TraceAdmin.proto
diff --git a/crates/hdfs-native/src/proto/common/ZKFCProtocol.proto b/rust/src/proto/common/ZKFCProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/common/ZKFCProtocol.proto
rename to rust/src/proto/common/ZKFCProtocol.proto
diff --git a/crates/hdfs-native/src/proto/hadoop.common.rs b/rust/src/proto/hadoop.common.rs
similarity index 100%
rename from crates/hdfs-native/src/proto/hadoop.common.rs
rename to rust/src/proto/hadoop.common.rs
diff --git a/crates/hdfs-native/src/proto/hadoop.hdfs.rs b/rust/src/proto/hadoop.hdfs.rs
similarity index 100%
rename from crates/hdfs-native/src/proto/hadoop.hdfs.rs
rename to rust/src/proto/hadoop.hdfs.rs
diff --git a/crates/hdfs-native/src/proto/hdfs/ClientDatanodeProtocol.proto b/rust/src/proto/hdfs/ClientDatanodeProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/ClientDatanodeProtocol.proto
rename to rust/src/proto/hdfs/ClientDatanodeProtocol.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/ClientNamenodeProtocol.proto b/rust/src/proto/hdfs/ClientNamenodeProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/ClientNamenodeProtocol.proto
rename to rust/src/proto/hdfs/ClientNamenodeProtocol.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/ReconfigurationProtocol.proto b/rust/src/proto/hdfs/ReconfigurationProtocol.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/ReconfigurationProtocol.proto
rename to rust/src/proto/hdfs/ReconfigurationProtocol.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/acl.proto b/rust/src/proto/hdfs/acl.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/acl.proto
rename to rust/src/proto/hdfs/acl.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/datatransfer.proto b/rust/src/proto/hdfs/datatransfer.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/datatransfer.proto
rename to rust/src/proto/hdfs/datatransfer.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/encryption.proto b/rust/src/proto/hdfs/encryption.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/encryption.proto
rename to rust/src/proto/hdfs/encryption.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/erasurecoding.proto b/rust/src/proto/hdfs/erasurecoding.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/erasurecoding.proto
rename to rust/src/proto/hdfs/erasurecoding.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/hdfs.proto b/rust/src/proto/hdfs/hdfs.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/hdfs.proto
rename to rust/src/proto/hdfs/hdfs.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/inotify.proto b/rust/src/proto/hdfs/inotify.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/inotify.proto
rename to rust/src/proto/hdfs/inotify.proto
diff --git a/crates/hdfs-native/src/proto/hdfs/xattr.proto b/rust/src/proto/hdfs/xattr.proto
similarity index 100%
rename from crates/hdfs-native/src/proto/hdfs/xattr.proto
rename to rust/src/proto/hdfs/xattr.proto
diff --git a/crates/hdfs-native/src/proto/mod.rs b/rust/src/proto/mod.rs
similarity index 100%
rename from crates/hdfs-native/src/proto/mod.rs
rename to rust/src/proto/mod.rs
diff --git a/crates/hdfs-native/src/security/digest.rs b/rust/src/security/digest.rs
similarity index 100%
rename from crates/hdfs-native/src/security/digest.rs
rename to rust/src/security/digest.rs
diff --git a/crates/hdfs-native/src/security/gssapi.rs b/rust/src/security/gssapi.rs
similarity index 100%
rename from crates/hdfs-native/src/security/gssapi.rs
rename to rust/src/security/gssapi.rs
diff --git a/crates/hdfs-native/src/security/mod.rs b/rust/src/security/mod.rs
similarity index 100%
rename from crates/hdfs-native/src/security/mod.rs
rename to rust/src/security/mod.rs
diff --git a/crates/hdfs-native/src/security/sasl.rs b/rust/src/security/sasl.rs
similarity index 100%
rename from crates/hdfs-native/src/security/sasl.rs
rename to rust/src/security/sasl.rs
diff --git a/crates/hdfs-native/src/security/user.rs b/rust/src/security/user.rs
similarity index 100%
rename from crates/hdfs-native/src/security/user.rs
rename to rust/src/security/user.rs


Test patch:
diff --git a/crates/hdfs-native-object-store/tests/test_object_store.rs b/crates/hdfs-native-object-store/tests/test_object_store.rs
deleted file mode 100644
index 805bc5c..0000000
--- a/crates/hdfs-native-object-store/tests/test_object_store.rs
+++ /dev/null
@@ -1,299 +0,0 @@
-#[cfg(feature = "integration-test")]
-mod test {
-    use bytes::{Buf, BufMut, BytesMut};
-    use hdfs_native::{
-        minidfs::{DfsFeatures, MiniDfs},
-        Client, WriteOptions,
-    };
-    use hdfs_native_object_store::{HdfsErrorConvert, HdfsObjectStore};
-    use object_store::{PutMode, PutOptions, PutPayload};
-    use serial_test::serial;
-    use std::{collections::HashSet, sync::Arc};
-
-    pub const TEST_FILE_INTS: usize = 64 * 1024 * 1024;
-
-    #[tokio::test]
-    #[serial]
-    async fn test_object_store() -> object_store::Result<()> {
-        let dfs = MiniDfs::with_features(&HashSet::from([DfsFeatures::HA]));
-        let client = Client::new(&dfs.url).to_object_store_err()?;
-
-        // Create a test file with the client directly to sanity check reads and lists
-        let mut file = client
-            .create("/testfile", WriteOptions::default())
-            .await
-            .unwrap();
-        let mut buf = BytesMut::new();
-        for i in 0..TEST_FILE_INTS as i32 {
-            buf.put_i32(i);
-        }
-        file.write(buf.freeze()).await.unwrap();
-        file.close().await.unwrap();
-
-        let store = HdfsObjectStore::new(Arc::new(client));
-
-        test_object_store_head(&store).await?;
-        test_object_store_list(&store).await?;
-        test_object_store_rename(&store).await?;
-        test_object_store_read(&store).await?;
-        test_object_store_write(&store).await?;
-        test_object_store_write_multipart(&store).await?;
-        test_object_store_copy(&store).await?;
-
-        Ok(())
-    }
-
-    async fn test_object_store_head(store: &HdfsObjectStore) -> object_store::Result<()> {
-        use object_store::{path::Path, ObjectStore};
-
-        let head = store.head(&Path::from("/testfile")).await?;
-        assert_eq!(head.location, Path::from("/testfile"));
-        assert_eq!(head.size, TEST_FILE_INTS * 4);
-
-        assert!(store.head(&Path::from("/testfile2")).await.is_err());
-
-        Ok(())
-    }
-
-    async fn test_object_store_list(store: &HdfsObjectStore) -> object_store::Result<()> {
-        use futures::StreamExt;
-        use object_store::{path::Path, ObjectMeta, ObjectStore};
-
-        let list: Vec<object_store::Result<ObjectMeta>> =
-            store.list(Some(&Path::from("/"))).collect().await;
-
-        assert_eq!(list.len(), 1);
-        assert_eq!(list[0].as_ref().unwrap().location, Path::from("/testfile"));
-
-        // Listing of a prefix that doesn't exist should return an empty result, not an error
-        assert_eq!(
-            store.list(Some(&Path::from("/doesnt/exist"))).count().await,
-            0
-        );
-
-        let list = store
-            .list_with_delimiter(Some(&Path::from("/doesnt/exist")))
-            .await?;
-
-        assert_eq!(list.common_prefixes.len(), 0);
-        assert_eq!(list.objects.len(), 0);
-
-        Ok(())
-    }
-
-    async fn test_object_store_rename(store: &HdfsObjectStore) -> object_store::Result<()> {
-        use object_store::{path::Path, ObjectStore};
-
-        store
-            .rename(&Path::from("/testfile"), &Path::from("/testfile2"))
-            .await?;
-
-        assert!(store.head(&Path::from("/testfile2")).await.is_ok());
-        assert!(store.head(&Path::from("/testfile")).await.is_err());
-
-        store
-            .rename(&Path::from("/testfile2"), &Path::from("/testfile"))
-            .await?;
-
-        assert!(store.head(&Path::from("/testfile")).await.is_ok());
-        assert!(store.head(&Path::from("/testfile2")).await.is_err());
-
-        Ok(())
-    }
-
-    async fn test_object_store_read(store: &HdfsObjectStore) -> object_store::Result<()> {
-        use object_store::{path::Path, ObjectStore};
-
-        let location = Path::from("/testfile");
-
-        let mut buf = store.get(&location).await?.bytes().await?;
-        for i in 0..TEST_FILE_INTS as i32 {
-            assert_eq!(buf.get_i32(), i);
-        }
-
-        // Read a single integer from the file
-        let offset = TEST_FILE_INTS / 2 * 4;
-        let mut buf = store.get_range(&location, offset..(offset + 4)).await?;
-        assert_eq!(buf.get_i32(), TEST_FILE_INTS as i32 / 2);
-        Ok(())
-    }
-
-    async fn test_object_store_write(store: &HdfsObjectStore) -> object_store::Result<()> {
-        use object_store::{path::Path, ObjectStore};
-
-        store
-            .put(&Path::from("/newfile"), PutPayload::new())
-            .await?;
-        store.head(&Path::from("/newfile")).await?;
-
-        // PutMode = Create should fail for existing file
-        match store
-            .put_opts(
-                &Path::from("/newfile"),
-                PutPayload::new(),
-                PutOptions {
-                    mode: PutMode::Create,
-                    ..Default::default()
-                },
-            )
-            .await
-        {
-            Err(object_store::Error::AlreadyExists { .. }) => (),
-            Err(e) => panic!("Wrong error was thrown for put without overewrite: {:?}", e),
-            Ok(_) => panic!("No error was thrown for put without overwrite for existing file"),
-        }
-
-        // Check a small files, a file that is exactly one block, and a file slightly bigger than a block
-        for size_to_check in [16i32, 128 * 1024 * 1024, 130 * 1024 * 1024] {
-            let ints_to_write = size_to_check / 4;
-
-            let mut data = BytesMut::with_capacity(size_to_check as usize);
-            for i in 0..ints_to_write {
-                data.put_i32(i);
-            }
-
-            let buf = data.freeze();
-            store
-                .put(&Path::from("/newfile"), PutPayload::from_bytes(buf.clone()))
-                .await?;
-
-            assert_eq!(
-                store.head(&Path::from("/newfile")).await?.size,
-                size_to_check as usize
-            );
-
-            let read_data = store.get(&Path::from("/newfile")).await?.bytes().await?;
-
-            assert_eq!(buf.len(), read_data.len());
-
-            for pos in 0..buf.len() {
-                assert_eq!(
-                    buf[pos], read_data[pos],
-                    "data is different as position {} for size {}",
-                    pos, size_to_check
-                );
-            }
-        }
-
-        store.delete(&Path::from("/newfile")).await?;
-
-        Ok(())
-    }
-
-    async fn test_object_store_write_multipart(
-        store: &HdfsObjectStore,
-    ) -> object_store::Result<()> {
-        use object_store::{path::Path, ObjectStore};
-
-        let mut uploader = store.put_multipart(&"/newfile".into()).await?;
-        uploader.complete().await?;
-
-        store.head(&Path::from("/newfile")).await?;
-
-        // Check a small files, a file that is exactly one block, and a file slightly bigger than a block
-        for size_to_check in [16i32, 128 * 1024 * 1024, 130 * 1024 * 1024] {
-            let ints_to_write = size_to_check / 4;
-
-            let mut uploader = store.put_multipart(&"/newfile".into()).await?;
-
-            let mut data = BytesMut::with_capacity(size_to_check as usize);
-            for i in 0..ints_to_write {
-                data.put_i32(i);
-            }
-
-            // Write in 10 MiB chunks
-            let mut buf = data.freeze();
-            while !buf.is_empty() {
-                let to_write = usize::min(buf.len(), 10 * 1024 * 1024);
-                uploader.put_part(buf.split_to(to_write).into()).await?;
-            }
-
-            uploader.complete().await?;
-
-            assert_eq!(
-                store.head(&Path::from("/newfile")).await?.size,
-                size_to_check as usize
-            );
-
-            let mut read_data = store.get(&Path::from("/newfile")).await?.bytes().await?;
-
-            assert_eq!(size_to_check as usize, read_data.len());
-
-            for pos in 0..ints_to_write {
-                assert_eq!(
-                    pos,
-                    read_data.get_i32(),
-                    "data is different at integer {} for size {}",
-                    pos,
-                    size_to_check
-                );
-            }
-        }
-
-        store.delete(&Path::from("/newfile")).await?;
-
-        // Test aborting
-        let mut uploader = store.put_multipart(&"/newfile".into()).await?;
-        assert!(store.head(&"/.newfile.tmp.1".into()).await.is_ok());
-        uploader.abort().await?;
-        assert!(store.head(&"/.newfile.tmp.1".into()).await.is_err());
-        assert!(store.head(&"/newfile".into()).await.is_err());
-
-        // Test multiple uploads to the same destination, default is to overwrite
-        let mut uploader1 = store.put_multipart(&"/newfile".into()).await?;
-        let mut uploader2 = store.put_multipart(&"/newfile".into()).await?;
-        uploader1.put_part(vec![1].into()).await?;
-        uploader2.put_part(vec![2].into()).await?;
-        uploader1.complete().await?;
-        uploader2.complete().await?;
-
-        let result = store.get(&"/newfile".into()).await?;
-        assert!(result.bytes().await?.to_vec() == vec![2]);
-
-        Ok(())
-    }
-
-    async fn test_object_store_copy(store: &HdfsObjectStore) -> object_store::Result<()> {
-        use object_store::{path::Path, ObjectStore};
-
-        store
-            .put(&Path::from("/newfile"), PutPayload::new())
-            .await?;
-
-        let size_to_check = 128 * 1024 * 1024;
-        let ints_to_write = size_to_check / 4;
-        let mut data = BytesMut::with_capacity(size_to_check as usize);
-        for i in 0..ints_to_write {
-            data.put_i32(i);
-        }
-
-        let buf = data.freeze();
-        store
-            .put(&Path::from("/newfile"), buf.clone().into())
-            .await?;
-        store
-            .copy(&Path::from("/newfile"), &Path::from("/newfile2"))
-            .await?;
-
-        let read_data = store.get(&Path::from("/newfile2")).await?.bytes().await?;
-        assert_eq!(buf.len(), read_data.len());
-
-        for pos in 0..buf.len() {
-            assert_eq!(
-                buf[pos], read_data[pos],
-                "data is different as position {} for size {}",
-                pos, size_to_check
-            );
-        }
-
-        assert!(store
-            .copy_if_not_exists(&Path::from("/newfile"), &Path::from("/newfile2"))
-            .await
-            .is_err());
-
-        store.delete(&Path::from("/newfile")).await?;
-        store.delete(&Path::from("/newfile2")).await?;
-
-        Ok(())
-    }
-}
diff --git a/python/tests/conftest.py b/python/tests/conftest.py
index 427d45c..df7aa4f 100644
--- a/python/tests/conftest.py
+++ b/python/tests/conftest.py
@@ -8,7 +8,7 @@ def minidfs():
         [
             "mvn",
             "-f",
-            "../crates/hdfs-native/minidfs",
+            "../rust/minidfs",
             "--quiet",
             "clean",
             "compile",
diff --git a/crates/hdfs-native/src/test.rs b/rust/src/test.rs
similarity index 100%
rename from crates/hdfs-native/src/test.rs
rename to rust/src/test.rs
diff --git a/crates/hdfs-native/tests/common/mod.rs b/rust/tests/common/mod.rs
similarity index 100%
rename from crates/hdfs-native/tests/common/mod.rs
rename to rust/tests/common/mod.rs
diff --git a/crates/hdfs-native/tests/test_ec.rs b/rust/tests/test_ec.rs
similarity index 100%
rename from crates/hdfs-native/tests/test_ec.rs
rename to rust/tests/test_ec.rs
diff --git a/crates/hdfs-native/tests/test_integration.rs b/rust/tests/test_integration.rs
similarity index 100%
rename from crates/hdfs-native/tests/test_integration.rs
rename to rust/tests/test_integration.rs
diff --git a/crates/hdfs-native/tests/test_read.rs b/rust/tests/test_read.rs
similarity index 100%
rename from crates/hdfs-native/tests/test_read.rs
rename to rust/tests/test_read.rs
diff --git a/crates/hdfs-native/tests/test_viewfs.rs b/rust/tests/test_viewfs.rs
similarity index 100%
rename from crates/hdfs-native/tests/test_viewfs.rs
rename to rust/tests/test_viewfs.rs
diff --git a/crates/hdfs-native/tests/test_write.rs b/rust/tests/test_write.rs
similarity index 100%
rename from crates/hdfs-native/tests/test_write.rs
rename to rust/tests/test_write.rs
diff --git a/crates/hdfs-native/tests/test_write_resiliency.rs b/rust/tests/test_write_resiliency.rs
similarity index 100%
rename from crates/hdfs-native/tests/test_write_resiliency.rs
rename to rust/tests/test_write_resiliency.rs