Fix 5592: Colon (:) in in object_store::path::{Path} is not handled on Windows (#5830)

* Fix issue #5800: Handle missing files in list_with_delimiter

* draft

* cargo fmt

* Handle leading colon

* Add windows CI

* Fix CI job

* Only run local tests and set target family for failing tests

* Run all tests without my changes and removed target os

* Restore changes again

* Add back newline (removed by mistake)

* Fix test after merge with master

Author: hesampakdaman

.github/workflows/object_store.yml

@@ -198,3 +198,16 @@ jobs:
         run: cargo build --target wasm32-unknown-unknown
       - name: Build wasm32-wasi
         run: cargo build --target wasm32-wasi
+
+  windows:
+    name: cargo test LocalFileSystem (win64)
+    runs-on: windows-latest
+    defaults:
+      run:
+        working-directory: object_store
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: true
+      - name: Run LocalFileSystem tests
+        run: cargo test local::tests

object_store/src/local.rs

@@ -297,7 +297,22 @@ impl LocalFileSystem {
                 path: location.as_ref()
             }
         );
-        self.config.prefix_to_filesystem(location)
+        let path = self.config.prefix_to_filesystem(location)?;
+
+        #[cfg(target_os = "windows")]
+        let path = {
+            let path = path.to_string_lossy();
+
+            // Assume the first char is the drive letter and the next is a colon.
+            let mut out = String::new();
+            let drive = &path[..2]; // The drive letter and colon (e.g., "C:")
+            let filepath = &path[2..].replace(':', "%3A"); // Replace subsequent colons
+            out.push_str(drive);
+            out.push_str(filepath);
+            PathBuf::from(out)
+        };
+
+        Ok(path)
     }
 
     /// Enable automatic cleanup of empty directories when deleting files
@@ -1053,6 +1068,7 @@ mod tests {
     use super::*;
 
     #[tokio::test]
+    #[cfg(target_family = "unix")]
     async fn file_test() {
         let root = TempDir::new().unwrap();
         let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
@@ -1069,6 +1085,7 @@ mod tests {
     }
 
     #[test]
+    #[cfg(target_family = "unix")]
     fn test_non_tokio() {
         let root = TempDir::new().unwrap();
         let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
@@ -1481,6 +1498,28 @@ mod tests {
         assert_eq!(list, vec![c, a]);
     }
 
+    #[tokio::test]
+    #[cfg(target_os = "windows")]
+    async fn filesystem_filename_with_colon() {
+        let root = TempDir::new().unwrap();
+        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
+        let path = Path::parse("file%3Aname.parquet").unwrap();
+        let location = Path::parse("file:name.parquet").unwrap();
+
+        integration.put(&location, "test".into()).await.unwrap();
+        let list = flatten_list_stream(&integration, None).await.unwrap();
+        assert_eq!(list, vec![path.clone()]);
+
+        let result = integration
+            .get(&location)
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        assert_eq!(result, Bytes::from("test"));
+    }
+
     #[tokio::test]
     async fn delete_dirs_automatically() {
         let root = TempDir::new().unwrap();