From 1df9df01a55f76c43074249537223c7e7aef5821 Mon Sep 17 00:00:00 2001 From: carloea2 Date: Sun, 25 Oct 2026 22:02:35 +0000 Subject: [PATCH] fix(pyamber): isolate REST catalog S3 signers (#9129) ### What changes were proposed in this PR? Disable fsspec's global S3 instance cache when creating a REST catalog. PyIceberg already caches each filesystem per FileIO thread. Sharing another cached filesystem lets a second table temporarily remove the first table's request signer, causing `403 AccessDenied` during materialization reads. Before: signer replacement can interrupt an existing read. After: FileIO instances have separate signers and retain their own filesystem caches. This is a workaround for apache/iceberg-python#3896. It disables the global S3 instance cache for the Python process, including other fsspec S3 users. Local-filesystem caching is unchanged. It does not change credentials or retry failed reads. ### Any related issues, documentation, discussions? Closes #9128 Related to the failing loop integration check on #8957. Its timestamp change stays separate. ### How was this PR tested? Fresh detached worktree from upstream main at `98c63374b7`, Python 3.12 on Windows. The new regression first failed in two cache configurations. After the fix, all 71 focused storage tests pass, with the separately marked REST integration test excluded. The regression uses real FileIO and S3 filesystem instances, verifies distinct signers between FileIO instances, and verifies that per-thread and local-filesystem caches remain intact. ```text bash bin/python-proto-gen.sh cd amber python -m pytest src/test/python/core/storage/iceberg -m "not integration" -q --tb=short -p no:cacheprovider python -m ruff check src/main/python src/test/python python -m ruff format --check src/main/python src/test/python ``` Live verification used an isolated RustFS rc.6 and Lakekeeper 0.11.0 stack. A real Parquet read succeeds before signer replacement, fails during a controlled replacement gap, and succeeds after registration resumes. With the fix, 40 concurrent real S3 reads pass. Both Ruff checks pass for all 217 Python files. The full engine loop suite also passes with this worktree's freshly built JVM classes and Python workers against that REST-backed stack: 8 passed, 0 failed, 0 canceled. This checks materialized row counts for single and nested loops, JVM operators, loop-variable references, reconverging branches, and state-emitting UDFs. No test assertions or timeouts were changed. From the repository root, with the private stack's storage environment configured: ```text sbt -batch "WorkflowExecutionService/testOnly *LoopIntegrationSpec -- -oDF" sbt -batch scalafmtCheckAll ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Codex (backported from commit 3b8e43e949025292757cf283c3926ee3d1d00ffb) Co-authored-by: Xuan Gu <162244362+xuang7@users.noreply.github.com> --- .../core/storage/iceberg/iceberg_utils.py | 4 +++ .../iceberg/test_iceberg_utils_catalog.py | 26 +++++++++++++++++++ 2 files changed, 30 insertions(+) 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 88e1774b6a6..63148598c35 100644 --- a/amber/src/main/python/core/storage/iceberg/iceberg_utils.py +++ b/amber/src/main/python/core/storage/iceberg/iceberg_utils.py @@ -21,6 +21,7 @@ import pyarrow as pa import pyiceberg.table +from fsspec.config import conf as fsspec_conf from pyiceberg.catalog import Catalog, load_catalog from pyiceberg.catalog.sql import SqlCatalog from pyiceberg.expressions import AlwaysTrue @@ -233,6 +234,9 @@ def create_rest_catalog( :param rest_uri: the URI of the REST catalog endpoint. :return: a Catalog instance (REST catalog). """ + # PyIceberg caches filesystems per FileIO thread. The additional global S3 + # cache shares mutable signers between them: apache/iceberg-python#3896. + fsspec_conf.setdefault("s3", {})["skip_instance_cache"] = True # 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. 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 c3fb6e35413..4c180e751f6 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 @@ -17,6 +17,10 @@ from unittest.mock import MagicMock, patch +import pytest +from fsspec.config import conf +from fsspec.implementations.local import LocalFileSystem +from pyiceberg.io.fsspec import FsspecFileIO from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC from core.storage.iceberg import iceberg_utils @@ -284,6 +288,28 @@ def _make(self, catalog_name, warehouse_name, rest_uri): assert catalog is mock_load_catalog.return_value return mock_load_catalog.call_args + @pytest.mark.parametrize("cached", [None, False, True]) + def test_signed_filesystems_are_isolated_between_file_ios( + self, monkeypatch, cached + ): + options = {} if cached is None else {"skip_instance_cache": cached} + monkeypatch.setitem(conf, "s3", options) + self._make("review", "warehouse", "http://127.0.0.1:1/catalog") + properties = { + "s3.signer": "S3V4RestSigner", + "uri": "http://127.0.0.1:1/catalog", + "s3.endpoint": "http://127.0.0.1:1", + "s3.region": "us-west-2", + "s3.access-key-id": "test", + "s3.secret-access-key": "test", + } + first = FsspecFileIO(properties) + second = FsspecFileIO(properties) + filesystem = first.get_fs("s3") + assert second.get_fs("s3") is not filesystem + assert first.get_fs("s3") is filesystem + assert LocalFileSystem() is LocalFileSystem() + def test_catalog_name_uri_and_warehouse_are_forwarded_by_value(self): args, kwargs = self._make( "texera_iceberg", "texera-warehouse", "http://lakekeeper:8181/catalog"