diff --git a/.github/workflows/nightly-pypi-build.yml b/.github/workflows/nightly-pypi-build.yml index a302daa436..1f4aae490c 100644 --- a/.github/workflows/nightly-pypi-build.yml +++ b/.github/workflows/nightly-pypi-build.yml @@ -44,7 +44,7 @@ jobs: python-version: 3.12 - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 - name: Set version id: set-version diff --git a/.github/workflows/pypi-build-artifacts.yml b/.github/workflows/pypi-build-artifacts.yml index 5e742fbd8c..2898a63ee7 100644 --- a/.github/workflows/pypi-build-artifacts.yml +++ b/.github/workflows/pypi-build-artifacts.yml @@ -54,7 +54,7 @@ jobs: 3.14 - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 - name: Set version with RC shell: bash diff --git a/.github/workflows/python-ci-docs.yml b/.github/workflows/python-ci-docs.yml index 83f97135e3..cd46fdf258 100644 --- a/.github/workflows/python-ci-docs.yml +++ b/.github/workflows/python-ci-docs.yml @@ -45,7 +45,7 @@ jobs: with: python-version: 3.12 - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 - name: Build docs run: make docs-build - name: Run linters diff --git a/.github/workflows/python-ci.yml b/.github/workflows/python-ci.yml index 8ab526097f..cdc1b2abe9 100644 --- a/.github/workflows/python-ci.yml +++ b/.github/workflows/python-ci.yml @@ -63,7 +63,7 @@ jobs: with: python-version: ${{ matrix.python }} - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: true - name: Install system dependencies @@ -95,7 +95,7 @@ jobs: with: python-version: ${{ matrix.python }} - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: true - name: Install @@ -113,7 +113,7 @@ jobs: with: python-version: '3.12' - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: true # Why this exists: diff --git a/.github/workflows/python-integration.yml b/.github/workflows/python-integration.yml index 8c7337d1d3..94f0eb9044 100644 --- a/.github/workflows/python-integration.yml +++ b/.github/workflows/python-integration.yml @@ -51,7 +51,7 @@ jobs: with: python-version: '3.12' - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: true - name: Install system dependencies @@ -81,7 +81,7 @@ jobs: with: python-version: '3.12' - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: true - name: Install system dependencies @@ -111,7 +111,7 @@ jobs: with: python-version: '3.12' - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: true - name: Install system dependencies @@ -141,7 +141,7 @@ jobs: with: python-version: '3.12' - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: true - name: Install system dependencies @@ -171,7 +171,7 @@ jobs: with: python-version: '3.12' - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: true - name: Install dependencies diff --git a/.github/workflows/python-release-docs.yml b/.github/workflows/python-release-docs.yml index f893f9798c..12475f9791 100644 --- a/.github/workflows/python-release-docs.yml +++ b/.github/workflows/python-release-docs.yml @@ -42,7 +42,7 @@ jobs: with: python-version: 3.12 - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 - name: Build docs run: make docs-build - name: Copy diff --git a/.github/workflows/python-release.yml b/.github/workflows/python-release.yml index 28f784c31e..d409757251 100644 --- a/.github/workflows/python-release.yml +++ b/.github/workflows/python-release.yml @@ -110,7 +110,7 @@ jobs: python-version: 3.12 - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 with: enable-cache: false diff --git a/.github/workflows/sidequery-pypi.yml b/.github/workflows/sidequery-pypi.yml index dd0608ed26..5a60e0fdc3 100644 --- a/.github/workflows/sidequery-pypi.yml +++ b/.github/workflows/sidequery-pypi.yml @@ -26,7 +26,7 @@ on: version: description: Distribution version required: true - default: 0.12.0.post1 + default: 0.12.0.post2 type: string permissions: @@ -36,7 +36,7 @@ jobs: build: uses: ./.github/workflows/pypi-build-artifacts.yml with: - VERSION: ${{ inputs.version || '0.12.0.post1' }} + VERSION: ${{ inputs.version || '0.12.0.post2' }} publish: needs: build @@ -50,7 +50,7 @@ jobs: - name: Download release artifacts uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 with: - name: pypi-release-candidate-${{ inputs.version || '0.12.0.post1' }} + name: pypi-release-candidate-${{ inputs.version || '0.12.0.post2' }} path: dist - name: Publish to PyPI diff --git a/.github/workflows/svn-build-artifacts.yml b/.github/workflows/svn-build-artifacts.yml index 250901b9f9..04e2a3c9c3 100644 --- a/.github/workflows/svn-build-artifacts.yml +++ b/.github/workflows/svn-build-artifacts.yml @@ -54,7 +54,7 @@ jobs: 3.14 - name: Install UV - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 # Publish the source distribution with the version that's in # the repository, otherwise the tests will fail diff --git a/Makefile b/Makefile index 9a73b240a7..7bbc05e9db 100644 --- a/Makefile +++ b/Makefile @@ -129,7 +129,7 @@ test-integration-rebuild: ## Rebuild integration Docker services from scratch docker compose -f dev/docker-compose-integration.yml build --no-cache test-s3: ## Run tests marked with @pytest.mark.s3 - sh ./dev/run-minio.sh + sh ./dev/run-s3.sh $(TEST_RUNNER) pytest tests/ -m s3 $(PYTEST_ARGS) test-adls: ## Run tests marked with @pytest.mark.adls diff --git a/dev/docker-compose-integration.yml b/dev/docker-compose-integration.yml index 32cdccbb1a..3eab22d9f7 100644 --- a/dev/docker-compose-integration.yml +++ b/dev/docker-compose-integration.yml @@ -28,7 +28,7 @@ services: depends_on: - rest - hive - - minio + - object-store ports: - 15002:15002 # Spark Connect - 4040:4040 # Spark UI @@ -39,7 +39,7 @@ services: links: - rest:rest - hive:hive - - minio:minio + - object-store:object-store healthcheck: test: ["CMD", "sh", "-c", "netstat -an | grep 15002 | grep LISTEN"] interval: 30s @@ -59,41 +59,51 @@ services: - AWS_REGION=us-east-1 - CATALOG_WAREHOUSE=s3://warehouse/ - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO - - CATALOG_S3_ENDPOINT=http://minio:9000 + - CATALOG_S3_ENDPOINT=http://object-store:9000 - CATALOG_JDBC_STRICT__MODE=true - minio: - image: minio/minio - container_name: pyiceberg-minio + object-store: + image: rustfs/rustfs:1.0.0-rc.5 + container_name: pyiceberg-object-store networks: iceberg_net: aliases: - - warehouse.minio + - warehouse.object-store ports: - 9001:9001 - 9000:9000 environment: - - MINIO_ROOT_USER=admin - - MINIO_ROOT_PASSWORD=password - - MINIO_DOMAIN=minio - command: ["server", "/data", "--console-address", ":9001"] - mc: - image: minio/mc - container_name: pyiceberg-mc + - RUSTFS_ACCESS_KEY=admin + - RUSTFS_SECRET_KEY=password + - RUSTFS_VOLUMES=/data + - RUSTFS_ADDRESS=:9000 + - RUSTFS_CONSOLE_ENABLE=true + - RUSTFS_CONSOLE_ADDRESS=:9001 + # Resolves virtual-hosted-style `warehouse.object-store` to the warehouse + # bucket. The port is required until rustfs/rustfs#7051 ships. + - RUSTFS_SERVER_DOMAINS=object-store:9000 + healthcheck: + test: ["CMD-SHELL", "curl --fail http://127.0.0.1:9000/health"] + interval: 5s + timeout: 5s + retries: 10 + start_period: 5s + aws-cli: + image: amazon/aws-cli:2.36.40 + container_name: pyiceberg-aws-cli networks: iceberg_net: depends_on: - - minio + object-store: + condition: service_healthy environment: - AWS_ACCESS_KEY_ID=admin - AWS_SECRET_ACCESS_KEY=password - AWS_REGION=us-east-1 - entrypoint: > - /bin/sh -c " - until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done; - /usr/bin/mc mb minio/warehouse; - /usr/bin/mc policy set public minio/warehouse; - tail -f /dev/null - " + - AWS_ENDPOINT_URL=http://object-store:9000 + entrypoint: /bin/sh + command: + - -c + - "aws s3 rb s3://warehouse --force 2>/dev/null; aws s3 mb s3://warehouse" hive: image: pyiceberg-hive:latest build: diff --git a/dev/docker-compose.yml b/dev/docker-compose.yml index 609ac8d51f..8ef82731e7 100644 --- a/dev/docker-compose.yml +++ b/dev/docker-compose.yml @@ -16,31 +16,37 @@ # under the License. services: - minio: - image: minio/minio - container_name: pyiceberg-minio - environment: - - MINIO_ROOT_USER=admin - - MINIO_ROOT_PASSWORD=password - - MINIO_DOMAIN=minio + object-store: + image: rustfs/rustfs:1.0.0-rc.5 + container_name: pyiceberg-object-store ports: - 9001:9001 - 9000:9000 - command: ["server", "/data", "--console-address", ":9001"] - mc: + environment: + - RUSTFS_ACCESS_KEY=admin + - RUSTFS_SECRET_KEY=password + - RUSTFS_VOLUMES=/data + - RUSTFS_ADDRESS=:9000 + - RUSTFS_CONSOLE_ENABLE=true + - RUSTFS_CONSOLE_ADDRESS=:9001 + healthcheck: + test: ["CMD-SHELL", "curl --fail http://127.0.0.1:9000/health"] + interval: 5s + timeout: 5s + retries: 10 + start_period: 5s + aws-cli: + image: amazon/aws-cli:2.36.40 + container_name: pyiceberg-aws-cli depends_on: - - minio - image: minio/mc - container_name: pyiceberg-mc + object-store: + condition: service_healthy environment: - AWS_ACCESS_KEY_ID=admin - AWS_SECRET_ACCESS_KEY=password - AWS_REGION=us-east-1 - entrypoint: > - /bin/sh -c " - until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done; - /usr/bin/mc rm -r --force minio/warehouse; - /usr/bin/mc mb minio/warehouse; - /usr/bin/mc policy set public minio/warehouse; - exit 0; - " + - AWS_ENDPOINT_URL=http://object-store:9000 + entrypoint: /bin/sh + command: + - -c + - "aws s3 rb s3://warehouse --force 2>/dev/null; aws s3 mb s3://warehouse" diff --git a/dev/hive/Dockerfile b/dev/hive/Dockerfile index e46a035727..56dfc44b46 100644 --- a/dev/hive/Dockerfile +++ b/dev/hive/Dockerfile @@ -22,19 +22,10 @@ ARG MAVEN_MIRROR=https://repo1.maven.org/maven2 USER root -# Install curl (separate layer - rarely changes) -RUN apt-get update -qq && \ - apt-get -qq -y install --no-install-recommends curl && \ - apt-get clean && \ - rm -rf /var/lib/apt/lists/* - -# Download JARs with retry logic (slow layer - only changes when versions change) -RUN curl -fsSL --retry 3 --retry-delay 5 \ - -o /opt/hive/lib/hadoop-aws-${HADOOP_VERSION}.jar \ - "${MAVEN_MIRROR}/org/apache/hadoop/hadoop-aws/${HADOOP_VERSION}/hadoop-aws-${HADOOP_VERSION}.jar" && \ - curl -fsSL --retry 3 --retry-delay 5 \ - -o /opt/hive/lib/aws-java-sdk-bundle-${AWS_SDK_BUNDLE}.jar \ - "${MAVEN_MIRROR}/com/amazonaws/aws-java-sdk-bundle/${AWS_SDK_BUNDLE}/aws-java-sdk-bundle-${AWS_SDK_BUNDLE}.jar" +# Download JARs (slow layer - only changes when versions change). ADD fetches them without curl, because the +# image's Debian 11 security repository moved to archive.debian.org and apt-get can no longer install curl. +ADD --chmod=644 ${MAVEN_MIRROR}/org/apache/hadoop/hadoop-aws/${HADOOP_VERSION}/hadoop-aws-${HADOOP_VERSION}.jar /opt/hive/lib/ +ADD --chmod=644 ${MAVEN_MIRROR}/com/amazonaws/aws-java-sdk-bundle/${AWS_SDK_BUNDLE}/aws-java-sdk-bundle-${AWS_SDK_BUNDLE}.jar /opt/hive/lib/ # Copy configuration last (changes more frequently than JARs) COPY core-site.xml /opt/hadoop/etc/hadoop/core-site.xml diff --git a/dev/hive/core-site.xml b/dev/hive/core-site.xml index f5a9473b51..d17656f01a 100644 --- a/dev/hive/core-site.xml +++ b/dev/hive/core-site.xml @@ -32,7 +32,7 @@ fs.s3a.endpoint - http://minio:9000 + http://object-store:9000 fs.s3a.access.key diff --git a/dev/run-minio.sh b/dev/run-s3.sh similarity index 72% rename from dev/run-minio.sh rename to dev/run-s3.sh index 3e4a55c19a..359abd5b1e 100755 --- a/dev/run-minio.sh +++ b/dev/run-s3.sh @@ -20,14 +20,9 @@ set -ex -if [ $(docker ps -q --filter "name=pyiceberg-minio" --filter "status=running" ) ]; then - echo "Minio backend running" +if [ $(docker ps -q --filter "name=pyiceberg-object-store" --filter "health=healthy" ) ]; then + echo "S3 backend running" else docker compose -f dev/docker-compose.yml kill - docker compose -f dev/docker-compose.yml up -d - while [ -z $(docker ps -q --filter "name=pyiceberg-minio" --filter "status=running" ) ] - do - echo "Waiting for Minio" - sleep 1 - done + docker compose -f dev/docker-compose.yml up -d --wait fi diff --git a/dev/spark/spark-defaults.conf b/dev/spark/spark-defaults.conf index 4e50f590c7..a50fd82d62 100644 --- a/dev/spark/spark-defaults.conf +++ b/dev/spark/spark-defaults.conf @@ -23,7 +23,7 @@ spark.sql.catalog.rest.type rest spark.sql.catalog.rest.uri http://rest:8181 spark.sql.catalog.rest.io-impl org.apache.iceberg.aws.s3.S3FileIO spark.sql.catalog.rest.warehouse s3://warehouse/rest/ -spark.sql.catalog.rest.s3.endpoint http://minio:9000 +spark.sql.catalog.rest.s3.endpoint http://object-store:9000 spark.sql.catalog.rest.cache-enabled false # Configure Iceberg Hive catalog @@ -32,13 +32,13 @@ spark.sql.catalog.hive.type hive spark.sql.catalog.hive.uri thrift://hive:9083 spark.sql.catalog.hive.io-impl org.apache.iceberg.aws.s3.S3FileIO spark.sql.catalog.hive.warehouse s3://warehouse/hive/ -spark.sql.catalog.hive.s3.endpoint http://minio:9000 +spark.sql.catalog.hive.s3.endpoint http://object-store:9000 # Configure Spark's default session catalog (spark_catalog) to use Iceberg backed by the Hive Metastore spark.sql.catalog.spark_catalog org.apache.iceberg.spark.SparkSessionCatalog spark.sql.catalog.spark_catalog.type hive spark.sql.catalog.spark_catalog.uri thrift://hive:9083 -spark.hadoop.fs.s3a.endpoint http://minio:9000 +spark.hadoop.fs.s3a.endpoint http://object-store:9000 spark.sql.catalogImplementation hive spark.sql.warehouse.dir s3a://warehouse/hive/ diff --git a/mkdocs/docs/contributing.md b/mkdocs/docs/contributing.md index 0155907a65..025109346a 100644 --- a/mkdocs/docs/contributing.md +++ b/mkdocs/docs/contributing.md @@ -155,7 +155,7 @@ For Python, `pytest` is used a testing framework in combination with `coverage` make test ``` -By default, S3 and ADLS tests are ignored because that require minio and azurite to be running. +By default, S3 and ADLS tests are ignored because that require RustFS and azurite to be running. To run the S3 suite: ```bash @@ -257,7 +257,7 @@ This command spins up the full integration test infrastructure via Docker Compos - **Spark** (with Spark Connect) - **Iceberg REST Catalog** (using the [`apache/iceberg-rest-fixture`](https://hub.docker.com/r/apache/iceberg-rest-fixture) image) - **Hive Metastore** -- **S3-compatible object storage** (Minio) +- **S3-compatible object storage** (RustFS) **Spark Example Notebook** (`notebooks/spark_integration_example.ipynb`) is based on the [Spark Getting Started](https://iceberg.apache.org/docs/nightly/spark-getting-started/) guide. This notebook demonstrates how to work with PyIceberg alongside Spark, leveraging the Docker-based testing setup for a complete local development environment. diff --git a/pyiceberg/__init__.py b/pyiceberg/__init__.py index 28fdae65bc..42d07c9a14 100644 --- a/pyiceberg/__init__.py +++ b/pyiceberg/__init__.py @@ -15,4 +15,4 @@ # specific language governing permissions and limitations # under the License. -__version__ = "0.12.0.post1" +__version__ = "0.12.0.post2" diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index bf9470e2b4..fb17ccfb69 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -86,7 +86,12 @@ update_table_metadata, ) from pyiceberg.table.update.schema import UpdateSchema -from pyiceberg.table.update.snapshot import ManageSnapshots, UpdateSnapshot, _FastAppendFiles +from pyiceberg.table.update.snapshot import ( + ManageSnapshots, + UpdateSnapshot, + _FastAppendFiles, + _validate_equality_delete_spec, +) from pyiceberg.table.update.sorting import UpdateSortOrder from pyiceberg.table.update.spec import UpdateSpec from pyiceberg.table.update.statistics import UpdateStatistics @@ -1012,9 +1017,10 @@ def upsert_by_equality_delete( Concurrent commits are ordered by their Iceberg snapshot sequence. A later equality-delete commit can remove matching rows from an earlier concurrent commit; rows committed later survive. Partitioned tables require every - partition source column in ``join_cols`` and currently reject evolved - partition specs because one partition-scoped delete cannot safely cover - data written under a different spec. + partition source column in ``join_cols``. After partition spec evolution, + every live data file on ``branch`` must already use the current spec (for + example after ``rewrite_data_files``), because a partition-scoped delete + only applies to data files written under the same spec. """ try: import pyarrow as pa @@ -1063,8 +1069,7 @@ def upsert_by_equality_delete( raise ValueError("Duplicate rows found in source datasets based on the key columns. No upsert executed") if delete_rows.num_rows == 0: return - if not self.table_metadata.spec().is_unpartitioned() and len(self.table_metadata.specs()) > 1: - raise NotImplementedError("Equality-delete upserts do not yet support evolved partition specs") + _validate_equality_delete_spec(self.table_metadata, self._table.io, branch, self.table_metadata.default_spec_id) equality_ids = [ self.table_metadata.schema().find_field(column_name, case_sensitive=case_sensitive).field_id diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 6359163767..952f4e4bb1 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -97,6 +97,49 @@ def _new_manifest_list_file_name(snapshot_id: int, attempt: int, commit_uuid: uu return f"snap-{snapshot_id}-{attempt}-{commit_uuid}.avro" +def _live_data_file_spec_ids(snapshot: Snapshot, io: FileIO) -> set[int]: + """Return the partition spec ids of live data files in a snapshot, read from manifest metadata.""" + spec_ids: set[int] = set() + for manifest in snapshot.manifests(io): + if manifest.content != ManifestContent.DATA or manifest.partition_spec_id in spec_ids: + continue + if manifest.added_files_count is not None and manifest.existing_files_count is not None: + has_live_files = manifest.added_files_count + manifest.existing_files_count > 0 + else: + has_live_files = len(manifest.fetch_manifest_entry(io, discard_deleted=True)) > 0 + if has_live_files: + spec_ids.add(manifest.partition_spec_id) + return spec_ids + + +def _validate_equality_delete_spec(table_metadata: TableMetadata, io: FileIO, branch: str | None, spec_id: int) -> None: + """Validate that equality deletes written under ``spec_id`` apply to every live data file on ``branch``. + + A partitioned equality delete only applies to data files with the same spec id, so every live data + file must use that spec. An unpartitioned equality delete is global and applies to every data file. + Delete files under older partitioned specs do not block the upsert: they only apply to data files with + their own spec id, so once no data files remain under those specs they are dangling and have no effect + on reads. + """ + if table_metadata.default_spec_id != spec_id: + raise ValidationException( + f"Default partition spec changed from {spec_id} to {table_metadata.default_spec_id} " + "while writing equality deletes; retry the upsert" + ) + if table_metadata.specs()[spec_id].is_unpartitioned(): + return + snapshot = table_metadata.snapshot_by_name(branch) + if snapshot is None: + return + other_spec_ids = sorted(_live_data_file_spec_ids(snapshot, io) - {spec_id}) + if other_spec_ids: + raise ValidationException( + f"Equality-delete upserts require every live data file to use the current partition spec {spec_id}, " + f"but data files remain under partition spec ids {other_spec_ids}; " + "rewrite them into the current spec first (for example with rewrite_data_files)" + ) + + @dataclass(frozen=True) class CommitWindow: """Tracks the commit range to validate against during retry. @@ -713,6 +756,29 @@ def _deleted_entries(self) -> list[ManifestEntry]: return [] +class _RowDeltaFiles(_FastAppendFiles): + """Fast append of data and delete files that re-validates equality-delete spec safety on retry.""" + + _spec_id: int + + def __init__( + self, + operation: Operation, + transaction: Transaction, + io: FileIO, + snapshot_properties: dict[str, str] = EMPTY_DICT, + branch: str | None = MAIN_BRANCH, + ) -> None: + super().__init__(operation, transaction, io, snapshot_properties=snapshot_properties, branch=branch) + # Added files are written under the default spec at the time the row delta starts. + self._spec_id = transaction.table_metadata.default_spec_id + + def _validate_concurrency(self) -> None: + """Reject a retry when a concurrent commit left data that the added equality deletes cannot reach.""" + if any(data_file.content == DataFileContent.EQUALITY_DELETES for data_file in self._added_data_files): + _validate_equality_delete_spec(self._transaction.table_metadata, self._io, self._target_branch, self._spec_id) + + class _MergeAppendFiles(_FastAppendFiles): _target_size_bytes: int _min_count_to_merge: int @@ -892,9 +958,9 @@ def merge_append(self) -> _MergeAppendFiles: snapshot_properties=self._snapshot_properties, ) - def row_delta(self) -> _FastAppendFiles: + def row_delta(self) -> _RowDeltaFiles: """Create a snapshot that atomically adds data and delete files.""" - return _FastAppendFiles( + return _RowDeltaFiles( operation=Operation.OVERWRITE if self._transaction.table_metadata.snapshot_by_name(name=self._branch) is not None else Operation.APPEND, diff --git a/pyproject.toml b/pyproject.toml index d50c8bdc30..369bb81789 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ # under the License. [project] name = "sidequery-pyiceberg" -version = "0.12.0.post1" +version = "0.12.0.post2" description = "Sidequery's PyIceberg distribution with equality-delete read and write support" authors = [{ name = "Apache Software Foundation", email = "dev@iceberg.apache.org" }] maintainers = [{ name = "Sidequery", email = "hello@sidequery.com" }] @@ -174,8 +174,10 @@ markers = [ # Turns a warning into an error filterwarnings = [ "error", - # Ignore Python version deprecation warning from google.api_core while we still support 3.10 - "ignore:You are using a Python version.*which Google will stop supporting:FutureWarning:google.api_core", + # Ignore Python version support warnings from google.api_core while we still support 3.10. The wording changes + # by date: "which Google will stop supporting" before EOL, "past its end of life" during the one-week grace + # period, and "You are using a non-supported Python version" afterwards (3.10 reached that on 2026-10-12). + "ignore:You are using a (non-supported )?Python version:FutureWarning:google.api_core", # Python 3.13 sqlite3 module ResourceWarnings for unclosed database connections "ignore:unclosed database in pa.Table: + return pa.table( + { + "id": [row[0] for row in rows], + "ts": [row[1] for row in rows], + "value": [row[2] for row in rows], + }, + schema=schema_to_pyarrow(_EVENTS_SCHEMA, include_field_ids=False), + ) + + +def _sorted_events(table: Table, branch: str | None = None) -> list[tuple[int, str]]: + snapshot = table.metadata.snapshot_by_name(branch) + assert snapshot is not None + result = table.scan(snapshot_id=snapshot.snapshot_id).to_arrow().sort_by("id") + return list(zip(result.column("id").to_pylist(), result.column("value").to_pylist(), strict=True)) + + +def _evolve_day_to_month(table: Table) -> None: + with table.update_spec() as update: + update.remove_field("ts_day") + update.add_field("ts", MonthTransform(), "ts_month") + + +def _rewrite_into_current_spec(table: Table) -> None: + table.overwrite(table.scan().to_arrow()) + + +_JAN_1 = datetime(2026, 1, 1, 12) +_JAN_2 = datetime(2026, 1, 2, 12) + + +def test_equality_delete_upsert_on_unevolved_partitioned_table(catalog: InMemoryCatalog) -> None: + table = catalog.create_table("default.equality_day_partitioned", _EVENTS_SCHEMA, partition_spec=_EVENTS_DAY_SPEC) + table.append(_events([(1, _JAN_1, "one"), (2, _JAN_2, "two")])) + + table.upsert_by_equality_delete(_events([(2, _JAN_2, "updated"), (3, _JAN_2, "three")]), join_cols=["id", "ts"]) + + assert _sorted_events(table) == [(1, "one"), (2, "updated"), (3, "three")] + + +def test_equality_delete_upsert_rejects_live_data_under_old_spec(catalog: InMemoryCatalog) -> None: + table = catalog.create_table("default.equality_evolved_old_data", _EVENTS_SCHEMA, partition_spec=_EVENTS_DAY_SPEC) + table.append(_events([(1, _JAN_1, "one")])) + _evolve_day_to_month(table) + table.append(_events([(2, _JAN_2, "two")])) + snapshot_before = table.current_snapshot() + + with pytest.raises( + ValidationException, match=r"current partition spec 1, but data files remain under partition spec ids \[0\]" + ): + table.upsert_by_equality_delete(_events([(1, _JAN_1, "updated")]), join_cols=["id", "ts"]) + + assert table.current_snapshot() == snapshot_before + assert _sorted_events(table) == [(1, "one"), (2, "two")] + + +def test_equality_delete_upsert_after_rewrite_into_evolved_spec(catalog: InMemoryCatalog) -> None: + table = catalog.create_table("default.equality_evolved_rewritten", _EVENTS_SCHEMA, partition_spec=_EVENTS_DAY_SPEC) + table.append(_events([(1, _JAN_1, "one"), (2, _JAN_2, "two")])) + # Leaves an equality-delete manifest under the old spec, which is dangling after the rewrite. + table.upsert_by_equality_delete(_events([(2, _JAN_2, "two-v2")]), join_cols=["id", "ts"]) + _evolve_day_to_month(table) + _rewrite_into_current_spec(table) + + snapshot = table.current_snapshot() + assert snapshot is not None + manifests = snapshot.manifests(table.io) + assert {m.partition_spec_id for m in manifests if m.content == ManifestContent.DELETES} == {0} + assert {m.partition_spec_id for m in manifests if m.content == ManifestContent.DATA and m.has_added_files()} == {1} + + table.upsert_by_equality_delete( + _events([(1, _JAN_1, "one-v2"), (3, _JAN_2, "three")]), + join_cols=["id", "ts"], + delete_df=_events([(2, _JAN_2, "")]).select(["id", "ts"]), + ) + + assert _sorted_events(table) == [(1, "one-v2"), (3, "three")] + snapshot = table.current_snapshot() + assert snapshot is not None + new_delete_files = [ + entry.data_file + for manifest in snapshot.manifests(table.io) + if manifest.content == ManifestContent.DELETES and manifest.added_snapshot_id == snapshot.snapshot_id + for entry in manifest.fetch_manifest_entry(table.io) + ] + assert new_delete_files + assert {data_file.spec_id for data_file in new_delete_files} == {1} + + +def test_equality_delete_upsert_checks_target_branch_spec(catalog: InMemoryCatalog) -> None: + table = catalog.create_table("default.equality_evolved_branch", _EVENTS_SCHEMA, partition_spec=_EVENTS_DAY_SPEC) + table.append(_events([(1, _JAN_1, "one")])) + old_snapshot = table.current_snapshot() + assert old_snapshot is not None + table.manage_snapshots().create_branch(old_snapshot.snapshot_id, "stale").commit() + _evolve_day_to_month(table) + _rewrite_into_current_spec(table) + + with pytest.raises(ValidationException, match=r"partition spec ids \[0\]"): + table.upsert_by_equality_delete(_events([(1, _JAN_1, "branch")]), join_cols=["id", "ts"], branch="stale") + + table.upsert_by_equality_delete(_events([(1, _JAN_1, "main")]), join_cols=["id", "ts"]) + assert _sorted_events(table) == [(1, "main")] + assert _sorted_events(table, branch="stale") == [(1, "one")] + + +def test_equality_delete_upsert_revalidates_spec_on_retry(catalog: InMemoryCatalog) -> None: + table = catalog.create_table("default.equality_evolved_concurrent", _EVENTS_SCHEMA, partition_spec=_EVENTS_DAY_SPEC) + table.append(_events([(1, _JAN_1, "one")])) + stale_table = catalog.load_table("default.equality_evolved_concurrent") + + # A concurrent writer evolves the spec and rewrites the data before the stale writer commits. + _evolve_day_to_month(table) + _rewrite_into_current_spec(table) + + with pytest.raises(ValidationException, match="Default partition spec changed from 0 to 1"): + stale_table.upsert_by_equality_delete(_events([(1, _JAN_1, "stale")]), join_cols=["id", "ts"]) + + table.refresh() + assert _sorted_events(table) == [(1, "one")] + + @pytest.mark.parametrize( ( "join_cols, src_start_row, src_end_row, target_start_row, target_end_row, " diff --git a/uv.lock b/uv.lock index de34a78d27..d4f0672c09 100644 --- a/uv.lock +++ b/uv.lock @@ -5949,7 +5949,7 @@ wheels = [ [[package]] name = "sidequery-pyiceberg" -version = "0.12.0.post1" +version = "0.12.0.post2" source = { editable = "." } dependencies = [ { name = "cachetools" },