feat(iceberg): stream Iceberg writes with constant memory - #3753

Updated with both changes:

GC parametrization: Added iceberg_gc_collect_interval config option (default 0 = disabled). Any positive integer triggers gc.collect() every N batches. Threaded through from FilesystemDestinationClientConfiguration -> IcebergLoadFilesystemJob -> streaming functions. Progress logging is decoupled from GC and always runs every 10 batches.

Tests: Added 3 tests in test_open_table_pipeline.py using table_format_local_configs=True with with_table_format="iceberg":

• test_iceberg_streamed_append_replace: parametrized for append and replace
• test_iceberg_streamed_upsert: verifies updates and inserts with primary key

All passing locally on filesystem. Let me know if this looks good or if any other changes are needed.

I was looking at the failed test cases on CI, all the filesystem-parquet-iceberg tests were failing with:

Unexpected physical type FIXED_LEN_BYTE_ARRAY for decimal(10, 3), expected INT64

1. Decimal encoding mismatch

PyArrow defaults to encoding decimals as FIXED_LEN_BYTE_ARRAY, but Iceberg expects INT64 for small-precision decimals. The new manual pq.write_table() calls in the streaming path were missing the flag to force integer encoding.

Fix: Included store_decimal_as_integer=True in all pq.write_table() calls.

2. Partitioned tables failing on streamed writes

While fixing the above, I noticed the streaming path used table.add_files() for all tables. This breaks for partitioned tables, add_files requires each file to contain data for exactly one partition value, but our batches can span multiple partitions.

Fix: Split into two strategies:

• Unpartitioned: manual parquet write + add_files (single atomic commit)
• Partitioned: txn.append() per batch, letting pyiceberg handle partition-aware file layout

3. SIGILL crash during upsert on partitioned tables

After fixing the decimal encoding, I re-ran the tests locally. The tests that were failing earlier now got past the load step but the upsert test (test_merge_on_keys_in_schema_nested_hints) crashed with SIGILL (exit code 132). This was masked by Bug 1. The root cause is a pyiceberg 0.9.x bug: mixing txn.overwrite() + txn.append() in the same transaction triggers a native code crash.

Fix: Separated updates (overwrite) and inserts (append) into two transactions for partitioned tables. This trades full atomicity for not crashing, acceptable since dlt pipelines are idempotent and a re-run recovers any missing inserts. Can be collapsed back into one transaction once pyiceberg fixes this upstream.

Cleanup

• Extracted _upload_parquet_to_remote() to deduplicate the temp-file-write-and-upload logic shared by streamed writes and upserts
• Extracted _process_upsert_batch() to flatten the deeply nested upsert loop

@ajinzrathod I reviewed the code again. the fundamental change - to stream data via batch reader - makes total sense. but it seems to me that if we bump pyiceberg to 0.10 we won't need most of the custom upsert code here. pyiceberg implementation significantly improved and is also exposed in a way that allows to stream several batches in one transaction. I include implementation plan in next comment - I think it explains best what I think we should do here... pls let me know if you want to go with it, It is not a prob for us to take over - the goal is pretty clear now and mostly a matter of executing the plan.

Review & Suggested Simplification

Thanks for the PR — it solves a real OOM problem and the streaming approach is the right direction. After investigation, we believe the implementation can be significantly simplified by bumping the pyiceberg minimum to >=0.10.0.

Context

IcebergLoadFilesystemJob calls self.arrow_dataset.to_table() which loads all parquet files into memory before writing to Iceberg. For large datasets this is fatal. The fix is: stream batches from the arrow dataset and pass them one at a time to pyiceberg's transaction API.

With pyiceberg >=0.10.0, txn.append() and txn.upsert() both write files to storage immediately (via _dataframe_to_data_files() → write_file() → pq.ParquetWriter) — only one batch + one parquet file in memory at a time, with lightweight DataFile metadata accumulating in the transaction. This means most of the manual parquet writing and internal API usage in this PR is unnecessary.

Key findings

txn.upsert() available in 0.10+: pyiceberg 0.10.0 moved upsert from Table to Transaction (PR #1817). The old code on devel already has the TODO: "replace the batching method with transaction with pyiceberg's release after 0.9.1". With txn.upsert(), the entire _process_upsert_batch reimplementation using internal APIs (upsert_util, expression_to_pyarrow, bind) becomes unnecessary.

txn.upsert() is better than the PR's reimplementation:

• PR uses table.scan().to_arrow() (committed data only) — inserts from batch N are invisible to batch N+1, risking duplicate inserts across batches
• txn.upsert() uses DataScan(table_metadata=self.table_metadata).to_arrow_batch_reader() — streams matched rows and sees uncommitted changes from previous batches within the same transaction
• PR uses internal pyiceberg APIs; txn.upsert() is stable public API

SIGILL crash doesn't reproduce: We tested the single-transaction approach (mixing txn.overwrite() + txn.append()) on both pyiceberg 0.9.1 and 0.10.0 — all tests pass. The two-transaction workaround for partitioned tables is unnecessary.

store_decimal_as_integer is native in 0.10+: The monkey-patch (lines 47-58) can be removed.

Memory is constant: txn.append() and txn.upsert() write parquet files immediately to storage via pyiceberg's IO. The transaction only accumulates lightweight file metadata. Per-batch memory is bounded by batch size.

Proposed changes

1. Bump pyiceberg minimum to >=0.10.0

Enables txn.upsert() and native store_decimal_as_integer=True.

2. Rewrite write_iceberg_table — stream via txn.append()

Accept Union[pa.Table, pa.RecordBatchReader]. For RecordBatchReader, iterate batches in a single transaction:

def write_iceberg_table(table, data, write_disposition):
    if isinstance(data, pa.RecordBatchReader):
        with table.transaction() as txn:
            if write_disposition == "replace" and table.current_snapshot():
                txn.delete(delete_filter=AlwaysTrue())
            for batch in data:
                batch_tbl = ensure_iceberg_compatible_arrow_data(pa.Table.from_batches([batch]))
                txn.append(batch_tbl)
    else:
        # pa.Table path — backward compatible
        ...

Delete: _write_iceberg_table_streamed, _write_streamed_partitioned, _write_streamed_add_files, _upload_parquet_to_remote, _UPLOAD_CHUNK_SIZE.

3. Rewrite merge_iceberg_table — stream via txn.upsert()

Iterate batches, call txn.upsert() per batch in a single transaction:

batches = (
    data if isinstance(data, pa.RecordBatchReader)
    else data.to_batches(max_chunksize=1_000)
)
with table.transaction() as txn:
    for batch in batches:
        batch_tbl = ensure_iceberg_compatible_arrow_data(pa.Table.from_batches([batch]))
        txn.upsert(
            df=batch_tbl, join_cols=join_cols,
            when_matched_update_all=strategy == "upsert",
            when_not_matched_insert_all=True, case_sensitive=True,
        )

Delete: _upsert_iceberg_table, _process_upsert_batch.

4. Simplify IcebergLoadFilesystemJob.run() — stream from arrow dataset

Keep reading schema from parquet header. Use the same scanner approach for all paths:

source_ds = self.arrow_dataset
with source_ds.scanner(
    batch_readahead=0, fragment_readahead=0, use_threads=False
).to_reader() as arrow_rbr:
    if self._load_table["write_disposition"] == "merge" and table is not None:
        merge_iceberg_table(table=table, data=arrow_rbr, ...)
    else:
        write_iceberg_table(table=table, data=arrow_rbr, ...)
del source_ds

Delete: _iter_parquet_batches static method.

5. Remove store_decimal_as_integer monkey-patch

Delete the if pyiceberg_semver < Version("0.10.0"): block. Native in 0.10+.

6. Remove recommended_file_size = 128MB

This is a global behavioral change affecting all filesystem users (jsonl, parquet, delta, iceberg). It belongs in a separate PR if desired.

7. Remove iceberg_gc_collect_interval config

Not needed — pyiceberg handles file writing and memory management internally via txn.append()/txn.upsert(). The scanner with readahead=0 already ensures constant memory on the read side.

What gets preserved from the PR

• Streaming from arrow dataset via RecordBatchReader instead of .to_table() (the core fix)
• Schema from parquet header via pq.read_schema()
• write_iceberg_table() and merge_iceberg_table() accepting Union[pa.Table, RecordBatchReader]
• Tests (adjusted)

What gets deleted

Thanks for the detailed review! Regarding the SIGILL crash. I investigated further and it's an Apple Silicon specific issue in pyiceberg's native code.

Here's what I did:

I reverted the two-transaction workaround back to a single transaction (mixing txn.overwrite() + txn.append() in one transaction) and ran the test that was crashing:

ACTIVE_DESTINATIONS='["filesystem"]' ALL_FILESYSTEM_DRIVERS='["memory", "file"]' \
uv run pytest "tests/load/pipeline/test_merge_disposition.py::test_merge_on_keys_in_schema_nested_hints[upsert-filesystem-parquet-iceberg-no-staging-data]" -v

On my Mac (Apple Silicon), it crashes with SIGILL (exit code 132) — on both pyiceberg 0.9.1 and 0.10.0.

See the screenshot below:

I then ran the exact same test inside a Docker container on Linux x86_64. Passes on Linux.

The crash is specific to Apple Silicon, likely in pyiceberg-core's native Rust code which is compiled differently for ARM64 vs x86_64. The test uses partitioned table, which seems to trigger a code path in the native writer that causes this.

Since CI runs on Linux, the single-transaction approach works fine there. But Apple Silicon users on pyiceberg 0.9.1 or 0.10.0 would hit this crash on partitioned upserts with complex schemas.

Agreed on bumping to >=0.10.0 and simplifying. I've been testing this approach locally with large datasets (5GB, 11GB, 55GB in a 4 GB container). Have some findings around memory, snapshot behavior, and a schema evolution edge case. Will share details shortly.

1. Scanner OOMs on large datasets

source_ds.scanner(batch_readahead=0, fragment_readahead=0, use_threads=False).to_reader() causes OOM (Killed) on large data. Tested with 11 GB in a 4 GB container. The PyArrow dataset scanner buffers row groups internally in its C++ memory pool regardless of batch_readahead=0.

• gc.collect() and malloc_trim(0) don't help, it's Arrow's internal C++ allocation, not Python memory.

The scanner works fine on small data, our earlier tests with LIMIT_FILES=3 (~1.8 GB) completed successfully. It only OOMs when data exceeds the container's memory limit. The scanner's internal buffering accumulates over time; small data finishes before it becomes a problem.

Fix: pq.ParquetFile.iter_batches(batch_size=10_000), reads one row group at a time with explicit memory control. Only Parquet files are used in the Iceberg path, so pq.ParquetFile is always valid.

def _iter_batches():
    for file_path in self.file_paths:
        pf = pq.ParquetFile(file_path)
        for batch in pf.iter_batches(batch_size=10_000):
            yield batch
        del pf
batches_gen = _iter_batches()
first_batch = next(batches_gen)
arrow_rbr = pa.RecordBatchReader.from_batches(
    first_batch.schema,
    itertools.chain([first_batch], batches_gen),
)
del first_batch

Streaming via pq.ParquetFile.iter_batches(batch_size=10_000) + txn.append()/txn.upsert() per batch solves the OOM problem. This produces multiple snapshots (not a single snapshot per operation), but the transaction commit is atomic. All snapshots apply together, or none do. No partial state is visible to readers.

2. Schema evolution breaks txn.upsert() scan

After table.update_schema(), the snapshot still references the old schema_id. txn.upsert() internally scans using use_ref() which resolves to the snapshot's schema_id, so it returns fewer columns than expected → ValueError: Target schema's field names are not matching.

How it happens:

1. table.update_schema() updates the table's metadata to point to a new schema (e.g., schema_id=2), but the latest snapshot still references the old schema (schema_id=1).
2. When txn.upsert() scans, it resolves the schema from the snapshot (schema_id=1, say 10 columns), but the incoming data has 11 columns.
3. The scan returns 10-column rows, upsert expects 11 → ValueError.

This affects any dataset size, it's a metadata issue, not a scale issue. txn.append() is unaffected because it only writes and never scans existing data.

Workaround: Append an empty table after schema evolution to force a new snapshot with the updated schema_id.

N.B: Using pq.ParquetFile.iter_batches approach, txn.upsert() also eliminates duplicate inserts across batches, unlike table.scan().to_arrow(), it sees uncommitted data from earlier batches within the same transaction.

3. Trade-off in replace operation: txn.append() files are deletable but O(N²), fast_append() is O(N) but files can't be deleted

For streaming replace, we need to delete existing data first, then append new batches. We tested txn.delete(AlwaysTrue()) + fast_append().append_data_file(). O(N) performance, constant memory.

Problem: Files written by fast_append().append_data_file() + _dataframe_to_data_files() are not deleted by table.delete(AlwaysTrue()) or txn.delete(AlwaysTrue()). On a clean table with 20 files, only 1–2 files get deleted instead of all 20.

What we verified:

• table.delete(AlwaysTrue()) on files written by txn.append() deleted all entries (60 → 0) — the delete mechanism itself works.
• _StrictMetricsEvaluator returns the correct value for AlwaysTrue() — verified.
• Snapshot refs and snapshot_by_name("main") are correct — verified.
• The difference is in how files are registered in the manifest. txn.append() uses pyiceberg's full internal pipeline (_append_snapshot_producer → _FastAppendFiles → commit), while fast_append().append_data_file() manually registers DataFile objects from _dataframe_to_data_files(). The DataFile objects appear identical, but something in the manifest entry metadata (likely partition summaries, statistics format, or field IDs) causes _DeleteFiles._compute_deletes to classify them as "not relevant" and skip them.

Using txn.delete(AlwaysTrue()) + txn.append() per batch in a single transaction. Replace is correct and atomic, but subject to the O(N²) performance issue below.

4. txn.append() and txn.upsert() have O(N²) performance on large datasets

Transaction.table_metadata is a property that replays all accumulated _updates with model_copy(deep=True) on every access (~15 times per txn.append() call). Each txn.append() adds 2 updates. After N appends: N × 15 × replay(2N) = O(N²).

For small datasets (<1 GB, ~20 batches) it's fine. For 55 GB (~5500 batches at batch_size=10_000), the append load step ran for 57 minutes and 34 seconds.

Upsert is even worse, txn.upsert() scans existing data on every call in addition to the O(N²) metadata replay. Upserting 10 files (~3 GB, 500,000 rows) against 55 GB of existing data has been running for over 2 hours and is still not complete.

Summary:

FWIW, it would be valuable to validate append, replace, and merge operations on datasets larger than 10 GB (with container RAM smaller than the data size) to confirm the streaming approach at scale. I tested the proposed approach on Apple Silicon using a 4 GB Docker container with datasets ranging from 2 GB to 55 GB.

The test test_merge_on_keys_in_schema_nested_hints[upsert-filesystem-parquet-iceberg-no-staging-data] crashes with SIGILL (exit code 132) on Apple Silicon (ARM64). This happens with txn.upsert() on partitioned tables. The crash is in pyiceberg-core's Rust extension. It reproduces regardless of the read approach (pq.ParquetFile.iter_batches() or scanner). This is a pyiceberg-core bug and does not reproduce on Linux x86_64 (CI should pass).

I'm not pushing code to this repo yet, will do after we have proper sign-off on the approach. My current working code is on my personal fork in the feat/3752-v2-iceberg-streaming-atomic-commit branch for reference against the findings above.

I think it explains best what I think we should do here... pls let me know if you want to go with it, It is not a prob for us to take over

Sounds good, happy to look at the implementation plan. Given the findings above (especially the O(N²) txn.append(), performance, SIGILL, and the fast_append() delete issue), interested to hear your thoughts. Open to collaborating on this or letting you take it forward, whatever works best.

Really looking forward to this — we've been hitting OOMs on Iceberg loads as well, so a streaming write path would be a huge win for us.

Any rough estimate when this could land in a released version of dlt?

@tderk I'm afraid we won't get far here. pyiceberg has a few problems that prevent from moving forward:

• no schema evolution on upsert without any workaround
• problems with memory usage - still waiting for merge
• a whole set of problem with writing paritioned data

ducklabs released pretty good support for iceberg via iceberg extensions. I'd probably spend time there. it requires REST catalog though so this nice catalog-less iceberg we have here will be hard to reproduce

Thanks @rudolfix — that makes sense. On the partitioned-data problem specifically: would Table.add_files() be a way around it, or am I missing something?

From what I understand it just registers already-written Parquet via footer reads — no data rewrite, no per-partition writer buffers — so wouldn't that sidestep the memory/SIGILL issues that come from pyiceberg actually writing the partitioned data?

The only limit I see is that it's per-file (each Parquet would need to hold a single partition value), not per-load. Since dlt stages Parquet rotated by size, would the files just need to be split by partition key before registering — or is there a reason that wouldn't work here?

Really looking forward on this, if you need any help also