Skip to content

Replace predicate merges with equality-delete row deltas #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nicosuave merged 5 commits into from
Aug 18, 2026
Merged

Replace predicate merges with equality-delete row deltas #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
754c14d
Pin the equality-delete PyIceberg fork
nicosuave Aug 18, 2026
ac51be5
Allow pinned Git dependencies in package metadata
nicosuave Aug 18, 2026
f2ef363
Replace predicate merges with equality-delete row deltas
nicosuave Aug 18, 2026
9b7e445
Remove the obsolete predicate merge builder
nicosuave Aug 18, 2026
a8965ff
Prove large composite-key merges
nicosuave Aug 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 21 additions & 3 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ iceberg_rest(
# Performance tuning
max_retries=5, # Retry attempts for transient failures
retry_backoff_base=2.0, # Exponential backoff multiplier
merge_batch_size=500000, # Rows per batch for merge operations
merge_batch_size=None, # Deprecated compatibility option; ignored
strict_casting=False, # Fail on potential data loss

# Table management
Expand Down Expand Up @@ -339,7 +339,9 @@ Truncates table and inserts new data.
primary_key="user_id"
)
```
Deletes matching rows then inserts new data. Single atomic transaction.
Writes equality deletes for the incoming keys and appends the replacements in a
single Iceberg row-delta snapshot. It does not build a composite predicate or
split the source into independently visible batches.

#### Upsert Strategy
```python
Expand All @@ -348,7 +350,23 @@ Deletes matching rows then inserts new data. Single atomic transaction.
primary_key="user_id"
)
```
Updates existing rows, inserts new rows.
Uses the same equality-delete row delta as delete-insert. Existing rows with the
incoming keys are hidden and the incoming rows are appended atomically.

Merge tables must use Iceberg format version 2 or later, and every reader must
support equality deletes. Composite-key merge size is not capped by a row-count
setting. The deprecated `merge_batch_size` option is accepted for configuration
compatibility but has no effect.

Snapshot visibility is atomic: readers see either the old snapshot or both the
delete files and replacement data. Data/delete files are written before the
catalog commit, so a failed metadata commit can leave unreferenced files for
normal Iceberg orphan-file cleanup. Concurrent commits follow Iceberg sequence
semantics: a later equality delete can hide matching rows from an earlier commit,
while rows committed after that delete survive.

For partitioned tables, every partition source column must be part of the merge
key. Equality-delete merges currently reject tables with evolved partition specs.

#### Hard Deletes

Expand Down
5 changes: 4 additions & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ dependencies = [
"pandas>=2.3.3",
"pyarrow>=21.0.0",
"pydantic<2.11",
"pyiceberg[pyiceberg-core]>=0.10.0",
"pyiceberg[pyiceberg-core] @ git+https://github.com/sidequery/iceberg-python.git@968fb9677987d7a308ce4aeb70e8c441fe8f4e44",
"requests>=2.32.5",
"s3fs>=2024.0.0", # Modern async version required for credential vending compatibility
"sqlalchemy>=2.0.44",
Expand All @@ -45,6 +45,9 @@ build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/dlt_iceberg"]

[tool.hatch.metadata]
allow-direct-references = true

[dependency-groups]
dev = [
"pytest>=8.4.2",
Expand Down
42 changes: 10 additions & 32 deletions src/dlt_iceberg/destination.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
log_error_with_context,
get_user_friendly_error_message,
)
from .merge_utils import build_primary_key_delete_filter
from pyiceberg.io.pyarrow import schema_to_pyarrow

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -60,8 +59,7 @@ def _get_merge_strategy(table_schema: TTableSchema) -> str:
def _execute_delete_insert(iceberg_table, arrow_table, primary_keys: list, identifier: str):
"""Execute delete-insert merge strategy.

Deletes rows matching primary keys in incoming data, then appends new data.
Uses PyIceberg transaction for atomic delete + append.
Writes source-key equality deletes and replacement rows in one row delta.

Args:
iceberg_table: PyIceberg table object
Expand All @@ -72,20 +70,14 @@ def _execute_delete_insert(iceberg_table, arrow_table, primary_keys: list, ident
Returns:
Tuple of (rows_deleted_estimate, rows_inserted)
"""
# Build delete filter from primary key values in incoming data
delete_filter, deleted_estimate = build_primary_key_delete_filter(
arrow_table, primary_keys
)
deleted_estimate = len(arrow_table)

logger.info(
f"Delete-insert for {identifier}: deleting up to {deleted_estimate} "
f"matching rows, inserting {len(arrow_table)} rows"
)

# Execute atomic delete + append using transaction
with iceberg_table.transaction() as txn:
txn.delete(delete_filter)
txn.append(arrow_table)
iceberg_table.upsert_by_equality_delete(df=arrow_table, join_cols=primary_keys)

return (deleted_estimate, len(arrow_table))

Expand Down Expand Up @@ -361,27 +353,13 @@ def _iceberg_rest_handler(
f"using strategy: {merge_strategy}"
)

if merge_strategy == "delete-insert":
# Atomic delete + insert
deleted, inserted = _execute_delete_insert(
iceberg_table, arrow_table, primary_keys, identifier
)
logger.info(
f"Delete-insert completed: ~{deleted} deleted, "
f"{inserted} inserted"
)
else:
# Default: upsert strategy
upsert_result = iceberg_table.upsert(
df=arrow_table,
join_cols=primary_keys,
when_matched_update_all=True,
when_not_matched_insert_all=True,
)
logger.info(
f"Upsert completed: {upsert_result.rows_updated} updated, "
f"{upsert_result.rows_inserted} inserted"
)
deleted, inserted = _execute_delete_insert(
iceberg_table, arrow_table, primary_keys, identifier
)
logger.info(
f"{merge_strategy} equality-delete merge completed: "
f"{deleted} replacement keys, {inserted} inserted rows"
)
else:
raise ValueError(f"Unknown write disposition: {disposition_type}")

Expand Down
117 changes: 37 additions & 80 deletions src/dlt_iceberg/destination_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,6 @@
log_error_with_context,
get_user_friendly_error_message,
)
from .merge_utils import build_primary_key_delete_filter
from pyiceberg.io.pyarrow import schema_to_pyarrow

# MODULE-LEVEL STATE: Accumulate files across client instances
Expand Down Expand Up @@ -147,8 +146,8 @@ class IcebergRestConfiguration(DestinationClientConfiguration):
# Schema casting configuration
strict_casting: bool = False

# Merge batch size (for upsert operations to avoid memory issues)
merge_batch_size: int = 500000
# Deprecated compatibility option. Equality-delete merges are not row-batched.
merge_batch_size: Optional[int] = None

# Table location layout - controls directory structure for table files
# Supports patterns: {namespace}, {dataset_name}, {table_name}
Expand Down Expand Up @@ -1198,61 +1197,52 @@ def _execute_delete_insert(
combined_table: pa.Table,
primary_keys: List[str],
identifier: str,
hard_delete_filter=None,
hard_delete_rows: Optional[pa.Table] = None,
) -> Tuple[int, int, int]:
"""Execute delete-insert merge strategy with optional hard deletes.

Deletes rows matching primary keys in incoming data, then appends new data.
Uses PyIceberg transaction for atomic hard-delete + delete + append.
Writes source-key equality deletes and replacement rows in one row delta.

Args:
iceberg_table: PyIceberg table object
combined_table: Arrow table with data to merge
primary_keys: List of primary key column names
identifier: Table identifier for logging
hard_delete_filter: Optional filter for hard deletes (rows to permanently remove)
hard_delete_rows: Optional delete-only source rows

Returns:
Tuple of (rows_deleted_estimate, rows_inserted, hard_deleted)
"""
# Build delete filter from primary key values in incoming data
delete_filter, deleted_estimate = build_primary_key_delete_filter(
combined_table, primary_keys
)
deleted_estimate = len(combined_table)

logger.info(
f"Delete-insert for {identifier}: deleting up to {deleted_estimate} "
f"matching rows, inserting {len(combined_table)} rows"
)

# Execute atomic hard-delete + delete + append using single transaction
with iceberg_table.transaction() as txn:
# Hard deletes first (permanent removal)
if hard_delete_filter is not None:
txn.delete(hard_delete_filter)
# Then delete-insert for merge
txn.delete(delete_filter)
txn.append(combined_table)
iceberg_table.upsert_by_equality_delete(
df=combined_table,
join_cols=primary_keys,
delete_df=hard_delete_rows,
)

return (deleted_estimate, len(combined_table), 1 if hard_delete_filter else 0)
hard_deleted = len(hard_delete_rows) if hard_delete_rows is not None else 0
return (deleted_estimate, len(combined_table), hard_deleted)

def _prepare_hard_deletes(
self,
combined_table: pa.Table,
primary_keys: List[str],
) -> Tuple[pa.Table, Optional[Any], int]:
) -> Tuple[pa.Table, Optional[pa.Table], int]:
"""
Prepare hard deletes from incoming data (does not execute).

Rows with the hard_delete_column set (non-null) will be deleted.
Returns the filter expression to use in a transaction.
Returns delete-only rows for the equality-delete row delta.

Args:
combined_table: Arrow table with data including possible delete markers
primary_keys: List of primary key column names

Returns:
Tuple of (remaining_rows, delete_filter_or_none, num_to_delete)
Tuple of (remaining_rows, delete_rows_or_none, num_to_delete)
"""
hard_delete_col = self.config.hard_delete_column

Expand All @@ -1273,10 +1263,7 @@ def _prepare_hard_deletes(
if len(rows_to_delete) == 0:
return rows_to_keep, None, 0

# Build delete filter from primary keys of rows to delete
delete_filter, _ = build_primary_key_delete_filter(rows_to_delete, primary_keys)

return rows_to_keep, delete_filter, len(rows_to_delete)
return rows_to_keep, rows_to_delete, len(rows_to_delete)

def _commit_table_files(
self,
Expand Down Expand Up @@ -1428,16 +1415,18 @@ def _commit_table_files(
iceberg_table.append(combined_table)
else:
# Prepare hard deletes (rows marked for deletion)
remaining_rows, hard_delete_filter, num_hard_deletes = self._prepare_hard_deletes(
combined_table, primary_keys
)
remaining_rows, hard_delete_rows, num_hard_deletes = self._prepare_hard_deletes(combined_table)
if num_hard_deletes > 0:
logger.info(f"Prepared {num_hard_deletes} rows for hard delete")

# If all rows were hard deletes, just execute the delete
# If all rows were hard deletes, commit a delete-only row delta.
if len(remaining_rows) == 0:
if hard_delete_filter is not None:
iceberg_table.delete(hard_delete_filter)
if hard_delete_rows is not None:
iceberg_table.upsert_by_equality_delete(
df=remaining_rows,
join_cols=primary_keys,
delete_df=hard_delete_rows,
)
logger.info(f"Executed {num_hard_deletes} hard deletes (no merge needed)")
return

Expand All @@ -1448,50 +1437,18 @@ def _commit_table_files(
f"using strategy: {merge_strategy}"
)

if merge_strategy == "delete-insert":
# Atomic hard-delete + delete + insert in single transaction
deleted, inserted, _ = self._execute_delete_insert(
iceberg_table, remaining_rows, primary_keys, identifier,
hard_delete_filter=hard_delete_filter
)
logger.info(
f"Delete-insert completed: ~{deleted} deleted, "
f"{inserted} inserted"
)
else:
# Default: upsert strategy
# Execute hard deletes first (separate transaction since upsert is atomic)
if hard_delete_filter is not None:
iceberg_table.delete(hard_delete_filter)
logger.info(f"Executed {num_hard_deletes} hard deletes before upsert")

batch_size = self.config.merge_batch_size
total_updated = 0
total_inserted = 0

for batch_start in range(0, len(remaining_rows), batch_size):
batch_end = min(batch_start + batch_size, len(remaining_rows))
batch = remaining_rows.slice(batch_start, batch_end - batch_start)

logger.info(
f"Upserting batch {batch_start//batch_size + 1}: "
f"rows {batch_start} to {batch_end} ({len(batch)} rows)"
)

upsert_result = iceberg_table.upsert(
df=batch,
join_cols=primary_keys,
when_matched_update_all=True,
when_not_matched_insert_all=True,
)

total_updated += upsert_result.rows_updated
total_inserted += upsert_result.rows_inserted

logger.info(
f"Upsert completed: {total_updated} updated, "
f"{total_inserted} inserted across {(len(remaining_rows) + batch_size - 1) // batch_size} batches"
)
deleted, inserted, hard_deleted = self._execute_delete_insert(
iceberg_table,
remaining_rows,
primary_keys,
identifier,
hard_delete_rows=hard_delete_rows,
)
logger.info(
f"{merge_strategy} equality-delete merge completed: "
f"{deleted} replacement keys, {inserted} inserted rows, "
f"{hard_deleted} hard-delete keys"
)
else:
raise ValueError(f"Unknown write disposition: {disposition_type}")

Expand Down
42 changes: 0 additions & 42 deletions src/dlt_iceberg/merge_utils.py
View file
Open in desktop

This file was deleted.

Loading
Loading