CI: Add Windows unit test job

.github/workflows/python-ci.yml

@@ -79,6 +79,30 @@ jobs:
     - name: Generate coverage report (85%) # Coverage threshold should only increase over time — never decrease it!
       run: COVERAGE_FAIL_UNDER=85 make coverage-report
 
+  windows-unit-test:
+    runs-on: windows-latest
+    timeout-minutes: 15
+    strategy:
+      fail-fast: true
+      matrix:
+        python: ['3.12']
+
+    steps:
+    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      with:
+        persist-credentials: false
+    - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+      with:
+        python-version: ${{ matrix.python }}
+    - name: Install UV
+      uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+      with:
+        enable-cache: true
+    - name: Install
+      run: uv sync --all-extras --no-extra hive-kerberos
+    - name: Run unit tests
+      run: uv run python -m pytest tests/ -m "(unmarked or parametrize) and not integration" --ignore=tests/integration -v -x
+
   cibw-dev-env-smoke-test:
     runs-on: ubuntu-latest
     steps:

pyiceberg/io/__init__.py

@@ -27,6 +27,7 @@
 
 import importlib
 import logging
+import sys
 import warnings
 from abc import ABC, abstractmethod
 from io import SEEK_SET
@@ -41,6 +42,17 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _is_windows_drive_letter(scheme: str) -> bool:
+    r"""Check if a parsed URL scheme is actually a Windows drive letter.
+
+    On Windows, Python's urlparse treats paths like 'C:\\Users\\...' as having
+    scheme='c'. This detects that case so callers can route to the local filesystem.
+    Only returns True on Windows — no behavior change on other platforms.
+    """
+    return sys.platform == "win32" and len(scheme) == 1 and scheme.isalpha()
+
+
 AWS_PROFILE_NAME = "client.profile-name"
 AWS_REGION = "client.region"
 AWS_ACCESS_KEY_ID = "client.access-key-id"
@@ -336,13 +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):
+    scheme = parsed_url.scheme
+
+    if scheme and _is_windows_drive_letter(scheme):
+        # Windows drive letter (e.g. 'C:\...') misidentified as a URL scheme
+        scheme = "file"
+
+    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_windows_drive_letter,
 )
 from pyiceberg.typedef import Properties
 from pyiceberg.types import strtobool
@@ -480,6 +481,8 @@ def delete(self, location: str | InputFile | OutputFile) -> None:
 
     def _get_fs_from_uri(self, uri: "ParseResult") -> AbstractFileSystem:
         """Get a filesystem from a parsed URI, using hostname for ADLS account resolution."""
+        if _is_windows_drive_letter(uri.scheme):
+            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_windows_drive_letter,
 )
 from pyiceberg.io.fileformat import DataFileStatistics as DataFileStatistics
 from pyiceberg.io.fileformat import FileFormatFactory, FileFormatModel, FileFormatWriter
@@ -404,10 +405,12 @@ def parse_location(location: str, properties: Properties = EMPTY_DICT) -> tuple[
         """Return (scheme, netloc, path) for the given location.
 
         Uses DEFAULT_SCHEME and DEFAULT_NETLOC if scheme/netloc are missing.
+        On Windows, single-letter schemes (drive letters like 'C') are treated
+        as local file paths rather than URI schemes.
         """
         uri = urlparse(location)
 
-        if not uri.scheme:
+        if not uri.scheme or _is_windows_drive_letter(uri.scheme):
             default_scheme = properties.get("DEFAULT_SCHEME", "file")
             default_netloc = properties.get("DEFAULT_NETLOC", "")
             return default_scheme, default_netloc, os.path.abspath(location)

tests/catalog/test_hive.py

@@ -18,6 +18,7 @@
 import base64
 import copy
 import struct
+import sys
 import threading
 import uuid
 from collections.abc import Generator
@@ -300,7 +301,7 @@ def test_create_table(
     # it to construct the assert_called_with
     metadata_location: str = called_hive_table.parameters["metadata_location"]
     assert metadata_location.endswith(".metadata.json")
-    assert "/database/table/metadata/" in metadata_location
+    assert "/database/table/metadata/" in metadata_location.replace("\\", "/")
     catalog._client.__enter__().create_table.assert_called_with(
         HiveTable(
             tableName="table",
@@ -481,7 +482,7 @@ def test_create_table_with_given_location_removes_trailing_slash(
     # it to construct the assert_called_with
     metadata_location: str = called_hive_table.parameters["metadata_location"]
     assert metadata_location.endswith(".metadata.json")
-    assert "/database/table-given-location/metadata/" in metadata_location
+    assert "/database/table-given-location/metadata/" in metadata_location.replace("\\", "/")
     catalog._client.__enter__().create_table.assert_called_with(
         HiveTable(
             tableName="table",
@@ -1369,6 +1370,7 @@ def test_create_hive_client_failure() -> None:
         assert mock_hive_client.call_count == 2
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Kerberos/puresasl not available on Windows")
 def test_create_hive_client_with_kerberos(
     kerberized_hive_metastore_fake_url: str,
 ) -> None:
@@ -1381,6 +1383,7 @@ def test_create_hive_client_with_kerberos(
     assert client is not None
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Kerberos/puresasl not available on Windows")
 def test_create_hive_client_with_kerberos_using_context_manager(
     kerberized_hive_metastore_fake_url: str,
 ) -> None:
@@ -1412,6 +1415,7 @@ def test_create_hive_client_with_kerberos_using_context_manager(
             assert open_client._iprot.trans.isOpen()
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Kerberos/puresasl not available on Windows")
 def test_kerberized_client_uses_fresh_transport_on_reuse(
     kerberized_hive_metastore_fake_url: str,
 ) -> None:

tests/catalog/test_sql.py

@@ -61,7 +61,7 @@ def catalog_memory(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog,
 @pytest.fixture(scope="module")
 def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]:
     props = {
-        "uri": f"sqlite:////{warehouse}/sql-catalog",
+        "uri": f"sqlite:///{warehouse.as_posix()}/sql-catalog",
         "warehouse": f"file://{warehouse}",
     }
     catalog = SqlCatalog(catalog_name, **props)
@@ -72,7 +72,7 @@ def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog,
 
 @pytest.fixture(scope="module")
 def catalog_uri(warehouse: Path) -> str:
-    return f"sqlite:////{warehouse}/sql-catalog"
+    return f"sqlite:///{warehouse.as_posix()}/sql-catalog"
 
 
 @pytest.fixture(scope="module")
@@ -96,7 +96,7 @@ def test_creation_with_echo_parameter(catalog_name: str, warehouse: Path) -> Non
 
     for echo_param, expected_echo_value in test_cases:
         props = {
-            "uri": f"sqlite:////{warehouse}/sql-catalog",
+            "uri": f"sqlite:///{warehouse.as_posix()}/sql-catalog",
             "warehouse": f"file://{warehouse}",
         }
         # None is for default value
@@ -119,7 +119,7 @@ def test_creation_with_pool_pre_ping_parameter(catalog_name: str, warehouse: Pat
 
     for pool_pre_ping_param, expected_pool_pre_ping_value in test_cases:
         props = {
-            "uri": f"sqlite:////{warehouse}/sql-catalog",
+            "uri": f"sqlite:///{warehouse.as_posix()}/sql-catalog",
             "warehouse": f"file://{warehouse}",
         }
         # None is for default value
@@ -139,7 +139,7 @@ def test_creation_from_impl(catalog_name: str, warehouse: Path) -> None:
             catalog_name,
             **{
                 "py-catalog-impl": "pyiceberg.catalog.sql.SqlCatalog",
-                "uri": f"sqlite:////{warehouse}/sql-catalog",
+                "uri": f"sqlite:///{warehouse.as_posix()}/sql-catalog",
                 "warehouse": f"file://{warehouse}",
             },
         ),
@@ -268,7 +268,7 @@ def get_columns(engine: Engine) -> set[str]:
 
 
 def test_adds_iceberg_type_column_to_old_schema(warehouse: Path) -> None:
-    uri = f"sqlite:////{warehouse}/test-migration-add-col"
+    uri = f"sqlite:///{warehouse.as_posix()}/test-migration-add-col"
     engine = _create_v0_db(uri)
 
     # Verify the column does not exist in the old schema
@@ -334,7 +334,7 @@ def test_list_tables_filters_by_iceberg_type(warehouse: Path) -> None:
 
 
 def test_migration_to_v1_with_property_set(warehouse: Path) -> None:
-    uri = f"sqlite:////{warehouse}/test-v1-migrate"
+    uri = f"sqlite:///{warehouse.as_posix()}/test-v1-migrate"
     engine = _create_v0_db(uri)
 
     assert "iceberg_type" not in get_columns(engine)
@@ -361,7 +361,7 @@ def test_invalid_schema_version_raises(warehouse: Path) -> None:
 
 def test_list_tables_works_on_v0_schema(warehouse: Path) -> None:
     """list_tables should work on V0 schemas without iceberg_type column."""
-    uri = f"sqlite:////{warehouse}/test-v0-list"
+    uri = f"sqlite:///{warehouse.as_posix()}/test-v0-list"
     _create_v0_db(uri)
 
     catalog = SqlCatalog(
@@ -441,15 +441,15 @@ def _make_v0_catalog(uri: str, warehouse: Path) -> SqlCatalog:
 
 def test_namespace_exists_on_v0_schema(warehouse: Path) -> None:
     """namespace_exists should not fail on V0 schema (no iceberg_type column)."""
-    catalog = _make_v0_catalog(f"sqlite:////{warehouse}/test-v0-ns-exists", warehouse)
+    catalog = _make_v0_catalog(f"sqlite:///{warehouse.as_posix()}/test-v0-ns-exists", warehouse)
     catalog.create_namespace("ns")
     assert catalog.namespace_exists("ns")
     assert not catalog.namespace_exists("missing")
 
 
 def test_create_and_load_table_on_v0_schema(warehouse: Path) -> None:
     """create_table and load_table should work on V0 schema without iceberg_type column."""
-    catalog = _make_v0_catalog(f"sqlite:////{warehouse}/test-v0-create", warehouse)
+    catalog = _make_v0_catalog(f"sqlite:///{warehouse.as_posix()}/test-v0-create", warehouse)
     catalog.create_namespace("ns")
     schema = Schema(NestedField(1, "id", StringType(), required=True))
     tbl = catalog.create_table(("ns", "tbl"), schema)

tests/conftest.py

@@ -2388,7 +2388,7 @@ def empty_home_dir_path(tmp_path_factory: pytest.TempPathFactory) -> str:
     return home_path
 
 
-RANDOM_LENGTH = 20
+RANDOM_LENGTH = 8
 NUM_TABLES = 2
 
 
@@ -2445,15 +2445,15 @@ def hierarchical_namespace_list(hierarchical_namespace_name: str) -> list[str]:
 
 BUCKET_NAME = "test_bucket"
 TABLE_METADATA_LOCATION_REGEX = re.compile(
-    r"""s3://test_bucket/my_iceberg_database-[a-z]{20}.db/
-    my_iceberg_table-[a-z]{20}/metadata/
+    r"""s3://test_bucket/my_iceberg_database-[a-z]{8}.db/
+    my_iceberg_table-[a-z]{8}/metadata/
     [0-9]{5}-[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}.metadata.json""",
     re.X,
 )
 
 BQ_TABLE_METADATA_LOCATION_REGEX = re.compile(
-    r"""gs://alexstephen-test-bq-bucket/my_iceberg_database_[a-z]{20}.db/
-    my_iceberg_table-[a-z]{20}/metadata/
+    r"""gs://alexstephen-test-bq-bucket/my_iceberg_database_[a-z]{8}.db/
+    my_iceberg_table-[a-z]{8}/metadata/
     [0-9]{5}-[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}.metadata.json""",
     re.X,
 )
@@ -3138,7 +3138,7 @@ def _create_sql_without_rowcount_catalog(name: str, warehouse: Path) -> Catalog:
     from pyiceberg.catalog.sql import SqlCatalog
 
     props = {
-        "uri": f"sqlite:////{warehouse}/sql-catalog",
+        "uri": f"sqlite:///{warehouse.as_posix()}/sql-catalog",
         "warehouse": f"file://{warehouse}",
     }
     catalog = SqlCatalog(name, **props)

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_windows_drive_letter,
     load_file_io,
 )
 from pyiceberg.io.pyarrow import PyArrowFileIO
@@ -51,6 +53,7 @@ def test_custom_local_input_file() -> None:
         data = f.read()
         assert data == b"foo"
         assert len(input_file) == 3
+        f.close()
 
 
 def test_custom_local_output_file() -> None:
@@ -89,6 +92,7 @@ def test_pickled_pyarrow_round_trip() -> None:
         data = f.read()
         assert data == b"foo"
         assert len(input_file) == 3
+        f.close()
         deserialized_file_io.delete(location=f"{absolute_file_location}")
 
 
@@ -339,3 +343,31 @@ 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(
+    "scheme",
+    ["s3", "hdfs", "file", "gs", ""],
+)
+def test_is_windows_drive_letter_false_for_real_schemes(scheme: str) -> None:
+    assert _is_windows_drive_letter(scheme) is False
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only behavior")
+@pytest.mark.parametrize("scheme", ["c", "D", "z"])
+def test_is_windows_drive_letter_true_on_windows(scheme: str) -> None:
+    assert _is_windows_drive_letter(scheme) is True
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only behavior")
+def test_infer_file_io_from_scheme_windows_drive_letter() -> None:
+    # A Windows-style path like 'C:\Users\...' should be treated as a local file path,
+    # not as a URL with scheme 'c'
+    result = _infer_file_io_from_scheme(r"C:\Users\test\warehouse", {})
+    assert isinstance(result, PyArrowFileIO)
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only behavior")
+def test_infer_file_io_from_scheme_windows_drive_letter_lowercase() -> None:
+    result = _infer_file_io_from_scheme(r"d:\data\iceberg\table", {})
+    assert isinstance(result, PyArrowFileIO)