IO: Fix Windows drive letters misidentified as URI schemes

pyiceberg/io/__init__.py

@@ -27,6 +27,7 @@
 
 import importlib
 import logging
+import os
 import warnings
 from abc import ABC, abstractmethod
 from io import SEEK_SET
@@ -41,6 +42,18 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _is_local_path(path: str) -> bool:
+    r"""Check if a path is a local filesystem path rather than a URI.
+
+    Uses os.path.splitdrive to detect Windows drive letters (e.g. 'C:\\...')
+    in a platform-aware way. On non-Windows systems, splitdrive always returns
+    an empty drive component, so this only triggers on Windows.
+    """
+    drive, _ = os.path.splitdrive(path)
+    return drive != ""
+
+
 AWS_PROFILE_NAME = "client.profile-name"
 AWS_REGION = "client.region"
 AWS_ACCESS_KEY_ID = "client.access-key-id"
@@ -335,14 +348,19 @@ def _import_file_io(io_impl: str, properties: Properties) -> FileIO | None:
 
 
 def _infer_file_io_from_scheme(path: str, properties: Properties) -> FileIO | None:
-    parsed_url = urlparse(path)
-    if parsed_url.scheme:
-        if file_ios := SCHEMA_TO_FILE_IO.get(parsed_url.scheme):
+    if _is_local_path(path):
+        scheme = "file"
+    else:
+        parsed_url = urlparse(path)
+        scheme = parsed_url.scheme
+
+    if scheme:
+        if file_ios := SCHEMA_TO_FILE_IO.get(scheme):
             for file_io_path in file_ios:
                 if file_io := _import_file_io(file_io_path, properties):
                     return file_io
         else:
-            warnings.warn(f"No preferred file implementation for scheme: {parsed_url.scheme}", stacklevel=2)
+            warnings.warn(f"No preferred file implementation for scheme: {scheme}", stacklevel=2)
     return None
 
 

pyiceberg/io/fsspec.py

@@ -89,6 +89,7 @@
     InputStream,
     OutputFile,
     OutputStream,
+    _is_local_path,
 )
 from pyiceberg.typedef import Properties
 from pyiceberg.types import strtobool
@@ -443,7 +444,7 @@ def new_input(self, location: str) -> FsspecInputFile:
             FsspecInputFile: An FsspecInputFile instance for the given location.
         """
         uri = urlparse(location)
-        fs = self._get_fs_from_uri(uri)
+        fs = self._get_fs_from_uri(uri, location)
         return FsspecInputFile(location=location, fs=fs)
 
     @override
@@ -457,7 +458,7 @@ def new_output(self, location: str) -> FsspecOutputFile:
             FsspecOutputFile: An FsspecOutputFile instance for the given location.
         """
         uri = urlparse(location)
-        fs = self._get_fs_from_uri(uri)
+        fs = self._get_fs_from_uri(uri, location)
         return FsspecOutputFile(location=location, fs=fs)
 
     @override
@@ -475,11 +476,13 @@ def delete(self, location: str | InputFile | OutputFile) -> None:
             str_location = location
 
         uri = urlparse(str_location)
-        fs = self._get_fs_from_uri(uri)
+        fs = self._get_fs_from_uri(uri, str_location)
         fs.rm(str_location)
 
-    def _get_fs_from_uri(self, uri: "ParseResult") -> AbstractFileSystem:
+    def _get_fs_from_uri(self, uri: "ParseResult", location: str = "") -> AbstractFileSystem:
         """Get a filesystem from a parsed URI, using hostname for ADLS account resolution."""
+        if _is_local_path(location):
+            return self.get_fs("file")
         if uri.scheme in _ADLS_SCHEMES:
             return self.get_fs(uri.scheme, uri.hostname)
         return self.get_fs(uri.scheme)

pyiceberg/io/pyarrow.py

@@ -121,6 +121,7 @@
     InputStream,
     OutputFile,
     OutputStream,
+    _is_local_path,
 )
 from pyiceberg.io.fileformat import DataFileStatistics as DataFileStatistics
 from pyiceberg.io.fileformat import FileFormatFactory, FileFormatModel, FileFormatWriter
@@ -401,10 +402,17 @@ def __init__(self, properties: Properties = EMPTY_DICT):
 
     @staticmethod
     def parse_location(location: str, properties: Properties = EMPTY_DICT) -> tuple[str, str, str]:
-        """Return (scheme, netloc, path) for the given location.
+        r"""Return (scheme, netloc, path) for the given location.
 
         Uses DEFAULT_SCHEME and DEFAULT_NETLOC if scheme/netloc are missing.
+        On Windows, paths with drive letters (e.g. 'C:\\...') are treated as
+        local file paths rather than URIs.
         """
+        if _is_local_path(location):
+            default_scheme = properties.get("DEFAULT_SCHEME", "file")
+            default_netloc = properties.get("DEFAULT_NETLOC", "")
+            return default_scheme, default_netloc, os.path.abspath(location)
+
         uri = urlparse(location)
 
         if not uri.scheme:

tests/io/test_io.py

@@ -17,6 +17,7 @@
 
 import os
 import pickle
+import sys
 import tempfile
 from typing import Any
 
@@ -27,6 +28,7 @@
     PY_IO_IMPL,
     _import_file_io,
     _infer_file_io_from_scheme,
+    _is_local_path,
     load_file_io,
 )
 from pyiceberg.io.pyarrow import PyArrowFileIO
@@ -339,3 +341,23 @@ def test_infer_file_io_from_schema_unknown() -> None:
         _infer_file_io_from_scheme("unknown://bucket/path/", {})
 
     assert str(w[0].message) == "No preferred file implementation for scheme: unknown"
+
+
+@pytest.mark.parametrize(
+    "path",
+    ["s3://bucket/key", "hdfs://cluster/path", "gs://bucket/obj", "/tmp/foo", ""],
+)
+def test_is_local_path_false_for_uris_and_posix(path: str) -> None:
+    assert _is_local_path(path) is False
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only behavior")
+@pytest.mark.parametrize("path", [r"C:\Users\test", r"D:\data\iceberg", "E:/warehouse"])
+def test_is_local_path_true_on_windows(path: str) -> None:
+    assert _is_local_path(path) is True
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only behavior")
+def test_infer_file_io_from_scheme_windows_path() -> None:
+    result = _infer_file_io_from_scheme(r"C:\Users\test\warehouse", {})
+    assert isinstance(result, PyArrowFileIO)

tests/io/test_pyarrow.py

@@ -17,6 +17,7 @@
 # pylint: disable=protected-access,unused-argument,redefined-outer-name
 import logging
 import os
+import sys
 import tempfile
 import uuid
 import warnings
@@ -2326,6 +2327,17 @@ def check_results(location: str, expected_schema: str, expected_netloc: str, exp
     check_results("/root/tmp/foo.txt", "file", "", "/root/tmp/foo.txt")
 
 
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only behavior")
+def test_parse_location_windows_drive_letter() -> None:
+    """Windows drive letters should be treated as local file paths, not URL schemes."""
+    for drive in ("C", "D", "c", "d"):
+        path = f"{drive}:\\Users\\test\\file.avro"
+        scheme, netloc, result_path = PyArrowFileIO.parse_location(path)
+        assert scheme == "file"
+        assert netloc == ""
+        assert result_path == os.path.abspath(path)
+
+
 def test_make_compatible_name() -> None:
     assert make_compatible_name("label/abc") == "label_x2Fabc"
     assert make_compatible_name("label?abc") == "label_x3Fabc"