Core Utilities

iceberg_ops.core.utils

General-purpose data utilities for PyIceberg.

Functions

create_arrow_table(schema, rows)

Build a PyArrow Table from a list of dicts using an explicit schema.

upsert_streaming(table, source, *, join_cols, batch_rows=50_000)

Apply table.upsert() one bounded chunk at a time over a streaming source of PyArrow record batches.

rollback_to_snapshot(table, snapshot_id)

Restore a table to an earlier Iceberg snapshot.

get_snapshots(table)

Return all snapshot metadata as a list of dicts.

print_time_travel(table)

Print every historical snapshot with its full data.

trace_field_history(table, field_id)

Show how a single field evolved across all schema versions.

iceberg_ops.core.utils.create_arrow_table(schema, rows)[source]

Build a PyArrow Table from a list of row dicts using an explicit schema.

Missing keys in a row dict are filled with None (null), so partial rows match the full table schema without raising KeyError.

Parameters:

  • schema (Schema) – The Arrow schema defining column names and types.

  • rows (list[dict]) – Each dict represents one row. Unknown keys are ignored.

Return type:

Table

Example

schema = pa.schema([pa.field("id", pa.int64()), pa.field("v", pa.float64())])
tbl    = create_arrow_table(schema, [{"id": 1, "v": 3.14}, {"id": 2}])
iceberg_ops.core.utils.upsert_streaming(table, source, *, join_cols, batch_rows=50000)[source]

Apply table.upsert() one bounded chunk at a time.

PyIceberg’s pyiceberg.table.Table.upsert() requires a fully materialised pyarrow.Table. For pipelines whose incoming rowset is too large to hold in memory, this helper streams the source in record batches and flushes a chunk to upsert once at least batch_rows rows have been buffered.

Memory: batch_rows bounds only the incoming-row buffer (roughly batch_rows x column count x column width; any leftover after the loop is flushed once at the end). Each upsert also loads the destination rows that match the chunk’s keys, so peak memory grows with the matched destination data as well.

Each chunk is one catalog commit, which creates up to three Iceberg snapshots (delete or overwrite of the matched rows, the rewritten rows, the new rows). A composite-key chunk of more than 1,000 rows is upserted in sub-batches inside that one commit (apache/iceberg-python#3508), with up to three snapshots per sub-batch. Pick a larger batch_rows for fewer commits, smaller for a smaller buffer.

Duplicate keys: two rows with the same key in one chunk raise ValueError (earlier chunks stay committed); across chunks the later row wins.

If the table’s current snapshot was written under an older schema (for example right after add_column), one empty append is committed before the first chunk so the upsert can run - a workaround for apache/iceberg-python#2467.

Parameters:

  • table (Table) – Destination Iceberg table.

  • source (Iterable[RecordBatch]) – Streaming source of incoming rows. Accepts either a pyarrow.RecordBatchReader or any Python iterator / generator that yields pyarrow.RecordBatch.

  • join_cols (list[str]) – Equality columns for the upsert. Forwarded verbatim to table.upsert(join_cols=...).

  • batch_rows (int, default: 50000) – Buffer at least this many rows before each upsert call. Default 50 000. Must be a positive integer.

Returns:

Total rows passed to table.upsert(). Equals sum(b.num_rows for b in source) — note this counts incoming rows, not the number of distinct keys actually inserted vs updated.

Return type:

int

Raises:

  • ValueError – If batch_rows is not a positive integer, or a chunk holds duplicate keys.

  • ImportError – If the table’s partition spec uses a non-identity transform and pyiceberg-core is not installed (pip install     'iceberg-ops[transforms]'). Raised before the source is read.

Example

import pyarrow.parquet as pq

pf = pq.ParquetFile("incoming.parquet")
n  = upsert_streaming(
    table,
    pf.iter_batches(batch_size=10_000),
    join_cols=["id"],
    batch_rows=50_000,
)

Notes

All batches yielded by source must share a schema compatible with the destination table; pyarrow.Table.from_batches() enforces this and raises on mismatch.

iceberg_ops.core.utils.rollback_to_snapshot(table, snapshot_id)[source]

Roll back table to the state it was in at snapshot_id.

Creates a new snapshot that points to the data files of the target historical snapshot. All snapshots created after the rollback point remain in the metadata log (for audit) but are no longer the current state.

Parameters:

  • table (Table) – The Iceberg table to roll back.

  • snapshot_id (int) – The snapshot ID to restore. Obtain valid IDs via get_snapshots().

Raises:

ValueError – If snapshot_id does not exist in this table’s snapshot history.

Return type:

None

Example

snaps = get_snapshots(table)
rollback_to_snapshot(table, snaps[0]["snapshot_id"])
iceberg_ops.core.utils.get_snapshots(table)[source]

Return all snapshot metadata for table as a list of dicts.

Each dict contains:

snapshot_idint

Unique snapshot identifier.

timestampstr

UTC datetime string in ISO 8601 form.

schema_idint

Iceberg schema ID active at this snapshot.

operationstr

e.g. "append", "overwrite", "delete".

Example

for snap in get_snapshots(table):
    print(snap["snapshot_id"], snap["timestamp"], snap["operation"])
Parameters:

table (Table)

Return type:

list[dict[str, Any]]

iceberg_ops.core.utils.print_time_travel(table)[source]

Print every historical snapshot of table with its full data.

Iterates all snapshots in chronological order, querying each one independently via PyIceberg’s snapshot scan. Demonstrates that data from every past schema version is still accessible even after schema evolution.

Parameters:

table (Table) – The Iceberg table to inspect.

Return type:

None

iceberg_ops.core.utils.trace_field_history(table, field_id)[source]

Trace how a single field evolved across all recorded schema versions.

Iceberg assigns every field a stable integer ID that persists through renames and type changes. This function leverages that ID to show the complete lifecycle of a column — from its creation, through any renames, to its archival as a legacy column.

Parameters:

  • table (Table) – The Iceberg table whose schema history to inspect.

  • field_id (int) – The stable Iceberg field ID to trace. Use table.schema().fields to find field IDs in the current schema.

Return type:

None

Example

for f in table.schema().fields:
    print(f.field_id, f.name)

trace_field_history(table, field_id=3)