# Licensed to the Apache Software Foundation (ASF) under one

# or more contributor license agreements. See the NOTICE file

# distributed with this work for additional information

# regarding copyright ownership. The ASF licenses this file

# to you under the Apache License, Version 2.0 (the

# "License"); you may not use this file except in compliance

# with the License. You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing,

# software distributed under the License is distributed on an

# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

# KIND, either express or implied. See the License for the

# specific language governing permissions and limitations

# under the License.

from __future__ import annotations

import itertools

import logging

import uuid

from abc import abstractmethod

from collections import defaultdict

from collections.abc import Callable

from dataclasses import dataclass

from datetime import datetime

from functools import cached_property

from typing import TYPE_CHECKING, Generic

from pyiceberg.avro.codecs import AvroCompressionCodec

from pyiceberg.exceptions import (

NoSuchSnapshotRefError,

NotAncestorError,

SnapshotRefTypeError,

ValidationException,

)

from pyiceberg.expressions import AlwaysFalse, BooleanExpression, Or

from pyiceberg.expressions.visitors import (

ROWS_MIGHT_NOT_MATCH,

ROWS_MUST_MATCH,

_InclusiveMetricsEvaluator,

_StrictMetricsEvaluator,

inclusive_projection,

manifest_evaluator,

)

from pyiceberg.io import FileIO, OutputFile

from pyiceberg.manifest import (

DataFile,

DataFileContent,

ManifestContent,

ManifestEntry,

ManifestEntryStatus,

ManifestFile,

ManifestWriter,

write_manifest,

write_manifest_list,

)

from pyiceberg.partitioning import PartitionSpec

from pyiceberg.schema import Schema

from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef, SnapshotRefType

from pyiceberg.table.snapshots import (

Operation,

Snapshot,

SnapshotSummaryCollector,

Summary,

ancestors_of,

is_ancestor_of,

latest_ancestor_before_timestamp,

update_snapshot_summaries,

)

from pyiceberg.table.update import (

AddSnapshotUpdate,

AssertRefSnapshotId,

RemoveSnapshotRefUpdate,

RemoveSnapshotsUpdate,

SetSnapshotRefUpdate,

TableRequirement,

TableUpdate,

U,

UpdatesAndRequirements,

UpdateTableMetadata,

)

from pyiceberg.typedef import EMPTY_DICT, KeyDefaultDict, Record

from pyiceberg.utils.bin_packing import ListPacker

from pyiceberg.utils.concurrent import ExecutorFactory

from pyiceberg.utils.datetime import datetime_to_millis

from pyiceberg.utils.properties import property_as_bool, property_as_int

if TYPE_CHECKING:

from pyiceberg.table import Transaction

from pyiceberg.table.metadata import TableMetadata

logger = logging.getLogger(__name__)

def _new_manifest_file_name(num: int, commit_uuid: uuid.UUID) -> str:

return f"{commit_uuid}-m{num}.avro"

def _new_manifest_list_file_name(snapshot_id: int, attempt: int, commit_uuid: uuid.UUID) -> str:

# Mimics the behavior in Java:

# https://github.com/apache/iceberg/blob/c862b9177af8e2d83122220764a056f3b96fd00c/core/src/main/java/org/apache/iceberg/SnapshotProducer.java#L491

return f"snap-{snapshot_id}-{attempt}-{commit_uuid}.avro"

@dataclass(frozen=True)

class CommitWindow:

"""Tracks the commit range to validate against during retry.

base: The snapshot when the operation began (exclusive lower bound, fixed across retries).

head: The branch HEAD snapshot after metadata refresh (inclusive upper bound).

"""

base: Snapshot | None

head: Snapshot | None

@classmethod

def resolve(cls, metadata: TableMetadata, base_id: int | None, branch: str | None) -> CommitWindow:

"""Resolve a CommitWindow from metadata, starting snapshot ID, and target branch."""

head = metadata.snapshot_by_name(branch)

base = metadata.snapshot_by_id(base_id) if base_id is not None else None

if base_id is not None and base is None:

raise ValidationException(f"Cannot find starting snapshot {base_id}")

return cls(base=base, head=head)

def is_empty(self) -> bool:

"""Return True if no concurrent commits occurred (validation can be skipped).

A None base means the operation started on a table with no snapshots. If a head

exists, another writer landed the first snapshot concurrently, so the window is not

empty and validation must run against the whole history.

"""

return self.head is None or (self.base is not None and self.base.snapshot_id == self.head.snapshot_id)

class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):

commit_uuid: uuid.UUID

_io: FileIO

_operation: Operation

_snapshot_id: int

_parent_snapshot_id: int | None

_starting_snapshot_id: int | None

_added_data_files: list[DataFile]

_manifest_num_counter: itertools.count[int]

_deleted_data_files: set[DataFile]

_compression: AvroCompressionCodec

_target_branch: str | None

_predicate: BooleanExpression

_case_sensitive: bool

_commit_window: CommitWindow | None

_written_manifests: list[str]

_uncommitted_manifests: list[str]

_written_manifest_lists: list[str]

_isolation_operation: Operation

def __init__(

self,

operation: Operation,

transaction: Transaction,

io: FileIO,

commit_uuid: uuid.UUID | None = None,

snapshot_properties: dict[str, str] = EMPTY_DICT,

branch: str | None = MAIN_BRANCH,

) -> None:

super().__init__(transaction)

self.commit_uuid = commit_uuid or uuid.uuid4()

self._io = io

self._operation = operation

self._snapshot_id = self._transaction.table_metadata.new_snapshot_id()

self._added_data_files = []

self._deleted_data_files = set()

self.snapshot_properties = snapshot_properties

self._manifest_num_counter = itertools.count(0)

self._written_manifests = []

self._uncommitted_manifests = []

self._written_manifest_lists = []

from pyiceberg.table import TableProperties

self._compression = self._transaction.table_metadata.properties.get( # type: ignore

TableProperties.WRITE_AVRO_COMPRESSION, TableProperties.WRITE_AVRO_COMPRESSION_DEFAULT

)

self._target_branch = self._validate_target_branch(branch=branch)

self._parent_snapshot_id = self._current_branch_head_id()

self._starting_snapshot_id = self._parent_snapshot_id

self._predicate = AlwaysFalse()

self._case_sensitive = True

self._commit_window = None

self._isolation_operation: Operation = Operation.DELETE

def _validate_target_branch(self, branch: str | None) -> str | None:

# if branch is none, write will be written into a staging snapshot

if branch is not None:

if branch in self._transaction.table_metadata.refs:

ref = self._transaction.table_metadata.refs[branch]

if ref.snapshot_ref_type != SnapshotRefType.BRANCH:

raise ValueError(f"{branch} is a tag, not a branch. Tags cannot be targets for producing snapshots")

return branch

def _current_branch_head_id(self) -> int | None:

"""Return the snapshot id at the head of the target branch, or None if the branch has no snapshot."""

snapshot = self._transaction.table_metadata.snapshot_by_name(self._target_branch)

return snapshot.snapshot_id if snapshot else None

def append_data_file(self, data_file: DataFile) -> _SnapshotProducer[U]:

self._added_data_files.append(data_file)

return self

def delete_data_file(self, data_file: DataFile) -> _SnapshotProducer[U]:

self._deleted_data_files.add(data_file)

return self

def _calculate_added_rows(self, manifests: list[ManifestFile]) -> int:

"""Calculate the number of added rows from a list of manifest files."""

added_rows = 0

for manifest in manifests:

if manifest.added_snapshot_id is None or manifest.added_snapshot_id == self._snapshot_id:

if manifest.added_rows_count is None:

raise ValueError(

"Cannot determine number of added rows in snapshot because "

f"the entry for manifest {manifest.manifest_path} is missing the field `added-rows-count`"

)

added_rows += manifest.added_rows_count

return added_rows

@abstractmethod

def _deleted_entries(self) -> list[ManifestEntry]: ...

@abstractmethod

def _existing_manifests(self) -> list[ManifestFile]: ...

def _process_manifests(self, manifests: list[ManifestFile]) -> list[ManifestFile]:

"""To perform any post-processing on the manifests before writing them to the new snapshot."""

return manifests

def _manifests(self) -> list[ManifestFile]:

def _write_added_manifest() -> list[ManifestFile]:

if self._added_data_files:

added_manifests = []

partition_groups: dict[int, list[DataFile]] = defaultdict(list)

for data_file in self._added_data_files:

partition_groups[data_file.spec_id].append(data_file)

for spec_id, data_files in partition_groups.items():

with self.new_manifest_writer(self.spec(spec_id)) as writer:

for data_file in data_files:

writer.add(

ManifestEntry.from_args(

status=ManifestEntryStatus.ADDED,

snapshot_id=self._snapshot_id,

sequence_number=None,

file_sequence_number=None,

data_file=data_file,

)

)

added_manifests.append(writer.to_manifest_file())

return added_manifests

else:

return []

def _write_delete_manifest() -> list[ManifestFile]:

if len(deleted_entries) > 0:

deleted_manifests = []

partition_groups: dict[int, list[ManifestEntry]] = defaultdict(list)

for deleted_entry in deleted_entries:

partition_groups[deleted_entry.data_file.spec_id].append(deleted_entry)

for spec_id, entries in partition_groups.items():

with self.new_manifest_writer(self.spec(spec_id)) as writer:

for entry in entries:

writer.add_entry(entry)

deleted_manifests.append(writer.to_manifest_file())

return deleted_manifests

else:

return []

# Updates self._predicate with computed partition predicate for manifest pruning

self._build_delete_files_partition_predicate()

# Plan deletes before starting manifest writers so validation failures do not leave orphaned manifests

deleted_entries = self._deleted_entries()

executor = ExecutorFactory.get_or_create()

added_manifests = executor.submit(_write_added_manifest)

delete_manifests = executor.submit(_write_delete_manifest)

existing_manifests = executor.submit(self._existing_manifests)

return self._process_manifests(added_manifests.result() + delete_manifests.result() + existing_manifests.result())

def _summary(self, snapshot_properties: dict[str, str] = EMPTY_DICT) -> Summary:

from pyiceberg.table import TableProperties

# avoid copying metadata for each data file

table_metadata = self._transaction.table_metadata

schema = table_metadata.schema()

specs = table_metadata.specs()

partition_summary_limit = int(

table_metadata.properties.get(

TableProperties.WRITE_PARTITION_SUMMARY_LIMIT, TableProperties.WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT

)

)

ssc = SnapshotSummaryCollector(partition_summary_limit=partition_summary_limit)

for data_file in self._added_data_files:

ssc.add_file(

data_file=data_file,

partition_spec=specs[data_file.spec_id],

schema=schema,

)

if len(self._deleted_data_files) > 0:

for data_file in self._deleted_data_files:

ssc.remove_file(

data_file=data_file,

partition_spec=specs[data_file.spec_id],

schema=schema,

)

previous_snapshot = (

table_metadata.snapshot_by_id(self._parent_snapshot_id) if self._parent_snapshot_id is not None else None

)

return update_snapshot_summaries(

summary=Summary(operation=self._operation, **ssc.build(), **snapshot_properties),

previous_summary=previous_snapshot.summary if previous_snapshot is not None else None,

)

def _commit(self) -> UpdatesAndRequirements:

new_manifests = self._manifests()

next_sequence_number = self._transaction.table_metadata.next_sequence_number()

summary = self._summary(self.snapshot_properties)

file_name = _new_manifest_list_file_name(

snapshot_id=self._snapshot_id,

attempt=0,

commit_uuid=self.commit_uuid,

)

location_provider = self._transaction._table.location_provider()

manifest_list_file_path = location_provider.new_metadata_location(file_name)

self._written_manifest_lists.append(manifest_list_file_path)

with write_manifest_list(

format_version=self._transaction.table_metadata.format_version,

output_file=self._io.new_output(manifest_list_file_path),

snapshot_id=self._snapshot_id,

parent_snapshot_id=self._parent_snapshot_id,

sequence_number=next_sequence_number,

avro_compression=self._compression,

) as writer:

writer.add_manifests(new_manifests)

first_row_id: int | None = None

if self._transaction.table_metadata.format_version >= 3:

first_row_id = self._transaction.table_metadata.next_row_id

snapshot = Snapshot(

snapshot_id=self._snapshot_id,

parent_snapshot_id=self._parent_snapshot_id,

manifest_list=manifest_list_file_path,

sequence_number=next_sequence_number,

summary=summary,

schema_id=self._transaction.table_metadata.current_schema_id,

first_row_id=first_row_id,

)

add_snapshot_update = AddSnapshotUpdate(snapshot=snapshot)

if self._target_branch is None:

return (

(add_snapshot_update,),

(),

)

else:

return (

(

add_snapshot_update,

SetSnapshotRefUpdate(

snapshot_id=self._snapshot_id,

parent_snapshot_id=self._parent_snapshot_id,

ref_name=self._target_branch,

type=SnapshotRefType.BRANCH,

),

),

(

AssertRefSnapshotId(

snapshot_id=self._transaction.table_metadata.refs[self._target_branch].snapshot_id

if self._target_branch in self._transaction.table_metadata.refs

else None,

ref=self._target_branch,

),

),

)

@property

def snapshot_id(self) -> int:

return self._snapshot_id

def schema(self) -> Schema:

return self._transaction.table_metadata.schema()

def spec(self, spec_id: int) -> PartitionSpec:

return self._transaction.table_metadata.specs()[spec_id]

def new_manifest_writer(self, spec: PartitionSpec) -> ManifestWriter:

return write_manifest(

format_version=self._transaction.table_metadata.format_version,

spec=spec,

schema=self.schema(),

output_file=self.new_manifest_output(),

snapshot_id=self._snapshot_id,

avro_compression=self._compression,

)

def new_manifest_output(self) -> OutputFile:

location_provider = self._transaction._table.location_provider()

file_name = _new_manifest_file_name(num=next(self._manifest_num_counter), commit_uuid=self.commit_uuid)

file_path = location_provider.new_metadata_location(file_name)

self._written_manifests.append(file_path)

return self._io.new_output(file_path)

def fetch_manifest_entry(self, manifest: ManifestFile, discard_deleted: bool = True) -> list[ManifestEntry]:

return manifest.fetch_manifest_entry(io=self._io, discard_deleted=discard_deleted)

def commit(self) -> None:

self._transaction._register_snapshot_producer(self)

super().commit()

def _cleanup_uncommitted(self) -> None:

"""Delete manifest files and manifest lists from failed retry attempts."""

for path in self._uncommitted_manifests:

try:

self._io.delete(path)

except Exception:

logger.warning("Failed to delete uncommitted manifest: %s", path, exc_info=True)

self._uncommitted_manifests.clear()

# Delete all manifest lists except the last one (which the committed snapshot references)

if len(self._written_manifest_lists) > 1:

for path in self._written_manifest_lists[:-1]:

try:

self._io.delete(path)

except Exception:

logger.warning("Failed to delete uncommitted manifest list: %s", path, exc_info=True)

self._written_manifest_lists = self._written_manifest_lists[-1:]

def _clean_all_uncommitted(self) -> None:

"""Clean up all manifests and manifest lists on abort."""

for path in itertools.chain(self._uncommitted_manifests, self._written_manifests):

try:

self._io.delete(path)

except Exception:

logger.warning("Failed to delete uncommitted manifest: %s", path, exc_info=True)

for path in self._written_manifest_lists:

try:

self._io.delete(path)

except Exception:

logger.warning("Failed to delete uncommitted manifest list: %s", path, exc_info=True)

self._uncommitted_manifests.clear()

self._written_manifests.clear()

self._written_manifest_lists.clear()

def _refresh_for_retry(self) -> None:

"""Reset state for a retry attempt with refreshed metadata."""

self._uncommitted_manifests.extend(self._written_manifests)

self._written_manifests.clear()

self._parent_snapshot_id = self._current_branch_head_id()

# The snapshot id is kept stable across attempts so it acts as an idempotency key: if a

# previous attempt actually landed but its response was lost, the id can be found in the

# refreshed metadata and the commit is not repeated.

self._manifest_num_counter = itertools.count(0)

self.commit_uuid = uuid.uuid4()

def _validate_concurrency(self) -> None:

"""Validate that concurrent changes do not conflict with this operation.

Uses the CommitWindow to determine which catalog commits to validate against.

The window spans from base (when the operation began) to head (latest branch

HEAD), covering all external concurrent commits.

Subclasses that do not require validation (e.g. fast append) should override

with a no-op.

"""

from pyiceberg.table.snapshots import IsolationLevel

from pyiceberg.table.update.validate import (

_validate_added_data_files,

_validate_data_files_exist,

_validate_deleted_data_files,

_validate_no_new_delete_files,

_validate_no_new_deletes_for_data_files,

)

if self._commit_window is None or self._commit_window.is_empty():

return

catalog_head = self._commit_window.head

starting_snapshot = self._commit_window.base

if catalog_head is None:

return

table = self._transaction._table

isolation_level = table.metadata.isolation_level(self._isolation_operation)

conflict_detection_filter = self._predicate if self._predicate != AlwaysFalse() else None

if isolation_level == IsolationLevel.SERIALIZABLE:

_validate_added_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot, self._case_sensitive)

if self._predicate != AlwaysFalse():

# partition_set is None, so any concurrently committed delete file counts as

# a conflict even in unrelated partitions. Scope this via partition projection

# (as Java does) when delete-file write support lands.

_validate_no_new_delete_files(

table, catalog_head, conflict_detection_filter, None, starting_snapshot, self._case_sensitive

)

_validate_deleted_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot, self._case_sensitive)

if self._deleted_data_files:

_validate_data_files_exist(table, catalog_head, self._deleted_data_files, starting_snapshot)

_validate_no_new_deletes_for_data_files(

table,

catalog_head,

conflict_detection_filter,

self._deleted_data_files,

starting_snapshot,

self._case_sensitive,

)

def _build_partition_projection(self, spec_id: int) -> BooleanExpression:

project = inclusive_projection(self.schema(), self.spec(spec_id), self._case_sensitive)

return project(self._predicate)

@cached_property

def partition_filters(self) -> KeyDefaultDict[int, BooleanExpression]:

return KeyDefaultDict(self._build_partition_projection)

def _build_manifest_evaluator(self, spec_id: int) -> Callable[[ManifestFile], bool]:

return manifest_evaluator(self.spec(spec_id), self.schema(), self.partition_filters[spec_id], self._case_sensitive)

def delete_by_predicate(self, predicate: BooleanExpression, case_sensitive: bool = True) -> None:

self._predicate = Or(self._predicate, predicate)

self._case_sensitive = case_sensitive

def _build_delete_files_partition_predicate(self) -> None:

"""Build BooleanExpression based on deleted data files partitions."""

partition_to_overwrite: dict[int, set[Record]] = {}

for data_file in self._deleted_data_files:

group = partition_to_overwrite.setdefault(data_file.spec_id, set())

group.add(data_file.partition)

for spec_id, partition_records in partition_to_overwrite.items():

self.partition_filters[spec_id] = Or(

self.partition_filters[spec_id],

self._transaction._build_partition_predicate(

partition_records=partition_records,

partition_fields=[field.name for field in self.spec(spec_id).fields],

),

)

class _DeleteFiles(_SnapshotProducer["_DeleteFiles"]):

"""Will delete manifest entries from the current snapshot based on the predicate.

This will produce a DELETE snapshot:

Data files were removed and their contents logically deleted and/or delete

files were added to delete rows.

From the specification

"""

# The set of data files this delete was planned to drop, frozen at first plan.

# Reused across retries so the delete is not replanned against a newer head.

_planned_delete_files: set[DataFile] | None = None

def _commit(self) -> UpdatesAndRequirements:

# Only produce a commit when there is something to delete

if self.files_affected:

return super()._commit()

else:

return (), ()

@cached_property

def _compute_deletes(self) -> tuple[list[ManifestFile], list[ManifestEntry], bool]:

"""Computes all the delete operation and cache it when nothing changes.

Returns:

- List of existing manifests that are not affected by the delete operation.

- The manifest-entries that are deleted based on the metadata.

- Flag indicating that rewrites of data-files are needed.

"""

def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) -> ManifestEntry:

return ManifestEntry.from_args(

status=status,

# When a file is replaced or deleted from the dataset, its manifest entry fields store the

# snapshot ID in which the file was deleted and status 2 (deleted).

snapshot_id=self.snapshot_id if status == ManifestEntryStatus.DELETED else entry.snapshot_id,

sequence_number=entry.sequence_number,

file_sequence_number=entry.file_sequence_number,

data_file=entry.data_file,

)

# avoid copying metadata for each evaluator

table_metadata = self._transaction.table_metadata

schema = table_metadata.schema()

manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)

strict_metrics_evaluator = _StrictMetricsEvaluator(schema, self._predicate, case_sensitive=self._case_sensitive).eval

inclusive_metrics_evaluator = _InclusiveMetricsEvaluator(

schema, self._predicate, case_sensitive=self._case_sensitive

).eval

existing_manifests = []

total_deleted_entries = []

partial_rewrites_needed = False

self._deleted_data_files = set()

# Determine the snapshot to read manifests from for deletion

# Should be the current tip of the _target_branch

parent_snapshot_id_for_delete_source = self._parent_snapshot_id

if parent_snapshot_id_for_delete_source is not None:

snapshot = table_metadata.snapshot_by_id(parent_snapshot_id_for_delete_source)

if snapshot: # Ensure snapshot is found

for manifest_file in snapshot.manifests(io=self._io):

if manifest_file.content == ManifestContent.DATA:

if not manifest_evaluators[manifest_file.partition_spec_id](manifest_file):

# If the manifest isn't relevant, we can just keep it in the manifest-list

existing_manifests.append(manifest_file)

else:

# It is relevant, let's check out the content

deleted_entries = []

existing_entries = []

for entry in manifest_file.fetch_manifest_entry(io=self._io, discard_deleted=True):

# On the first plan the predicate decides which files are dropped. On a

# retry the planned set is frozen, so a file is dropped only if it was

# planned for deletion originally. This keeps concurrently added files

# (which the writer never saw) from being dropped when the delete is

# replanned against a newer head under snapshot isolation.

if self._planned_delete_files is not None:

should_delete = entry.data_file in self._planned_delete_files

else:

should_delete = strict_metrics_evaluator(entry.data_file) == ROWS_MUST_MATCH

if should_delete:

# Based on the metadata, it can be dropped right away

deleted_entries.append(_copy_with_new_status(entry, ManifestEntryStatus.DELETED))

self._deleted_data_files.add(entry.data_file)

else:

# Based on the metadata, we cannot determine if it can be deleted

existing_entries.append(_copy_with_new_status(entry, ManifestEntryStatus.EXISTING))

if (

self._planned_delete_files is None

and inclusive_metrics_evaluator(entry.data_file) != ROWS_MIGHT_NOT_MATCH

):

partial_rewrites_needed = True

if len(deleted_entries) > 0:

total_deleted_entries += deleted_entries

# Rewrite the manifest

if len(existing_entries) > 0:

with self.new_manifest_writer(spec=self.spec(manifest_file.partition_spec_id)) as writer:

for existing_entry in existing_entries:

writer.add_entry(existing_entry)

existing_manifests.append(writer.to_manifest_file())

else:

existing_manifests.append(manifest_file)

else:

existing_manifests.append(manifest_file)

if self._planned_delete_files is None:

# Freeze the set of files this delete operates on so retries reuse it instead of replanning.

self._planned_delete_files = set(self._deleted_data_files)

return existing_manifests, total_deleted_entries, partial_rewrites_needed

def _existing_manifests(self) -> list[ManifestFile]:

return self._compute_deletes[0]

def _deleted_entries(self) -> list[ManifestEntry]:

return self._compute_deletes[1]

@property

def rewrites_needed(self) -> bool:

"""Indicate if data files need to be rewritten."""

return self._compute_deletes[2]

@property

def files_affected(self) -> bool:

"""Indicate if any manifest-entries can be dropped."""

return len(self._deleted_entries()) > 0

def _refresh_for_retry(self) -> None:

"""Reset state for a retry attempt, clearing the cached delete computation."""

super()._refresh_for_retry()

# Clear @cached_property by removing it from the instance __dict__.

# _compute_deletes depends on _parent_snapshot_id which changes on retry.

if "_compute_deletes" in self.__dict__:

del self.__dict__["_compute_deletes"]

class _FastAppendFiles(_SnapshotProducer["_FastAppendFiles"]):

def _validate_concurrency(self) -> None:

"""Skip validation; appends do not conflict with other operations."""

def _existing_manifests(self) -> list[ManifestFile]:

"""To determine if there are any existing manifest files.

A fast append will add another ManifestFile to the ManifestList.

All the existing manifest files are considered existing.

"""

existing_manifests = []

if self._parent_snapshot_id is not None:

previous_snapshot = self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)

if previous_snapshot is None:

raise ValueError(f"Snapshot could not be found: {self._parent_snapshot_id}")

for manifest in previous_snapshot.manifests(io=self._io):

if manifest.has_added_files() or manifest.has_existing_files() or manifest.added_snapshot_id == self._snapshot_id:

existing_manifests.append(manifest)

return existing_manifests

def _deleted_entries(self) -> list[ManifestEntry]:

"""To determine if we need to record any deleted manifest entries.

In case of an append, nothing is deleted.

"""

return []

class _MergeAppendFiles(_FastAppendFiles):

_target_size_bytes: int

_min_count_to_merge: int

_merge_enabled: bool

def __init__(

self,

operation: Operation,

transaction: Transaction,

io: FileIO,

branch: str | None = MAIN_BRANCH,

commit_uuid: uuid.UUID | None = None,

snapshot_properties: dict[str, str] = EMPTY_DICT,

) -> None:

from pyiceberg.table import TableProperties

super().__init__(operation, transaction, io, commit_uuid, snapshot_properties, branch)

table_properties = self._transaction.table_metadata.properties

self._target_size_bytes = property_as_int(

table_properties,

TableProperties.MANIFEST_TARGET_SIZE_BYTES,

TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT,

) # type: ignore

self._min_count_to_merge = property_as_int(

table_properties,

TableProperties.MANIFEST_MIN_MERGE_COUNT,

TableProperties.MANIFEST_MIN_MERGE_COUNT_DEFAULT,

) # type: ignore

self._merge_enabled = property_as_bool(

table_properties,

TableProperties.MANIFEST_MERGE_ENABLED,

TableProperties.MANIFEST_MERGE_ENABLED_DEFAULT,

)

def _process_manifests(self, manifests: list[ManifestFile]) -> list[ManifestFile]:

"""To perform any post-processing on the manifests before writing them to the new snapshot.

In _MergeAppendFiles, we merge manifests based on the target size and the minimum count to merge

if automatic merge is enabled.

"""

unmerged_data_manifests = [manifest for manifest in manifests if manifest.content == ManifestContent.DATA]

unmerged_deletes_manifests = [manifest for manifest in manifests if manifest.content == ManifestContent.DELETES]

data_manifest_merge_manager = _ManifestMergeManager(

target_size_bytes=self._target_size_bytes,

min_count_to_merge=self._min_count_to_merge,

merge_enabled=self._merge_enabled,

snapshot_producer=self,

)

return data_manifest_merge_manager.merge_manifests(unmerged_data_manifests) + unmerged_deletes_manifests

class _OverwriteFiles(_SnapshotProducer["_OverwriteFiles"]):

"""Overwrites data from the table. This will produce an OVERWRITE snapshot.

Data and delete files were added and removed in a logical overwrite operation.

"""

def _existing_manifests(self) -> list[ManifestFile]:

"""Determine if there are any existing manifest files."""

existing_files = []

manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)

if snapshot := self._transaction.table_metadata.snapshot_by_name(name=self._target_branch):

for manifest_file in snapshot.manifests(io=self._io):

# Manifest does not contain rows that match the files to delete partitions

if not manifest_evaluators[manifest_file.partition_spec_id](manifest_file):

existing_files.append(manifest_file)

continue

entries_to_write: set[ManifestEntry] = set()

found_deleted_entries: set[ManifestEntry] = set()

for entry in manifest_file.fetch_manifest_entry(io=self._io, discard_deleted=True):

if entry.data_file in self._deleted_data_files:

found_deleted_entries.add(entry)

else:

entries_to_write.add(entry)

# Is the intercept the empty set?

if len(found_deleted_entries) == 0:

existing_files.append(manifest_file)

continue

# Delete all files from manifest

if len(entries_to_write) == 0:

continue

# We have to rewrite the manifest file without the deleted data files

with self.new_manifest_writer(self.spec(manifest_file.partition_spec_id)) as writer:

for entry in entries_to_write:

writer.add_entry(

ManifestEntry.from_args(

status=ManifestEntryStatus.EXISTING,

snapshot_id=entry.snapshot_id,

sequence_number=entry.sequence_number,

file_sequence_number=entry.file_sequence_number,

data_file=entry.data_file,

)

)

existing_files.append(writer.to_manifest_file())

return existing_files

def _deleted_entries(self) -> list[ManifestEntry]:

"""To determine if we need to record any deleted entries.

With a full overwrite all the entries are considered deleted.

With partial overwrites we have to use the predicate to evaluate

which entries are affected.

"""

if self._parent_snapshot_id is not None:

previous_snapshot = self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)

if previous_snapshot is None:

# This should never happen since you cannot overwrite an empty table

raise ValueError(f"Could not find the previous snapshot: {self._parent_snapshot_id}")

executor = ExecutorFactory.get_or_create()

manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)

def _get_entries(manifest: ManifestFile) -> list[ManifestEntry]:

if not manifest_evaluators[manifest.partition_spec_id](manifest):

return []

return [

ManifestEntry.from_args(

status=ManifestEntryStatus.DELETED,

snapshot_id=self._snapshot_id,

sequence_number=entry.sequence_number,

file_sequence_number=entry.file_sequence_number,

data_file=entry.data_file,

)

for entry in manifest.fetch_manifest_entry(self._io, discard_deleted=True)

if entry.data_file.content == DataFileContent.DATA and entry.data_file in self._deleted_data_files

]

list_of_entries = executor.map(_get_entries, previous_snapshot.manifests(self._io))

deleted_entries = list(itertools.chain(*list_of_entries))

else:

deleted_entries = []

self._validate_required_deletes(deleted_entries)

return deleted_entries

def _validate_required_deletes(self, deleted_entries: list[ManifestEntry]) -> None:

"""Validate that explicitly deleted data files exist in the parent snapshot.

Files passed to `delete_data_file` are required deletes. If one is absent, an overwrite

could commit replacement files without the corresponding deletion and produce incorrect

snapshot summary totals.

Args:

deleted_entries: Live parent-snapshot entries selected for deletion.

Raises:

ValidationException: If a required data file is missing.

"""

found_data_files = {entry.data_file for entry in deleted_entries}

if missing := [data_file.file_path for data_file in self._deleted_data_files if data_file not in found_data_files]:

raise ValidationException(f"Missing required files to delete: {', '.join(sorted(missing))}")

class UpdateSnapshot:

_transaction: Transaction

_io: FileIO

_branch: str | None

_snapshot_properties: dict[str, str]

def __init__(

self,

transaction: Transaction,

io: FileIO,

branch: str | None = MAIN_BRANCH,

snapshot_properties: dict[str, str] = EMPTY_DICT,

) -> None:

self._transaction = transaction

self._io = io

self._snapshot_properties = snapshot_properties

self._branch = branch

def fast_append(self) -> _FastAppendFiles:

return _FastAppendFiles(

operation=Operation.APPEND,

transaction=self._transaction,

io=self._io,

branch=self._branch,

snapshot_properties=self._snapshot_properties,

)

def merge_append(self) -> _MergeAppendFiles:

return _MergeAppendFiles(

operation=Operation.APPEND,

transaction=self._transaction,

io=self._io,

branch=self._branch,

snapshot_properties=self._snapshot_properties,

)

def overwrite(self, commit_uuid: uuid.UUID | None = None) -> _OverwriteFiles:

return _OverwriteFiles(

commit_uuid=commit_uuid,

operation=Operation.OVERWRITE

if self._transaction.table_metadata.snapshot_by_name(name=self._branch) is not None

else Operation.APPEND,

transaction=self._transaction,

io=self._io,

branch=self._branch,

snapshot_properties=self._snapshot_properties,

)

def delete(self) -> _DeleteFiles:

return _DeleteFiles(

operation=Operation.DELETE,

transaction=self._transaction,

io=self._io,

branch=self._branch,

snapshot_properties=self._snapshot_properties,

)

class _ManifestMergeManager(Generic[U]):

_target_size_bytes: int

_min_count_to_merge: int

_merge_enabled: bool

_snapshot_producer: _SnapshotProducer[U]

def __init__(

self, target_size_bytes: int, min_count_to_merge: int, merge_enabled: bool, snapshot_producer: _SnapshotProducer[U]

) -> None:

self._target_size_bytes = target_size_bytes

self._min_count_to_merge = min_count_to_merge

self._merge_enabled = merge_enabled

self._snapshot_producer = snapshot_producer

def _group_by_spec(self, manifests: list[ManifestFile]) -> dict[int, list[ManifestFile]]:

groups = defaultdict(list)

for manifest in manifests:

groups[manifest.partition_spec_id].append(manifest)

return groups

def _create_manifest(self, spec_id: int, manifest_bin: list[ManifestFile]) -> ManifestFile:

with self._snapshot_producer.new_manifest_writer(spec=self._snapshot_producer.spec(spec_id)) as writer:

for manifest in manifest_bin:

for entry in self._snapshot_producer.fetch_manifest_entry(manifest=manifest, discard_deleted=False):

if entry.status == ManifestEntryStatus.DELETED and entry.snapshot_id == self._snapshot_producer.snapshot_id:

# only files deleted by this snapshot should be added to the new manifest

writer.delete(entry)

elif entry.status == ManifestEntryStatus.ADDED and entry.snapshot_id == self._snapshot_producer.snapshot_id:

# added entries from this snapshot are still added, otherwise they should be existing

writer.add(entry)

elif entry.is_live():

# add all non-deleted files from the old manifest as existing files

writer.existing(entry)

return writer.to_manifest_file()

def _merge_group(self, first_manifest: ManifestFile, spec_id: int, manifests: list[ManifestFile]) -> list[ManifestFile]:

packer: ListPacker[ManifestFile] = ListPacker(target_weight=self._target_size_bytes, lookback=1, largest_bin_first=False)

bins: list[list[ManifestFile]] = packer.pack_end(manifests, lambda m: m.manifest_length)

def merge_bin(manifest_bin: list[ManifestFile]) -> list[ManifestFile]:

output_manifests = []

if len(manifest_bin) == 1:

output_manifests.append(manifest_bin[0])

elif first_manifest in manifest_bin and len(manifest_bin) < self._min_count_to_merge:

# if the bin has the first manifest (the new data files or an appended manifest file) then only

# merge it if the number of manifests is above the minimum count. this is applied only to bins

# with an in-memory manifest so that large manifests don't prevent merging older groups.