diff --git a/crates/paimon/src/catalog/factory.rs b/crates/paimon/src/catalog/factory.rs index 7e8993c54..c531ccd5d 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 cedc62516..2fe7bb78e 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 2144ed754..4a62c73b5 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -79,12 +79,10 @@ impl FileIO { /// Reference: pub fn new_input(&self, path: &str) -> crate::Result { 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: pub fn new_output(&self, path: &str) -> Result { 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: pub async fn get_status(&self, path: &str) -> Result { 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> { 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> { 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 { 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 { - 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 { - 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 { - 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 { - 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 { - 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 { - 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 59d2740eb..8ab67e40d 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> { + // 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 bebd09e78..80463e413 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) -> Vec