This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 06d6818 fix(io): preserve Windows drive letter in local fs paths
(#398)
06d6818 is described below
commit 06d681874eef78a147ea1a8782fd81319cc05e91
Author: chaoyang <[email protected]>
AuthorDate: Sat Jun 20 15:37:24 2026 +0800
fix(io): preserve Windows drive letter in local fs paths (#398)
---
crates/paimon/src/catalog/factory.rs | 5 +-
crates/paimon/src/catalog/filesystem.rs | 6 +-
crates/paimon/src/io/file_io.rs | 95 +++++++++++++++++---------------
crates/paimon/src/io/storage.rs | 97 ++++++++++++++++++++++++++++-----
crates/paimon/src/spec/manifest.rs | 1 -
5 files changed, 142 insertions(+), 62 deletions(-)
diff --git a/crates/paimon/src/catalog/factory.rs
b/crates/paimon/src/catalog/factory.rs
index 7e8993c..c531ccd 100644
--- a/crates/paimon/src/catalog/factory.rs
+++ b/crates/paimon/src/catalog/factory.rs
@@ -114,7 +114,10 @@ impl CatalogFactory {
}
#[cfg(test)]
-#[cfg(not(windows))] // Skip on Windows due to path compatibility issues
+// Skip on Windows: these tests use a hardcoded POSIX warehouse
+// (`/tmp/test-warehouse`), which is not a valid absolute Windows path, so
+// `FileIO::from_path` cannot derive a `file://` URL from it. See #397.
+#[cfg(not(windows))]
mod tests {
use super::*;
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index cedc625..2fe7bb7 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -463,7 +463,11 @@ fn fill_table_name(err: Error, identifier: &Identifier) ->
Error {
}
#[cfg(test)]
-#[cfg(not(windows))] // Skip on Windows due to path compatibility issues
+// Skip on Windows: these tests list directories, and opendal's `fs` lister
+// panics (`StripPrefixError`) when listing under a drive-rooted path with the
+// `root="/"` operator setup. Single-file ops work after the drive-letter fix;
+// directory listing needs the opendal root rework tracked in #397.
+#[cfg(not(windows))]
mod tests {
use super::*;
use tempfile::TempDir;
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 2144ed7..4a62c73 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -79,12 +79,10 @@ impl FileIO {
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L76>
pub fn new_input(&self, path: &str) -> crate::Result<InputFile> {
let (op, relative_path) = self.storage.create(path)?;
- let path = path.to_string();
- let relative_path_pos = path.len() - relative_path.len();
Ok(InputFile {
op,
- path,
- relative_path_pos,
+ path: path.to_string(),
+ relative_path: relative_path.into_owned(),
})
}
@@ -93,12 +91,10 @@ impl FileIO {
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L87>
pub fn new_output(&self, path: &str) -> Result<OutputFile> {
let (op, relative_path) = self.storage.create(path)?;
- let path = path.to_string();
- let relative_path_pos = path.len() - relative_path.len();
Ok(OutputFile {
op,
- path,
- relative_path_pos,
+ path: path.to_string(),
+ relative_path: relative_path.into_owned(),
})
}
@@ -107,9 +103,12 @@ impl FileIO {
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L97>
pub async fn get_status(&self, path: &str) -> Result<FileStatus> {
let (op, relative_path) = self.storage.create(path)?;
- let meta = op.stat(relative_path).await.context(IoUnexpectedSnafu {
- message: format!("Failed to get file status for '{path}'"),
- })?;
+ let meta = op
+ .stat(relative_path.as_ref())
+ .await
+ .context(IoUnexpectedSnafu {
+ message: format!("Failed to get file status for '{path}'"),
+ })?;
Ok(FileStatus {
size: meta.content_length(),
@@ -128,10 +127,15 @@ impl FileIO {
/// FIXME: how to handle large dir? Better to return a stream instead?
pub async fn list_status(&self, path: &str) -> Result<Vec<FileStatus>> {
let (op, relative_path) = self.storage.create(path)?;
+ // `relative_path` is a byte-suffix of `path` for object stores and
POSIX
+ // local paths, so this recovers the scheme/root prefix. For a Windows
+ // local path the relative form only swaps `\`->`/`
(length-preserving),
+ // so this is `""` and entries are reported in opendal's normalized
+ // `/C:/...` form — which still round-trips back through `create`.
let base_path = &path[..path.len() - relative_path.len()];
// Opendal list() expects directory path to end with `/`.
// use normalize_root to make sure it end with `/`.
- let list_path = normalize_root(relative_path);
+ let list_path = normalize_root(relative_path.as_ref());
let entries = op.list_with(&list_path).await.context(IoUnexpectedSnafu
{
message: format!("Failed to list files in '{path}'"),
@@ -161,8 +165,10 @@ impl FileIO {
/// List all files recursively under the given directory path.
pub async fn list_status_recursive(&self, path: &str) ->
Result<Vec<FileStatus>> {
let (op, relative_path) = self.storage.create(path)?;
+ // See `list_status`: `relative_path` is a byte-suffix of `path` except
+ // for Windows local paths, where it only swaps separators (same
length).
let base_path = &path[..path.len() - relative_path.len()];
- let list_path = normalize_root(relative_path);
+ let list_path = normalize_root(relative_path.as_ref());
let entries =
op.list_with(&list_path)
@@ -202,9 +208,11 @@ impl FileIO {
pub async fn exists(&self, path: &str) -> Result<bool> {
let (op, relative_path) = self.storage.create(path)?;
- op.exists(relative_path).await.context(IoUnexpectedSnafu {
- message: format!("Failed to check existence of '{path}'"),
- })
+ op.exists(relative_path.as_ref())
+ .await
+ .context(IoUnexpectedSnafu {
+ message: format!("Failed to check existence of '{path}'"),
+ })
}
/// Delete a file.
@@ -213,9 +221,11 @@ impl FileIO {
pub async fn delete_file(&self, path: &str) -> Result<()> {
let (op, relative_path) = self.storage.create(path)?;
- op.delete(relative_path).await.context(IoUnexpectedSnafu {
- message: format!("Failed to delete file '{path}'"),
- })?;
+ op.delete(relative_path.as_ref())
+ .await
+ .context(IoUnexpectedSnafu {
+ message: format!("Failed to delete file '{path}'"),
+ })?;
Ok(())
}
@@ -226,7 +236,7 @@ impl FileIO {
pub async fn delete_dir(&self, path: &str) -> Result<()> {
let (op, relative_path) = self.storage.create(path)?;
- op.remove_all(relative_path)
+ op.remove_all(relative_path.as_ref())
.await
.context(IoUnexpectedSnafu {
message: format!("Failed to delete directory '{path}'"),
@@ -243,7 +253,7 @@ impl FileIO {
pub async fn mkdirs(&self, path: &str) -> Result<()> {
let (op, relative_path) = self.storage.create(path)?;
// Opendal create_dir expects the path to end with `/` to indicate a
directory.
- let dir_path = normalize_root(relative_path);
+ let dir_path = normalize_root(relative_path.as_ref());
op.create_dir(&dir_path).await.context(IoUnexpectedSnafu {
message: format!("Failed to create directory '{path}'"),
})?;
@@ -270,7 +280,7 @@ impl FileIO {
let (_, relative_path_dst) = self.storage.create(dst)?;
op_src
- .rename(relative_path_src, relative_path_dst)
+ .rename(relative_path_src.as_ref(), relative_path_dst.as_ref())
.await
.context(IoUnexpectedSnafu {
message: format!("Failed to rename '{src}' to '{dst}'"),
@@ -288,7 +298,8 @@ fn status_path(base_path: &str, entry_path: &str) -> String
{
}
}
-fn looks_like_windows_drive_path(path: &str) -> bool {
+/// Whether `path` begins with a Windows drive specifier such as `C:\` or
`C:/`.
+pub(crate) fn looks_like_windows_drive_path(path: &str) -> bool {
let bytes = path.as_bytes();
bytes.len() >= 3
&& bytes[0].is_ascii_alphabetic()
@@ -384,7 +395,9 @@ pub struct FileStatus {
pub struct InputFile {
op: Operator,
path: String,
- relative_path_pos: usize,
+ /// The opendal-relative path (see [`FileIO::new_input`]); not necessarily
a
+ /// suffix of `path`, since local paths are separator-normalized.
+ relative_path: String,
}
impl InputFile {
@@ -393,11 +406,11 @@ impl InputFile {
}
pub async fn exists(&self) -> crate::Result<bool> {
- Ok(self.op.exists(&self.path[self.relative_path_pos..]).await?)
+ Ok(self.op.exists(&self.relative_path).await?)
}
pub async fn metadata(&self) -> crate::Result<FileStatus> {
- let meta = self.op.stat(&self.path[self.relative_path_pos..]).await?;
+ let meta = self.op.stat(&self.relative_path).await?;
Ok(FileStatus {
size: meta.content_length(),
@@ -410,15 +423,11 @@ impl InputFile {
}
pub async fn read(&self) -> crate::Result<Bytes> {
- Ok(self
- .op
- .read(&self.path[self.relative_path_pos..])
- .await?
- .to_bytes())
+ Ok(self.op.read(&self.relative_path).await?.to_bytes())
}
pub async fn reader(&self) -> crate::Result<impl FileRead> {
- Ok(self.op.reader(&self.path[self.relative_path_pos..]).await?)
+ Ok(self.op.reader(&self.relative_path).await?)
}
}
@@ -426,7 +435,9 @@ impl InputFile {
pub struct OutputFile {
op: Operator,
path: String,
- relative_path_pos: usize,
+ /// The opendal-relative path (see [`FileIO::new_output`]); not
necessarily a
+ /// suffix of `path`, since local paths are separator-normalized.
+ relative_path: String,
}
impl OutputFile {
@@ -435,14 +446,14 @@ impl OutputFile {
}
pub async fn exists(&self) -> crate::Result<bool> {
- Ok(self.op.exists(&self.path[self.relative_path_pos..]).await?)
+ Ok(self.op.exists(&self.relative_path).await?)
}
pub fn to_input_file(self) -> InputFile {
InputFile {
op: self.op,
path: self.path,
- relative_path_pos: self.relative_path_pos,
+ relative_path: self.relative_path,
}
}
@@ -467,7 +478,7 @@ impl OutputFile {
}
async fn opendal_writer(&self) -> crate::Result<opendal::Writer> {
- Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
+ Ok(self.op.writer(&self.relative_path).await?)
}
}
@@ -731,20 +742,14 @@ mod object_storage_path_test {
fn assert_relative_paths(file_io: &FileIO, path: &str,
expected_relative_path: &str) {
let input = file_io.new_input(path).unwrap();
assert_eq!(input.location(), path);
- assert_eq!(
- &input.path[input.relative_path_pos..],
- expected_relative_path
- );
+ assert_eq!(input.relative_path, expected_relative_path);
let output = file_io.new_output(path).unwrap();
assert_eq!(output.location(), path);
- assert_eq!(
- &output.path[output.relative_path_pos..],
- expected_relative_path
- );
+ assert_eq!(output.relative_path, expected_relative_path);
let (_op, relative_path) = file_io.storage.create(path).unwrap();
- assert_eq!(relative_path, expected_relative_path);
+ assert_eq!(relative_path.as_ref(), expected_relative_path);
let base_path = &path[..path.len() - relative_path.len()];
assert_eq!(format!("{base_path}{relative_path}"), path);
diff --git a/crates/paimon/src/io/storage.rs b/crates/paimon/src/io/storage.rs
index 59d2740..8ab67e4 100644
--- a/crates/paimon/src/io/storage.rs
+++ b/crates/paimon/src/io/storage.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::borrow::Cow;
use std::collections::HashMap;
#[cfg(any(
feature = "storage-azdls",
@@ -184,10 +185,12 @@ impl Storage {
}
}
- pub(crate) fn create<'a>(&self, path: &'a str) -> crate::Result<(Operator,
&'a str)> {
+ pub(crate) fn create<'a>(&self, path: &'a str) -> crate::Result<(Operator,
Cow<'a, str>)> {
match self {
#[cfg(feature = "storage-memory")]
- Storage::Memory { op } => Ok((op.clone(),
Self::memory_relative_path(path)?)),
+ Storage::Memory { op } => {
+ Ok((op.clone(),
Cow::Borrowed(Self::memory_relative_path(path)?)))
+ }
#[cfg(feature = "storage-fs")]
Storage::LocalFs { op } => Ok((op.clone(),
Self::fs_relative_path(path)?)),
#[cfg(feature = "storage-oss")]
@@ -195,14 +198,14 @@ impl Storage {
let (bucket, relative_path) =
Self::bucket_and_relative_path(path, "OSS", &["oss"])?;
let op = Self::cached_oss_operator(config, operators, path,
&bucket)?;
- Ok((op, relative_path))
+ Ok((op, Cow::Borrowed(relative_path)))
}
#[cfg(feature = "storage-s3")]
Storage::S3 { config, operators } => {
let (bucket, relative_path) =
Self::bucket_and_relative_path(path, "S3", &["s3",
"s3a"])?;
let op = Self::cached_s3_operator(config, operators, path,
&bucket)?;
- Ok((op, relative_path))
+ Ok((op, Cow::Borrowed(relative_path)))
}
#[cfg(feature = "storage-cos")]
Storage::Cos { config, operators } => {
@@ -211,7 +214,7 @@ impl Storage {
let op = Self::cached_operator(operators, "COS", &bucket, || {
super::cos_config_build(config, path)
})?;
- Ok((op, relative_path))
+ Ok((op, Cow::Borrowed(relative_path)))
}
#[cfg(feature = "storage-azdls")]
Storage::Azdls { config, operators } => {
@@ -220,7 +223,7 @@ impl Storage {
let op = Self::cached_operator(operators, "Azure", &cache_key,
|| {
super::azdls_config_build(config, path)
})?;
- Ok((op, relative_path))
+ Ok((op, Cow::Borrowed(relative_path)))
}
#[cfg(feature = "storage-obs")]
Storage::Obs { config, operators } => {
@@ -229,7 +232,7 @@ impl Storage {
let op = Self::cached_operator(operators, "OBS", &bucket, || {
super::obs_config_build(config, path)
})?;
- Ok((op, relative_path))
+ Ok((op, Cow::Borrowed(relative_path)))
}
#[cfg(feature = "storage-gcs")]
Storage::Gcs { config, operators } => {
@@ -238,7 +241,7 @@ impl Storage {
let op = Self::cached_operator(operators, "GCS", &bucket, || {
super::gcs_config_build(config, path)
})?;
- Ok((op, relative_path))
+ Ok((op, Cow::Borrowed(relative_path)))
}
#[cfg(feature = "storage-hdfs")]
Storage::Hdfs { config, op } => {
@@ -254,7 +257,10 @@ impl Storage {
if guard.is_none() {
*guard = Some(super::hdfs_config_build(config, path)?);
}
- Ok((guard.as_ref().unwrap().clone(), relative_path))
+ Ok((
+ guard.as_ref().unwrap().clone(),
+ Cow::Borrowed(relative_path),
+ ))
}
}
}
@@ -270,15 +276,38 @@ impl Storage {
}
}
+ /// Turn an absolute local path into the relative path that opendal's `fs`
+ /// service joins onto its `/` root.
+ ///
+ /// On POSIX an absolute path `/tmp/wh` becomes `tmp/wh`: dropping the
single
+ /// leading separator lets opendal rebuild `/tmp/wh` from its `/` root.
+ ///
+ /// A bare drop-the-first-char would corrupt a Windows path such as
+ /// `C:\dir` into `:\dir` (the drive letter is lost — the historical source
+ /// of the "invalid filename" failures on Windows). Instead we keep the
+ /// drive specifier and only normalize separators to `/`, mirroring how
Java
+ /// Paimon's `Path` (modeled on Hadoop's) handles Windows paths. opendal
then
+ /// does `PathBuf::from("/").join("C:/dir")`, and because the argument
+ /// carries a drive prefix `Path::join` replaces the base, yielding the
real
+ /// `C:\dir` on Windows.
#[cfg(feature = "storage-fs")]
- fn fs_relative_path(path: &str) -> crate::Result<&str> {
+ fn fs_relative_path(path: &str) -> crate::Result<Cow<'_, str>> {
+ // A `file://` / `file:/` URL is already in scheme-relative form.
if let Some(stripped) = path.strip_prefix("file:/") {
- Ok(stripped)
- } else {
- path.get(1..).ok_or_else(|| error::Error::ConfigInvalid {
+ return Ok(if stripped.contains('\\') {
+ Cow::Owned(stripped.replace('\\', "/"))
+ } else {
+ Cow::Borrowed(stripped)
+ });
+ }
+ if super::looks_like_windows_drive_path(path) {
+ return Ok(Cow::Owned(path.replace('\\', "/")));
+ }
+ path.get(1..)
+ .map(Cow::Borrowed)
+ .ok_or_else(|| error::Error::ConfigInvalid {
message: format!("Invalid file path: {path}"),
})
- }
}
#[cfg(any(
@@ -398,3 +427,43 @@ impl Storage {
}
}
}
+
+#[cfg(all(test, feature = "storage-fs"))]
+mod fs_relative_path_tests {
+ use super::Storage;
+
+ fn rel(path: &str) -> String {
+ Storage::fs_relative_path(path).unwrap().into_owned()
+ }
+
+ #[test]
+ fn posix_absolute_path_drops_leading_separator() {
+ // opendal joins the result onto its `/` root, rebuilding `/tmp/wh`.
+ assert_eq!(rel("/tmp/wh"), "tmp/wh");
+ assert_eq!(rel("/tmp/wh/db.db/t"), "tmp/wh/db.db/t");
+ }
+
+ #[test]
+ fn file_scheme_is_stripped() {
+ assert_eq!(rel("file:/tmp/wh"), "tmp/wh");
+ // `file://` keeps the leading authority slash, matching prior
behavior.
+ assert_eq!(rel("file:///tmp/wh"), "//tmp/wh");
+ }
+
+ #[test]
+ fn windows_drive_path_keeps_drive_and_normalizes_separators() {
+ // The historical bug dropped the drive letter (`C:\wh` -> `:\wh`); we
+ // must keep it and only switch `\` to `/` so opendal's
+ // `PathBuf::from("/").join(..)` rebuilds the real `C:\wh` on Windows.
+ assert_eq!(rel(r"C:\Users\wh"), "C:/Users/wh");
+ assert_eq!(rel("C:/Users/wh"), "C:/Users/wh");
+ assert_eq!(rel(r"D:\a\b\c"), "D:/a/b/c");
+ }
+
+ #[test]
+ fn windows_mixed_separators_are_normalized() {
+ // make_path concatenates with `/`, so a Windows warehouse yields a
+ // mixed-separator path that must still normalize cleanly.
+ assert_eq!(rel(r"C:\Users\wh/db.db/t"), "C:/Users/wh/db.db/t");
+ }
+}
diff --git a/crates/paimon/src/spec/manifest.rs
b/crates/paimon/src/spec/manifest.rs
index bebd09e..80463e4 100644
--- a/crates/paimon/src/spec/manifest.rs
+++ b/crates/paimon/src/spec/manifest.rs
@@ -90,7 +90,6 @@ pub(crate) fn merge_active_entries(entries:
Vec<ManifestEntry>) -> Vec<ManifestE
}
#[cfg(test)]
-#[cfg(not(windows))] // Skip on Windows due to path compatibility issues
mod tests {
use super::*;
use crate::io::FileIO;
