This is an automated email from the ASF dual-hosted git repository.
aglinxinyuan pushed a commit to branch release/v1.2
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/release/v1.2 by this push:
new d70ae8c4ca fix(amber, v1.2): run Iceberg local storage on Windows in
the Python worker (#6546)
d70ae8c4ca is described below
commit d70ae8c4ca0f4d2fc77330d60e5b4fe91d3b3b43
Author: Xinyuan Lin <[email protected]>
AuthorDate: Tue Jul 21 17:20:33 2026 -0700
fix(amber, v1.2): run Iceberg local storage on Windows in the Python worker
(#6546)
### What changes were proposed in this PR?
Backport of apache/texera#6545 to `release/v1.2`.
The Python UDF worker creates its output-port Iceberg storage through
its own
pyiceberg path (`iceberg_document.py`/`document_factory.py` →
`IcebergCatalogInstance.get_instance()` → `create_postgres_catalog`),
which the
Scala/JVM fix in #6488 does not reach. On Windows with a
local-filesystem
`postgres` warehouse that path crashes:
| Symptom | Cause |
|---|---|
| `Unrecognized filesystem type in URI: c` | a bare drive path `C:\...`
is parsed by pyiceberg with the drive letter as a URI scheme |
| `WinError 123` | even a `file:///C:/...` URI is rejected by the
default pyarrow FileIO |
`create_postgres_catalog` now, **only** when the warehouse carries a
Windows drive
letter, normalizes it to a `file:///` URI (built with
`PureWindowsPath.as_posix()`
— not `.as_uri()`, which would percent-encode a space to `%20` that
fsspec would
not decode) and selects `pyiceberg.io.fsspec.FsspecFileIO`, whose
`LocalFileSystem`
handles Windows drive paths. POSIX paths, colon-stripped paths
(`C/...`), and
remote object stores (`s3://...`) keep the plain-path / default-FileIO
convention
the Scala engine relies on (#4409). `create_rest_catalog` is left
untouched
(remote identifier resolved by the REST server, S3FileIO).
### Any related issues, documentation, discussions?
Backport of #6545 (issue #6487, follow-up to #6488). Applies to
`release/v1.2`.
### How was this PR tested?
Clean cherry-pick: `release/v1.2`'s `iceberg_utils.py` and
`test_iceberg_utils_catalog.py` are byte-identical to the fix's base on
`main`,
and the branch pins the same `pyiceberg==0.11.1`, so behavior matches
the
verified `main` change exactly. Squashed from `daa71367b5` +
`900f2eb0f3`.
On this `release/v1.2` branch: the 14 unit tests in
`test_iceberg_utils_catalog.py` pass (OS-independent — they assert on
the
computed catalog props, so they run anywhere), and `ruff check` + `ruff
format
--check` are clean. No `LICENSE-binary` change — this is a pure-Python
code fix
with no new dependency.
No `release/*` label is added — this PR *is* the backport.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.../python/core/storage/iceberg/iceberg_utils.py | 82 ++++++++++-
.../storage/iceberg/test_iceberg_utils_catalog.py | 159 +++++++++++++++++++++
2 files changed, 234 insertions(+), 7 deletions(-)
diff --git a/amber/src/main/python/core/storage/iceberg/iceberg_utils.py
b/amber/src/main/python/core/storage/iceberg/iceberg_utils.py
index c66ba74ba2..88e1774b6a 100644
--- a/amber/src/main/python/core/storage/iceberg/iceberg_utils.py
+++ b/amber/src/main/python/core/storage/iceberg/iceberg_utils.py
@@ -15,6 +15,10 @@
# specific language governing permissions and limitations
# under the License.
+import re
+from pathlib import PureWindowsPath
+from urllib.parse import urlparse
+
import pyarrow as pa
import pyiceberg.table
from pyiceberg.catalog import Catalog, load_catalog
@@ -35,6 +39,58 @@ from core.models.schema.attribute_type import AttributeType,
TO_ARROW_MAPPING
# Suffix used to encode LARGE_BINARY fields in Iceberg (must match Scala
IcebergUtil)
LARGE_BINARY_FIELD_SUFFIX = "__texera_large_binary_ptr"
+# pyiceberg FileIO whose fsspec LocalFileSystem handles Windows drive paths,
+# which the default pyarrow FileIO cannot. See `_is_windows_local_warehouse`.
+_FSSPEC_FILE_IO = "pyiceberg.io.fsspec.FsspecFileIO"
+
+# Matches a filesystem path that starts with a Windows drive letter and a
+# separator, e.g. `C:\...` or `C:/...` (but not the drive-relative `C:foo`).
+_WINDOWS_DRIVE_PATH = re.compile(r"^[A-Za-z]:[\\/]")
+
+
+def _is_windows_local_warehouse(warehouse: str) -> bool:
+ """
+ Whether ``warehouse`` is a LOCAL filesystem path carrying a Windows drive
+ letter, either as a bare path (``C:\\...`` / ``C:/...``) or a ``file://``
+ URI (``file:///C:/...``).
+
+ Remote object stores (``s3://...``), POSIX paths, drive-less relative
paths,
+ and colon-stripped paths (the form the Scala side registers, ``C/...``) are
+ excluded, so the default pyarrow FileIO / plain-path convention that Linux,
+ macOS, CI, and the Scala engine rely on is left untouched (see #4409).
+ """
+ if not warehouse:
+ return False
+ text = str(warehouse)
+ if _WINDOWS_DRIVE_PATH.match(text):
+ return True
+ parsed = urlparse(text)
+ if parsed.scheme.lower() == "file": # `file` URI schemes are
case-insensitive
+ return bool(_WINDOWS_DRIVE_PATH.match(parsed.path.lstrip("/")))
+ return False
+
+
+def _to_file_uri(warehouse: str) -> str:
+ """
+ Normalize a bare Windows drive path (``C:\\...``) to a ``file:///`` URI so
+ pyiceberg does not mis-parse the drive letter as a URI scheme (``c://`` ->
+ "Unrecognized filesystem type in URI: c"). Values that are already URIs
+ (e.g. ``file:///C:/...``) are returned unchanged.
+
+ The URI is built from ``PureWindowsPath.as_posix()`` rather than
+ ``.as_uri()`` so the path is *not* percent-encoded: the fsspec
+ ``LocalFileSystem`` behind ``FsspecFileIO`` does not URL-decode, so an
+ encoded space (``C:\\Users\\John Doe`` -> ``.../John%20Doe``) would send
+ writes to a literally ``%20``-named directory. ``PureWindowsPath`` keeps
the
+ conversion deterministic on every host OS (the result is identical off
+ Windows).
+ """
+ text = str(warehouse)
+ if _WINDOWS_DRIVE_PATH.match(text):
+ return "file:///" + PureWindowsPath(text).as_posix()
+ return text
+
+
# Type mappings
_ICEBERG_TO_AMBER_TYPE_MAPPING = {
"string": "STRING",
@@ -144,13 +200,22 @@ def create_postgres_catalog(
:param password: the password of the postgres database.
:return: a SQLCatalog instance.
"""
- return SqlCatalog(
- catalog_name,
- **{
- "uri":
f"postgresql+pg8000://{username}:{password}@{uri_without_scheme}",
- "warehouse": warehouse_path,
- },
- )
+ props = {
+ "uri":
f"postgresql+pg8000://{username}:{password}@{uri_without_scheme}",
+ "warehouse": warehouse_path,
+ }
+ # On Windows with a local-filesystem warehouse, a bare drive path
(`C:\...`)
+ # crashes pyiceberg, and even a `file:///C:/...` URI is rejected by the
+ # default pyarrow FileIO. Normalize the path to a `file:///` URI and switch
+ # to fsspec's LocalFileSystem, which handles Windows drive paths. This is
+ # the Python-worker counterpart to the Scala/JVM fix in #6488
+ # (WinutilsFreeLocalFileSystem, issue #6487). Gated to warehouses that
+ # actually carry a Windows drive letter, so remote object stores and POSIX
+ # paths keep the default FileIO / plain-path convention (see #4409).
+ if _is_windows_local_warehouse(warehouse_path):
+ props["warehouse"] = _to_file_uri(warehouse_path)
+ props["py-io-impl"] = _FSSPEC_FILE_IO
+ return SqlCatalog(catalog_name, **props)
def create_rest_catalog(
@@ -168,6 +233,9 @@ def create_rest_catalog(
:param rest_uri: the URI of the REST catalog endpoint.
:return: a Catalog instance (REST catalog).
"""
+ # No Windows-local normalization here (unlike create_postgres_catalog):
+ # `warehouse_name` is a logical warehouse identifier resolved by the REST
+ # server, whose I/O goes through S3FileIO, not the local filesystem.
return load_catalog(
catalog_name,
**{
diff --git
a/amber/src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py
b/amber/src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py
index 902829d44c..a387a51597 100644
--- a/amber/src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py
+++ b/amber/src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py
@@ -32,6 +32,10 @@ class TestCreatePostgresCatalog:
written from Python UDFs become unreadable from the Scala/Java engine
(and vice versa). These tests pin the Python side to the same plain-path
convention used on the Scala side.
+
+ Windows drive-letter warehouses (e.g. ``C:\\...``) are the one documented
+ exception -- pyiceberg cannot parse them as-is -- and are covered by
+ ``TestCreatePostgresCatalogWindowsLocal`` below.
"""
def test_warehouse_is_passed_without_file_scheme(self):
@@ -96,3 +100,158 @@ class TestCreatePostgresCatalog:
)
# And warehouse is still the plain path.
assert kwargs["warehouse"] == warehouse_path
+
+
+# Reference the production constant (the fsspec FileIO selected for
Windows-local
+# warehouses) rather than duplicating its literal, so the tests cannot drift
from it.
+_FSSPEC_FILE_IO = iceberg_utils._FSSPEC_FILE_IO
+
+
+class TestCreatePostgresCatalogWindowsLocal:
+ """
+ Windows-local warehouse handling in `create_postgres_catalog`.
+
+ This is the Python-worker counterpart to the Scala/JVM fix in #6488
+ (`WinutilsFreeLocalFileSystem`, issue #6487). The Python UDF worker writes
+ its output-port Iceberg storage through its own pyiceberg path, which #6488
+ does not touch. On a Windows dev machine with a local-filesystem warehouse
+ (`postgres` catalog type), a bare drive path fails two ways:
+
+ 1. `C:\\...` is mis-parsed by pyiceberg as URI scheme ``c`` ->
"Unrecognized
+ filesystem type in URI: c" when a table is created.
+ 2. Even a `file:///C:/...` URI is rejected by pyiceberg's default pyarrow
+ FileIO ("/C:/..." -> WinError 123).
+
+ The fix normalizes a drive path to a `file:///` URI and selects
+ `FsspecFileIO`, whose `LocalFileSystem` handles Windows drive paths. It is
+ gated to warehouses that actually carry a Windows drive letter, so POSIX
+ paths, colon-stripped paths, and remote object stores (`s3://...`) keep the
+ default pyarrow FileIO / plain-path convention (see the invariant pinned by
+ `TestCreatePostgresCatalog`).
+ """
+
+ def _make(self, warehouse_path):
+ with patch.object(iceberg_utils, "SqlCatalog") as mock_sql_catalog:
+ create_postgres_catalog(
+ catalog_name="texera_iceberg",
+ warehouse_path=warehouse_path,
+ uri_without_scheme="localhost:5432/texera_iceberg_catalog",
+ username="texera",
+ password="password",
+ )
+ assert mock_sql_catalog.call_count == 1
+ _, kwargs = mock_sql_catalog.call_args
+ return kwargs
+
+ def test_backslash_drive_path_is_normalized_and_uses_fsspec(self):
+ """
+ A bare Windows drive path (as produced by Java `Path.toString`, using
+ backslashes) is normalized to a `file:///` URI and `FsspecFileIO` is
+ selected.
+ """
+ kwargs = self._make("C:\\Users\\texera\\amber\\workflow-results")
+
+ assert kwargs["warehouse"] ==
"file:///C:/Users/texera/amber/workflow-results"
+ assert kwargs["py-io-impl"] == _FSSPEC_FILE_IO
+
+ def test_forward_slash_drive_path_is_normalized_and_uses_fsspec(self):
+ """A forward-slash drive path is normalized the same way."""
+ kwargs = self._make("C:/Users/texera/warehouse")
+
+ assert kwargs["warehouse"] == "file:///C:/Users/texera/warehouse"
+ assert kwargs["py-io-impl"] == _FSSPEC_FILE_IO
+
+ def test_drive_path_with_space_is_not_percent_encoded(self):
+ """
+ A warehouse path containing a space (very common on Windows -- e.g.
+ ``C:\\Users\\John Doe\\...`` or ``C:\\Program Files\\...``) must be
+ normalized with a RAW space, never ``%20``. The fsspec LocalFileSystem
+ does not URL-decode, so a percent-encoded path would send writes to a
+ literally ``%20``-named directory that the Scala engine never looks in.
+ """
+ kwargs = self._make("C:\\Users\\John Doe\\amber\\workflow-results")
+
+ assert kwargs["warehouse"] == "file:///C:/Users/John
Doe/amber/workflow-results"
+ assert "%20" not in kwargs["warehouse"]
+ assert kwargs["py-io-impl"] == _FSSPEC_FILE_IO
+
+ def test_normalization_is_os_independent(self):
+ """
+ The drive-path normalization must be deterministic regardless of the
+ host OS (this test also runs on Linux CI), so it relies on
+ `PureWindowsPath` rather than the host's `pathlib.Path`.
+ """
+ kwargs = self._make("D:\\data\\iceberg")
+
+ assert kwargs["warehouse"] == "file:///D:/data/iceberg"
+ assert kwargs["py-io-impl"] == _FSSPEC_FILE_IO
+
+ def test_existing_file_uri_with_drive_uses_fsspec_unchanged(self):
+ """
+ A warehouse already expressed as a `file:///C:/...` URI still needs
+ `FsspecFileIO` (pyarrow FileIO rejects it), but must not be re-encoded.
+ """
+ kwargs = self._make("file:///C:/Users/texera/warehouse")
+
+ assert kwargs["warehouse"] == "file:///C:/Users/texera/warehouse"
+ assert kwargs["py-io-impl"] == _FSSPEC_FILE_IO
+
+ def test_uppercase_file_scheme_with_drive_is_detected(self):
+ """
+ `file` URI schemes are case-insensitive, so `FILE:///C:/...` must be
+ detected as Windows-local (and select `FsspecFileIO`) too.
+ """
+ kwargs = self._make("FILE:///C:/Users/texera/warehouse")
+
+ assert kwargs["warehouse"] == "FILE:///C:/Users/texera/warehouse"
+ assert kwargs["py-io-impl"] == _FSSPEC_FILE_IO
+
+ def test_posix_absolute_path_is_untouched(self):
+ """
+ A POSIX warehouse (Linux/macOS/CI) must keep the plain-path convention
+ and the default FileIO -- forcing `file://` here would reintroduce the
+ cross-runtime metadata mismatch fixed in #4409.
+ """
+ kwargs = self._make("/tmp/texera/iceberg-warehouse")
+
+ assert kwargs["warehouse"] == "/tmp/texera/iceberg-warehouse"
+ assert "py-io-impl" not in kwargs
+
+ def test_colon_stripped_path_is_untouched(self):
+ """
+ A colon-stripped path (the form the Scala side registers on Windows,
+ `C/...`) is a plain relative path pyiceberg accepts, so it is forwarded
+ verbatim with the default FileIO.
+ """
+ kwargs = self._make("C/Users/texera/iceberg-warehouse")
+
+ assert kwargs["warehouse"] == "C/Users/texera/iceberg-warehouse"
+ assert "py-io-impl" not in kwargs
+
+ def test_remote_object_store_is_untouched(self):
+ """
+ A remote object-store warehouse (`s3://...`) must not be forced onto
+ `FsspecFileIO`; pyiceberg selects the appropriate FileIO by scheme.
+ """
+ kwargs = self._make("s3://texera-bucket/iceberg-warehouse")
+
+ assert kwargs["warehouse"] == "s3://texera-bucket/iceberg-warehouse"
+ assert "py-io-impl" not in kwargs
+
+ def test_posix_file_uri_without_drive_is_untouched(self):
+ """
+ The `file://` detection branch must require a drive letter: a POSIX
+ `file:///tmp/...` URI (no drive) is handled fine by the default FileIO
+ and must not be forced onto `FsspecFileIO`.
+ """
+ kwargs = self._make("file:///tmp/texera/iceberg-warehouse")
+
+ assert kwargs["warehouse"] == "file:///tmp/texera/iceberg-warehouse"
+ assert "py-io-impl" not in kwargs
+
+ def test_empty_warehouse_is_untouched(self):
+ """An empty warehouse is treated as non-Windows-local (default
FileIO)."""
+ kwargs = self._make("")
+
+ assert kwargs["warehouse"] == ""
+ assert "py-io-impl" not in kwargs
