update

Author: xushiyan

crates/core/src/storage/mod.rs

@@ -283,10 +283,12 @@ pub async fn get_leaf_dirs(storage: &Storage, subdir: Option<&str>) -> Result<Ve
                 next_subdir.push(curr);
             }
             next_subdir.push(child_dir);
-            let next_subdir = next_subdir
+            let next_subdir_str = next_subdir
                 .to_str()
                 .ok_or_else(|| InvalidPath(format!("Failed to convert path: {:?}", next_subdir)))?;
-            let curr_leaf_dir = get_leaf_dirs(storage, Some(next_subdir)).await?;
+            // Normalize to forward slashes for consistency across platforms
+            let next_subdir_normalized = next_subdir_str.replace('\\', "/");
+            let curr_leaf_dir = get_leaf_dirs(storage, Some(&next_subdir_normalized)).await?;
             leaf_dirs.extend(curr_leaf_dir);
         }
     }

crates/core/src/storage/util.rs

@@ -25,6 +25,14 @@ use url::Url;
 use crate::storage::error::StorageError::{InvalidPath, UrlParseError};
 use crate::storage::Result;
 
+/// Normalizes path separators to forward slashes for cross-platform consistency.
+///
+/// Windows paths use backslashes, but for consistency with URLs and storage operations,
+/// we normalize all paths to use forward slashes.
+fn normalize_path_separators(path: &str) -> String {
+    path.replace('\\', "/")
+}
+
 /// Parses a URI string into a URL.
 pub fn parse_uri(uri: &str) -> Result<Url> {
     // Check for Windows drive paths first to prevent them from being parsed as URLs
@@ -33,10 +41,10 @@ pub fn parse_uri(uri: &str) -> Result<Url> {
         Url::from_directory_path(uri)
             .map_err(|_| UrlParseError(url::ParseError::EmptyHost))?
     } else {
-        match Url::parse(uri) {
-            Ok(url) => url,
-            Err(e) => Url::from_directory_path(uri).map_err(|_| UrlParseError(e))?,
-        }
+        // Try parsing as URL first, fall back to directory path for local paths
+        Url::parse(uri).or_else(|e| {
+            Url::from_directory_path(uri).map_err(|_| UrlParseError(e))
+        })?
     };
 
     if url.path().ends_with('/') {
@@ -52,6 +60,30 @@ pub fn get_scheme_authority(url: &Url) -> String {
     format!("{}://{}", url.scheme(), url.authority())
 }
 
+/// Converts a URL to a path string suitable for file system operations.
+///
+/// For file:// URLs, this returns the decoded native file path to avoid
+/// percent-encoding issues (e.g., ~ being encoded as %7E on Windows).
+/// For other schemes (s3://, gs://, etc.), returns the URL path as-is.
+///
+/// # Arguments
+/// * `url` - The URL to convert
+///
+/// # Returns
+/// A path string suitable for file operations
+pub fn url_to_path_string(url: &Url) -> Result<String> {
+    if url.scheme() == "file" {
+        // Use to_file_path() to properly decode file URLs and avoid percent-encoding issues
+        // This is critical on Windows where ~ gets encoded as %7E in URL paths
+        url.to_file_path()
+            .map_err(|_| InvalidPath(format!("Failed to convert file URL to path: {}", url)))
+            .map(|p| p.to_string_lossy().to_string())
+    } else {
+        // For cloud storage URLs (s3://, gs://, etc.), use the URL path directly
+        Ok(url.path().to_string())
+    }
+}
+
 /// Joins a base URL with a list of segments.
 ///
 /// # Arguments
@@ -113,7 +145,7 @@ pub fn join_base_path_with_relative_path(base: &str, relative: &str) -> Result<S
 /// * `segments` - Path segments to append
 ///
 /// # Returns
-/// The joined path as a string, or an error if the input is invalid
+/// The joined path as a string with forward slashes, or an error if the input is invalid
 pub fn join_base_path_with_segments(base: &str, segments: &[&str]) -> Result<String> {
     // 1) Windows absolute path check FIRST (before URL parsing)
     //    This prevents "C:/foo" from being parsed as a URL with scheme "C"
@@ -127,7 +159,8 @@ pub fn join_base_path_with_segments(base: &str, segments: &[&str]) -> Result<Str
                 }
             }
         }
-        return Ok(path.to_string_lossy().into_owned());
+        let path_str = path.to_string_lossy();
+        return Ok(normalize_path_separators(&path_str));
     }
 
     // 2) Try URL mode: parse base as absolute URL
@@ -164,7 +197,8 @@ pub fn join_base_path_with_segments(base: &str, segments: &[&str]) -> Result<Str
             }
         }
     }
-    Ok(path.to_string_lossy().into_owned())
+    let path_str = path.to_string_lossy();
+    Ok(normalize_path_separators(&path_str))
 }
 
 /// Detects if a string starts like a Windows drive path: `C:\` or `C:/`
@@ -224,15 +258,41 @@ mod tests {
         assert!(url.path().contains("test"));
         assert!(url.path().contains("data"));
 
-        // Windows path with drive letter only
-        let url = parse_uri("D:\\").unwrap();
+        // Windows path with drive letter and trailing backslash
+        let url = parse_uri(r"D:\").unwrap();
         assert_eq!(url.scheme(), "file");
 
-        // Windows path with drive letter and colon
-        let url = parse_uri("E:").unwrap();
+        // Windows path with drive letter and trailing forward slash
+        let url = parse_uri("D:/").unwrap();
         assert_eq!(url.scheme(), "file");
     }
 
+    #[test]
+    fn test_url_to_path_string_with_file_url() {
+        // Test that url_to_path_string properly decodes file:// URLs
+        // This is critical on Windows where ~ gets encoded as %7E in URL paths
+        let path_with_tilde = "/tmp/RUNNER~1/test.parquet";
+        let url = Url::from_directory_path(path_with_tilde).unwrap();
+
+        let path_str = url_to_path_string(&url).unwrap();
+
+        // The path should contain the original tilde (not %7E)
+        assert!(path_str.contains("RUNNER~1"));
+        assert!(!path_str.contains("%7E"));
+    }
+
+    #[test]
+    fn test_url_to_path_string_with_cloud_url() {
+        // Test that url_to_path_string returns URL path for cloud storage
+        let s3_url = Url::from_str("s3://bucket/path/to/file.parquet").unwrap();
+        let path_str = url_to_path_string(&s3_url).unwrap();
+        assert_eq!(path_str, "/path/to/file.parquet");
+
+        let gs_url = Url::from_str("gs://bucket/data/file.parquet").unwrap();
+        let path_str = url_to_path_string(&gs_url).unwrap();
+        assert_eq!(path_str, "/data/file.parquet");
+    }
+
     #[test]
     fn join_base_url_with_segments() {
         let base_url = Url::from_str("file:///base").unwrap();
@@ -316,13 +376,15 @@ mod tests {
 
     #[test]
     fn test_join_base_path_with_segments_windows() {
-        // Windows path with backslash
+        // Windows path with backslash - should normalize to forward slashes
         let result =
             join_base_path_with_segments(r"C:\Users\test", &["Documents", "file.txt"]).unwrap();
         assert!(result.contains("Users"));
         assert!(result.contains("test"));
         assert!(result.contains("Documents"));
         assert!(result.contains("file.txt"));
+        // Verify normalized to forward slashes
+        assert!(result.contains("Users/test/Documents/file.txt") || result.contains("C:/Users"));
 
         // Windows path with forward slash
         let result =
@@ -331,20 +393,26 @@ mod tests {
         assert!(result.contains("test"));
         assert!(result.contains("Documents"));
         assert!(result.contains("file.txt"));
+        // Verify normalized to forward slashes
+        assert!(result.contains("Users/test/Documents/file.txt") || result.contains("C:/Users"));
 
-        // Windows path with mixed separators in segments
+        // Windows path with mixed separators in segments - should normalize
         let result =
             join_base_path_with_segments(r"D:\data", &["dir/subdir", r"more\files"]).unwrap();
         assert!(result.contains("data"));
         assert!(result.contains("dir"));
         assert!(result.contains("subdir"));
         assert!(result.contains("more"));
         assert!(result.contains("files"));
+        // Verify normalized to forward slashes
+        assert!(!result.contains('\\'));
 
         // Just drive letter
         let result = join_base_path_with_segments("C:", &["temp", "file.txt"]).unwrap();
         assert!(result.contains("temp"));
         assert!(result.contains("file.txt"));
+        // Verify normalized to forward slashes
+        assert!(!result.contains('\\'));
     }
 
     #[test]

crates/datafusion/src/lib.rs

@@ -46,7 +46,7 @@ use datafusion_physical_expr::create_physical_expr;
 use crate::util::expr::exprs_to_filters;
 use hudi_core::config::read::HudiReadConfig::InputPartitions;
 use hudi_core::config::util::empty_options;
-use hudi_core::storage::util::{get_scheme_authority, join_url_segments};
+use hudi_core::storage::util::{get_scheme_authority, join_url_segments, url_to_path_string};
 use hudi_core::table::Table as HudiTable;
 
 /// Create a `HudiDataSource`.
@@ -195,7 +195,9 @@ impl TableProvider for HudiDataSource {
                 let url = join_url_segments(&base_url, &[relative_path.as_str()])
                     .map_err(|e| Execution(format!("Failed to join URL segments: {e:?}")))?;
                 let size = f.base_file.file_metadata.as_ref().map_or(0, |m| m.size);
-                let partitioned_file = PartitionedFile::new(url.path(), size as u64);
+                let path_str = url_to_path_string(&url)
+                    .map_err(|e| Execution(format!("Failed to convert URL to path: {e:?}")))?;
+                let partitioned_file = PartitionedFile::new(path_str, size as u64);
                 parquet_file_group_vec.push(partitioned_file);
             }
             parquet_file_groups.push(parquet_file_group_vec)