Merge branch 'main' of github.com:apache/iceberg-python into fd-rust

Author: Fokko

mkdocs/docs/contributing.md

@@ -155,6 +155,34 @@ make test-integration-rebuild
 
 To rebuild the containers from scratch.
 
+#### Running Integration Tests against REST Catalogs
+
+!!! warning "Do not run against production catalogs"
+    The integration tests will delete data throughout the entirety of your catalog. Running these integration tests against production catalogs will result in data loss.
+
+PyIceberg supports the ability to run our catalog tests against an arbitrary REST Catalog.
+
+In order to run the test catalog, you will need to specify which REST catalog to run against with the `PYICEBERG_TEST_CATALOG` environment variable
+
+```sh
+export PYICEBERG_TEST_CATALOG=test_catalog
+```
+
+The catalog in question can be configured either through the ~/.pyiceberg.yaml file or through environment variables.
+
+```yaml
+catalog:
+  test_catalog:
+    uri: http://rest-catalog/ws/
+    credential: t-1234:secret
+```
+
+```sh
+export PYICEBERG_CATALOG__TEST_CATALOG__URI=thrift://localhost:9083
+export PYICEBERG_CATALOG__TEST_CATALOG__ACCESS_KEY_ID=username
+export PYICEBERG_CATALOG__TEST_CATALOG__SECRET_ACCESS_KEY=password
+```
+
 ## Code standards
 
 Below are the formalized conventions that we adhere to in the PyIceberg project. The goal of this is to have a common agreement on how to evolve the codebase, but also using it as guidelines for newcomers to the project.

pyiceberg/io/fsspec.py

@@ -70,6 +70,7 @@
     S3_ANONYMOUS,
     S3_CONNECT_TIMEOUT,
     S3_ENDPOINT,
+    S3_FORCE_VIRTUAL_ADDRESSING,
     S3_PROXY_URI,
     S3_REGION,
     S3_REQUEST_TIMEOUT,
@@ -168,6 +169,9 @@ def _s3(properties: Properties) -> AbstractFileSystem:
     if request_timeout := properties.get(S3_REQUEST_TIMEOUT):
         config_kwargs["read_timeout"] = float(request_timeout)
 
+    if _force_virtual_addressing := properties.get(S3_FORCE_VIRTUAL_ADDRESSING):
+        config_kwargs["s3"] = {"addressing_style": "virtual"}
+
     if s3_anonymous := properties.get(S3_ANONYMOUS):
         anon = strtobool(s3_anonymous)
     else:

pyiceberg/table/__init__.py

@@ -569,7 +569,7 @@ def overwrite(
         An overwrite may produce zero or more snapshots based on the operation:
 
             - DELETE: In case existing Parquet files can be dropped completely.
-            - REPLACE: In case existing Parquet files need to be rewritten.
+            - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter.
             - APPEND: In case new data is being inserted into the table.
 
         Args:
@@ -629,7 +629,7 @@ def delete(
         A delete may produce zero or more snapshots based on the operation:
 
             - DELETE: In case existing Parquet files can be dropped completely.
-            - REPLACE: In case existing Parquet files need to be rewritten
+            - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the delete filter.
 
         Args:
             delete_filter: A boolean expression to delete rows from a table
@@ -1396,7 +1396,7 @@ def overwrite(
         An overwrite may produce zero or more snapshots based on the operation:
 
             - DELETE: In case existing Parquet files can be dropped completely.
-            - REPLACE: In case existing Parquet files need to be rewritten.
+            - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter..
             - APPEND: In case new data is being inserted into the table.
 
         Args:

pyiceberg/table/update/__init__.py

@@ -193,6 +193,11 @@ class RemoveStatisticsUpdate(IcebergBaseModel):
     snapshot_id: int = Field(alias="snapshot-id")
 
 
+class RemovePartitionSpecsUpdate(IcebergBaseModel):
+    action: Literal["remove-partition-specs"] = Field(default="remove-partition-specs")
+    spec_ids: List[int] = Field(alias="spec-ids")
+
+
 class RemoveSchemasUpdate(IcebergBaseModel):
     action: Literal["remove-schemas"] = Field(default="remove-schemas")
     schema_ids: List[int] = Field(alias="schema-ids")
@@ -227,6 +232,7 @@ class RemovePartitionStatisticsUpdate(IcebergBaseModel):
         RemovePropertiesUpdate,
         SetStatisticsUpdate,
         RemoveStatisticsUpdate,
+        RemovePartitionSpecsUpdate,
         RemoveSchemasUpdate,
         SetPartitionStatisticsUpdate,
         RemovePartitionStatisticsUpdate,
@@ -595,6 +601,21 @@ def _(update: RemoveStatisticsUpdate, base_metadata: TableMetadata, context: _Ta
     return base_metadata.model_copy(update={"statistics": statistics})
 
 
+@_apply_table_update.register(RemovePartitionSpecsUpdate)
+def _(update: RemovePartitionSpecsUpdate, base_metadata: TableMetadata, context: _TableMetadataUpdateContext) -> TableMetadata:
+    for remove_spec_id in update.spec_ids:
+        if not any(spec.spec_id == remove_spec_id for spec in base_metadata.partition_specs):
+            raise ValueError(f"Partition spec with id {remove_spec_id} does not exist")
+
+    if base_metadata.default_spec_id in update.spec_ids:
+        raise ValueError(f"Cannot remove default partition spec: {base_metadata.default_spec_id}")
+
+    partition_specs = [spec for spec in base_metadata.partition_specs if spec.spec_id not in update.spec_ids]
+
+    context.add_update(update)
+    return base_metadata.model_copy(update={"partition_specs": partition_specs})
+
+
 @_apply_table_update.register(RemoveSchemasUpdate)
 def _(update: RemoveSchemasUpdate, base_metadata: TableMetadata, context: _TableMetadataUpdateContext) -> TableMetadata:
     # This method should error if any schemas do not exist.

pyiceberg/table/update/snapshot.py

@@ -810,8 +810,13 @@ class ManageSnapshots(UpdateTableMetadata["ManageSnapshots"]):
        ms.create_tag(snapshot_id1, "Tag_A").create_tag(snapshot_id2, "Tag_B")
     """
 
-    _updates: Tuple[TableUpdate, ...] = ()
-    _requirements: Tuple[TableRequirement, ...] = ()
+    _updates: Tuple[TableUpdate, ...]
+    _requirements: Tuple[TableRequirement, ...]
+
+    def __init__(self, transaction: Transaction) -> None:
+        super().__init__(transaction)
+        self._updates = ()
+        self._requirements = ()
 
     def _commit(self) -> UpdatesAndRequirements:
         """Apply the pending changes and commit."""
@@ -924,9 +929,15 @@ class ExpireSnapshots(UpdateTableMetadata["ExpireSnapshots"]):
     Pending changes are applied on commit.
     """
 
-    _snapshot_ids_to_expire: Set[int] = set()
-    _updates: Tuple[TableUpdate, ...] = ()
-    _requirements: Tuple[TableRequirement, ...] = ()
+    _updates: Tuple[TableUpdate, ...]
+    _requirements: Tuple[TableRequirement, ...]
+    _snapshot_ids_to_expire: Set[int]
+
+    def __init__(self, transaction: Transaction) -> None:
+        super().__init__(transaction)
+        self._updates = ()
+        self._requirements = ()
+        self._snapshot_ids_to_expire = set()
 
     def _commit(self) -> UpdatesAndRequirements:
         """

tests/integration/test_catalog.py

@@ -15,12 +15,13 @@
 #  specific language governing permissions and limitations
 #  under the License.
 
+import os
 from pathlib import Path, PosixPath
 from typing import Generator, List
 
 import pytest
 
-from pyiceberg.catalog import Catalog, MetastoreCatalog
+from pyiceberg.catalog import Catalog, MetastoreCatalog, load_catalog
 from pyiceberg.catalog.hive import HiveCatalog
 from pyiceberg.catalog.memory import InMemoryCatalog
 from pyiceberg.catalog.rest import RestCatalog
@@ -74,6 +75,16 @@ def rest_catalog() -> Generator[Catalog, None, None]:
     clean_up(test_catalog)
 
 
+@pytest.fixture(scope="function")
+def rest_test_catalog() -> Generator[Catalog, None, None]:
+    if test_catalog_name := os.environ.get("PYICEBERG_TEST_CATALOG"):
+        test_catalog = load_catalog(test_catalog_name)
+        yield test_catalog
+        clean_up(test_catalog)
+    else:
+        pytest.skip("PYICEBERG_TEST_CATALOG environment variables not set")
+
+
 @pytest.fixture(scope="function")
 def hive_catalog() -> Generator[Catalog, None, None]:
     test_catalog = HiveCatalog(
@@ -95,6 +106,7 @@ def hive_catalog() -> Generator[Catalog, None, None]:
     pytest.lazy_fixture("sqlite_catalog_file"),
     pytest.lazy_fixture("rest_catalog"),
     pytest.lazy_fixture("hive_catalog"),
+    pytest.lazy_fixture("rest_test_catalog"),
 ]
 
 

tests/io/test_fsspec.py

@@ -306,6 +306,36 @@ def test_fsspec_s3_session_properties() -> None:
         )
 
 
+def test_fsspec_s3_session_properties_force_virtual_addressing() -> None:
+    session_properties: Properties = {
+        "s3.force-virtual-addressing": True,
+        "s3.endpoint": "http://localhost:9000",
+        "s3.access-key-id": "admin",
+        "s3.secret-access-key": "password",
+        "s3.region": "us-east-1",
+        "s3.session-token": "s3.session-token",
+        **UNIFIED_AWS_SESSION_PROPERTIES,
+    }
+
+    with mock.patch("s3fs.S3FileSystem") as mock_s3fs:
+        s3_fileio = FsspecFileIO(properties=session_properties)
+        filename = str(uuid.uuid4())
+
+        s3_fileio.new_input(location=f"s3://warehouse/{filename}")
+
+        mock_s3fs.assert_called_with(
+            anon=False,
+            client_kwargs={
+                "endpoint_url": "http://localhost:9000",
+                "aws_access_key_id": "admin",
+                "aws_secret_access_key": "password",
+                "region_name": "us-east-1",
+                "aws_session_token": "s3.session-token",
+            },
+            config_kwargs={"s3": {"addressing_style": "virtual"}},
+        )
+
+
 def test_fsspec_s3_session_properties_with_anonymous() -> None:
     session_properties: Properties = {
         "s3.anonymous": "true",

tests/table/test_expire_snapshots.py

@@ -14,13 +14,16 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-import datetime
-from unittest.mock import MagicMock
+import threading
+from datetime import datetime, timedelta
+from typing import Dict
+from unittest.mock import MagicMock, Mock
 from uuid import uuid4
 
 import pytest
 
 from pyiceberg.table import CommitTableResponse, Table
+from pyiceberg.table.update.snapshot import ExpireSnapshots
 
 
 def test_cannot_expire_protected_head_snapshot(table_v2: Table) -> None:
@@ -143,7 +146,7 @@ def test_expire_snapshots_by_timestamp_skips_protected(table_v2: Table) -> None:
     table_v2.catalog = MagicMock()
 
     # Attempt to expire all snapshots before a future timestamp (so both are candidates)
-    future_datetime = datetime.datetime.now() + datetime.timedelta(days=1)
+    future_datetime = datetime.now() + timedelta(days=1)
 
     # Mock the catalog's commit_table to return the current metadata (simulate no change)
     mock_response = CommitTableResponse(
@@ -223,3 +226,57 @@ def test_expire_snapshots_by_ids(table_v2: Table) -> None:
     assert EXPIRE_SNAPSHOT_1 not in remaining_snapshots
     assert EXPIRE_SNAPSHOT_2 not in remaining_snapshots
     assert len(table_v2.metadata.snapshots) == 1
+
+
+def test_thread_safety_fix() -> None:
+    """Test that ExpireSnapshots instances have isolated state."""
+    # Create two ExpireSnapshots instances
+    expire1 = ExpireSnapshots(Mock())
+    expire2 = ExpireSnapshots(Mock())
+
+    # Verify they have separate snapshot sets (this was the bug!)
+    # Before fix: both would have the same id (shared class attribute)
+    # After fix: they should have different ids (separate instance attributes)
+    assert id(expire1._snapshot_ids_to_expire) != id(expire2._snapshot_ids_to_expire), (
+        "ExpireSnapshots instances are sharing the same snapshot set - thread safety bug still exists"
+    )
+
+    # Test that modifications to one don't affect the other
+    expire1._snapshot_ids_to_expire.add(1001)
+    expire2._snapshot_ids_to_expire.add(2001)
+
+    # Verify no cross-contamination of snapshot IDs
+    assert 2001 not in expire1._snapshot_ids_to_expire, "Snapshot IDs are leaking between instances"
+    assert 1001 not in expire2._snapshot_ids_to_expire, "Snapshot IDs are leaking between instances"
+
+
+def test_concurrent_operations() -> None:
+    """Test concurrent operations with separate ExpireSnapshots instances."""
+    results: Dict[str, set[int]] = {"expire1_snapshots": set(), "expire2_snapshots": set()}
+
+    def worker1() -> None:
+        expire1 = ExpireSnapshots(Mock())
+        expire1._snapshot_ids_to_expire.update([1001, 1002, 1003])
+        results["expire1_snapshots"] = expire1._snapshot_ids_to_expire.copy()
+
+    def worker2() -> None:
+        expire2 = ExpireSnapshots(Mock())
+        expire2._snapshot_ids_to_expire.update([2001, 2002, 2003])
+        results["expire2_snapshots"] = expire2._snapshot_ids_to_expire.copy()
+
+    # Run both workers concurrently
+    thread1 = threading.Thread(target=worker1)
+    thread2 = threading.Thread(target=worker2)
+
+    thread1.start()
+    thread2.start()
+
+    thread1.join()
+    thread2.join()
+
+    # Check for cross-contamination
+    expected_1 = {1001, 1002, 1003}
+    expected_2 = {2001, 2002, 2003}
+
+    assert results["expire1_snapshots"] == expected_1, "Worker 1 snapshots contaminated"
+    assert results["expire2_snapshots"] == expected_2, "Worker 2 snapshots contaminated"

tests/table/test_init.py

@@ -66,6 +66,7 @@
 )
 from pyiceberg.table.statistics import BlobMetadata, PartitionStatisticsFile, StatisticsFile
 from pyiceberg.table.update import (
+    AddPartitionSpecUpdate,
     AddSnapshotUpdate,
     AddSortOrderUpdate,
     AssertCreate,
@@ -76,6 +77,7 @@
     AssertLastAssignedPartitionId,
     AssertRefSnapshotId,
     AssertTableUUID,
+    RemovePartitionSpecsUpdate,
     RemovePartitionStatisticsUpdate,
     RemovePropertiesUpdate,
     RemoveSchemasUpdate,
@@ -1294,6 +1296,38 @@ def test_update_metadata_log_overflow(table_v2: Table) -> None:
     assert len(new_metadata.metadata_log) == 1
 
 
+def test_remove_partition_spec_update(table_v2: Table) -> None:
+    base_metadata = table_v2.metadata
+    new_spec = PartitionSpec(PartitionField(source_id=2, field_id=1001, transform=IdentityTransform(), name="y"), spec_id=1)
+    metadata_with_new_spec = update_table_metadata(base_metadata, (AddPartitionSpecUpdate(spec=new_spec),))
+
+    assert len(metadata_with_new_spec.partition_specs) == 2
+
+    update = RemovePartitionSpecsUpdate(spec_ids=[1])
+    updated_metadata = update_table_metadata(
+        metadata_with_new_spec,
+        (update,),
+    )
+
+    assert len(updated_metadata.partition_specs) == 1
+
+
+def test_remove_partition_spec_update_spec_does_not_exist(table_v2: Table) -> None:
+    update = RemovePartitionSpecsUpdate(
+        spec_ids=[123],
+    )
+    with pytest.raises(ValueError, match="Partition spec with id 123 does not exist"):
+        update_table_metadata(table_v2.metadata, (update,))
+
+
+def test_remove_partition_spec_update_default_spec(table_v2: Table) -> None:
+    update = RemovePartitionSpecsUpdate(
+        spec_ids=[0],
+    )
+    with pytest.raises(ValueError, match="Cannot remove default partition spec: 0"):
+        update_table_metadata(table_v2.metadata, (update,))
+
+
 def test_remove_schemas_update(table_v2: Table) -> None:
     base_metadata = table_v2.metadata
     assert len(base_metadata.schemas) == 2