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"