fix(pyamber): isolate REST catalog S3 signers

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.

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"