Data Migration¶

Note

Experimental. The data migrator is a companion to the schema migration workflow, with documented limits: the destination table must already exist; overwrite_partition needs an identity-partitioned destination and a migration that fits in one commit; batch_rows bounds the rows in each commit, not peak memory; and the sample verify method can report a false mismatch on larger tables. The API may change in any release. See Cross-Table Data Migration for details.

The iceberg_ops.data_migration package re-exports its public classes and functions from the three submodules below. Tests and downstream code can import the names from either the package (from iceberg_ops.data_migration import DataMigrator) or the submodule – both bindings refer to the same object.

Package¶

iceberg_ops.data_migration¶

Cross-table, schema-reconciling, streaming Iceberg-to-Iceberg data migration.

This subpackage is the logical data migrator - moving rows between Iceberg tables with optional schema reconciliation, type coercion, and column renames - distinct from the schema migrations driven by iceberg_ops.runner.

Public API¶

SchemaReconciler: Declarative description of how to map source columns to dest columns, including renames, type coercions, and missing-column policies.
ReconciliationPlan: The structured plan returned by SchemaReconciler.compute_plan(). Lists per-column projections plus any auto-add columns and warnings.
ColumnProjection: One entry of a plan: how to derive a single dest column from the source batch (rename + cast, or null-fill).
CheckpointStore: JSON-file-backed checkpoint that records per-batch progress so a crashed migration can resume from the last committed batch.
CheckpointError: A RuntimeError raised when a checkpoint file is corrupt, unreadable, or has an unsupported version, and when DataMigrator refuses to resume one (completed, started with different settings, its source snapshot is gone, or the destination no longer holds its checkpointed batches).
DataMigrator: Streaming, schema-reconciling, resumable migrator. Reads from a source Iceberg table, projects through SchemaReconciler, and writes to a destination Iceberg table - appending, upserting, or partition-overwriting per the configured write_mode.
MigrationResult: Outcome of DataMigrator.run(): rows moved, bytes moved, snapshot list, schema changes, duration, dry-run estimates.
PartitionStrategy: How the migrator treats the destination partition spec. Only PRESERVE is implemented today.
VerificationResult: Outcome of DataMigrator.verify(): structured report of whether the destination matches the source under the chosen method (row_count / partition_hash / sample).

CLI surface¶

iceberg-ops migrate-data exposes the migrator from the shell.

Schema Reconciler¶

iceberg_ops.data_migration.reconciler¶

Schema reconciliation primitive for the cross-table data migrator.

A SchemaReconciler declares how to map a source Iceberg schema to a destination Iceberg schema - column renames, type coercions, and policies for columns missing from one side. It produces a ReconciliationPlan with a per-dest-column ColumnProjection, and projects individual pyarrow.RecordBatch instances against that plan in O(rows x columns) memory.

The reconciler is intentionally just schema-level: it does not read from the source table or write to the destination table. The streaming run loop and the snapshot-pinned read live in iceberg_ops.data_migration.migrator.

Workflow¶

Caller resolves source and destination schemas (Arrow or Iceberg).
plan = reconciler.compute_plan(source_schema, dest_schema) - once.
For every batch off source.scan(...).to_arrow_batch_reader(): out = reconciler.project_batch(batch, plan).
Caller writes out to the destination table via Table.append() / Table.upsert() / Table.dynamic_partition_overwrite().

Memory bound: project_batch allocates O(batch.num_rows x plan_columns) via pyarrow.compute.cast() and pyarrow.nulls() - independent of total source / destination size.

class iceberg_ops.data_migration.reconciler.ColumnProjection(dest_name, source_name, dest_arrow_type, fill_with_null=False)[source]¶

Bases: object

One column in the reconciled output.

Each ColumnProjection describes how SchemaReconciler.project_batch() derives a single destination column from a source pyarrow.RecordBatch.

Variables:

dest_name (str) – Column name in the destination schema.
source_name (str or None) – Column name in the source schema. None when fill_with_null is True (i.e., the source has no matching column and the reconciler’s policy is "null").
dest_arrow_type (pyarrow.DataType) – Cast target. The source column is cast to this type via pyarrow.compute.cast(); null-fill columns are allocated directly in this type.
fill_with_null (bool) – True when the column is missing from the source and the on_missing_in_source policy is "null". The projector emits pa.nulls(num_rows, type=dest_arrow_type).

Parameters:

dest_name (str)
source_name (str | None)
dest_arrow_type (DataType)
fill_with_null (bool, default: False)

dest_name: str¶

source_name: str | None¶

dest_arrow_type: DataType¶

fill_with_null: bool = False¶

class iceberg_ops.data_migration.reconciler.ReconciliationPlan(projections=list(), auto_add_columns=list(), warnings=list(), dest_arrow_schema=<factory>)[source]¶

Bases: object

Plan returned by SchemaReconciler.compute_plan().

The plan is the entire static description of how to project source batches into destination-shape batches. It is computed once during pre-flight and then used by every batch in the streaming loop.

Variables:

projections (list of ColumnProjection) – One entry per output column, in destination order. Source columns covered by no projection are dropped from the output (subject to the on_missing_in_dest policy at plan time).
auto_add_columns (list of (str, IcebergType)) – Columns the source has that the destination does not, when on_missing_in_dest="auto_add". The data migrator is responsible for adding these to the destination table before invoking SchemaReconciler.project_batch().
warnings (list of str) – Non-fatal issues surfaced during planning - e.g. doc-only differences, columns covered only by null-fill, type coercions that widen lossily.
dest_arrow_schema (pyarrow.Schema) – The full destination Arrow schema (after auto-adds). Useful for asserting that the projected batches match what the writer expects.

Parameters:

projections (list[ColumnProjection], default: list())
auto_add_columns (list[tuple[str, IcebergType]], default: list())
warnings (list[str], default: list())
dest_arrow_schema (Schema, default: <factory>)

projections: list[ColumnProjection]¶

auto_add_columns: list[tuple[str, IcebergType]]¶

warnings: list[str]¶

dest_arrow_schema: Schema¶

class iceberg_ops.data_migration.reconciler.SchemaReconciler(column_mapping=None, type_coercions=None, on_missing_in_source='error', on_missing_in_dest='error')[source]¶

Bases: object

Declarative source -> destination schema reconciliation policy.

A reconciler is a plan factory: it does not hold per-batch state. Construct one, call compute_plan() once, then reuse the resulting ReconciliationPlan for every batch in the streaming migration loop.

Parameters:

column_mapping (dict[str, str] | None, default: None) – Source-name to destination-name renames. {"order_total": "total_amount"} means the source column order_total lands in the destination as total_amount. Columns not in this mapping are matched by exact name. The mapping is one-to-one; many-to-one or one-to-many fan-out belongs in a custom transformation layer above the reconciler. Every key must name a source column; compute_plan reports any other key as an error.
type_coercions (dict[str, IcebergType] | None, default: None) – Per-destination-column override of the cast target. Keys are destination column names. Set this when the source’s natural type differs from what you want at rest (e.g. "qty": LongType() widens an int32 source to int64 at the destination). The override applies whether the destination already has the column or it is being auto-added. For a column the destination already has, the override must match that column’s type or be a legal Iceberg promotion of it; otherwise the write fails at the first batch, not at plan time.
on_missing_in_source (Literal['error', 'skip', 'null'], default: 'error') – Policy when a destination column has no matching source column (after mapping). "error" raises during compute_plan; "null" emits a column of nulls in the destination type; "skip" drops the column from the projected output (the downstream writer must accept a partial schema, or the destination column must allow nulls and the writer must fill them). With DataMigrator upserts, "skip" is rejected at pre-flight (an upsert must write every destination column), and "null" overwrites the existing values of those columns with null for every matched key - the plan carries a warning saying so.
on_missing_in_dest (Literal['error', 'auto_add'], default: 'error') – Policy when a source column has no matching destination column (after mapping). "error" raises during compute_plan; "auto_add" records the column in ReconciliationPlan.auto_add_columns. The data migrator is then expected to extend the destination table before writing.

Examples

Same-schema migration (the trivial case):

>>> reconciler = SchemaReconciler()
>>> plan = reconciler.compute_plan(src.schema(), dst.schema())
>>> for batch in src.scan().to_arrow_batch_reader():
...     out = reconciler.project_batch(batch, plan)
...     dst.append(pa.Table.from_batches([out]))

Cross-schema migration with rename, type widening, and a missing column filled with nulls:

>>> reconciler = SchemaReconciler(
...     column_mapping={"order_total": "total_amount"},
...     type_coercions={"qty": ice_types.LongType()},
...     on_missing_in_source="null",
...     on_missing_in_dest="error",
... )

Notes

The reconciler does not validate semantic compatibility beyond what pyarrow.compute.cast() enforces at projection time. Lossy casts (e.g. string -> int64 with non-numeric values) raise inside project_batch rather than during compute_plan; this is by design - many casts are data-dependent and cannot be decided statically.

column_mapping: dict[str, str] | None = None¶

type_coercions: dict[str, IcebergType] | None = None¶

on_missing_in_source: Literal['error', 'skip', 'null'] = 'error'¶

on_missing_in_dest: Literal['error', 'auto_add'] = 'error'¶

compute_plan(source_schema, dest_schema)[source]¶

Compute the plan that turns source_schema into dest_schema.

The plan is computed once per migration. It validates the reconciler’s policies eagerly and lists every per-column decision in ReconciliationPlan.projections.

Parameters:

source_schema (Schema | Schema) – Schema of the source Iceberg table.
dest_schema (Schema | Schema) – Schema of the destination Iceberg table - before any auto_add columns are applied.

Return type:

ReconciliationPlan

Raises:

ValueError – If on_missing_in_source="error" and at least one destination column has no source counterpart, or if on_missing_in_dest="error" and at least one source column has no destination counterpart. All such issues are collected and reported in a single error message.
TypeError – For unsupported schema kinds (neither pa.Schema nor IcebergSchema).

project_batch(batch, plan)[source]¶

Project a single source batch according to plan.

Each column in ReconciliationPlan.projections becomes one column in the returned batch. Source columns referenced by the plan are pulled by name and cast to the target type via pyarrow.compute.cast(). fill_with_null=True projections are realised as pa.nulls(num_rows, type=dest_arrow_type).

Parameters:

batch (RecordBatch) – One batch from the source’s scan(...).to_arrow_batch_reader().
plan (ReconciliationPlan) – The plan from compute_plan(). Must be produced by a compute_plan call whose source_schema was a superset of batch.schema; otherwise the column lookup below raises KeyError.

Returns:

A batch whose schema matches plan.dest_arrow_schema. num_rows equals batch.num_rows.

Return type:

RecordBatch

Raises:

KeyError – If batch is missing a source column the plan references - typically a sign that compute_plan was called against a stale source_schema.
pyarrow.lib.ArrowInvalid – If a cast fails on the actual data (e.g. string -> int with non-numeric values). This bubbles up unchanged.

Checkpoint Store¶

iceberg_ops.data_migration.checkpoint¶

Resumable-migration bookkeeping for the cross-table data migrator.

A CheckpointStore persists per-batch progress to a JSON file on local disk. The streaming DataMigrator writes one entry per committed batch; on restart, the migrator skips batches whose index is at or below last_committed_batch and picks up where the previous run died.

File format (version 1)¶

{
  "version": 1,
  "source_snapshot_id": 12345,
  "last_committed_batch": 42,
  "dest_snapshot_ids": [98765, 98766],
  "completed": false,
  "started_at": "2026-05-05T10:30:00Z",
  "job": {"batch_rows": 50000, "write_mode": "append", "...": "..."}
}

Fields:

version: File-format version. Currently 1. The store refuses to load a file with any other version - bump this and add migration logic when the format evolves.
source_snapshot_id: Iceberg snapshot ID of the source table the migration reads (or null if the source had no snapshot). On resume, DataMigrator reads the source at this snapshot, so the resumed batches line up with the ones already committed.
last_committed_batch: Index of the last successfully committed batch, or null if nothing has been committed yet.
dest_snapshot_ids: Every destination snapshot the committed batches created, in commit order (one batch can create several snapshots). Useful for post-mortems.
completed: true once CheckpointStore.mark_completed() has been called. The file remains on disk after completion so the user can audit the snapshot list; CheckpointStore.reset() removes it. DataMigrator refuses to reuse a completed checkpoint.
started_at: ISO 8601 UTC timestamp of the most recent CheckpointStore.start() call. Updated on every fresh initialisation; not updated when the file is merely re-loaded by a resuming migrator.
job (optional): The identity of the migration job that started the checkpoint: source and destination table UUIDs, row filter, batch_rows, write mode, upsert keys and reconciler settings, plus a random job ID and the destination snapshot that was current before the job wrote anything. null (or absent) when the caller did not record one. DataMigrator refuses to resume a checkpoint whose job differs from the current settings, or that has no job.

Resume rules¶

DataMigrator resumes from an existing checkpoint file only when all of these hold; otherwise it raises CheckpointError before writing anything:

the checkpoint is not completed;
it carries a job that matches the current settings;
the stored source_snapshot_id still exists in the source table and agrees with any explicit source_snapshot;
the destination’s current history still holds the snapshots of the checkpointed batches (a rollback of the destination would otherwise make the resume skip rows that are no longer there).

CheckpointStore.reset() forgets which batches are in the destination: roll the destination back to the job’s dest_snapshot_before (or delete those rows) before you reset and re-run, or an append writes them again.

Atomicity¶

Every write goes through <path>.tmp, which is flushed and fsync-ed before os.replace() moves it over the live file. The replace is atomic on POSIX and on Windows when source and destination are on the same filesystem (which is the only case we support). On POSIX the parent directory is fsync-ed as well so the rename itself survives a power loss.

Concurrency¶

Single-writer only. Two DataMigrator runs pointed at the same checkpoint will overwrite each other’s progress with no interlocking. Use one checkpoint file per migration job.

exception iceberg_ops.data_migration.checkpoint.CheckpointError[source]¶

Bases: RuntimeError

Raised when a checkpoint file cannot be used.

Covers a corrupt, unreadable or unsupported-version file, and the resume refusals of DataMigrator (completed checkpoint, job mismatch, missing source snapshot, checkpointed batches no longer in the destination).

The message identifies the offending path and suggests a remedy (typically: delete the file or call CheckpointStore.reset()). When batches were already committed, a resume refusal also names the destination snapshot to roll back to first, since a re-run after reset() writes them again. Subclasses RuntimeError so callers that catch runtime failures also catch it.

class iceberg_ops.data_migration.checkpoint.CheckpointStore(path)[source]¶

Bases: object

JSON-file-backed checkpoint for resumable data migrations.

The store is intentionally lightweight: the constructor does not touch disk. Call start() to initialise a fresh checkpoint, or call any accessor (last_committed() / is_committed() / dest_snapshot_ids) on an existing file to lazy-load it.

Parameters:: path (str | Path) – Filesystem path where the checkpoint JSON lives. Parent directory is created on first write if it does not exist.

Examples

Fresh migration:

store = CheckpointStore("/tmp/migrate.json")
store.start(source_snapshot_id=12345)
for batch_idx, batch in enumerate(reader):
    if store.is_committed(batch_idx):
        continue
    dest_snap = write_one_batch(batch)
    store.mark_committed(batch_idx, dest_snap)
store.mark_completed()

Resuming after a crash:

store = CheckpointStore("/tmp/migrate.json")
# No start() - file already exists; lazy-load on first use.
last = store.last_committed()  # e.g. 42
for batch_idx, batch in enumerate(reader):
    if store.is_committed(batch_idx):
        continue                # skips 0..42
    ...

start(source_snapshot_id=None, *, job=None)[source]¶

Initialise a fresh checkpoint, overwriting any previous file.

Use this for new migrations. To resume an existing migration, skip start() entirely - the store will lazy-load the existing JSON on the first accessor call.

Parameters:

source_snapshot_id (int | None, default: None) – Iceberg snapshot ID of the source table the migration reads. A resuming DataMigrator reads at this snapshot.
job (dict[str, Any] | None, default: None) – JSON-serialisable identity of the migration job (see the module docstring). A resuming DataMigrator compares it with its own settings and refuses on any difference.

Return type:

None

Notes

start() always overwrites - the caller is expected to have already decided “fresh run vs. resume”. The data migrator makes that decision with exists(): an existing file means resume, a missing one means a fresh run.

reset()[source]¶

Delete the checkpoint file and clear in-memory state.

Idempotent - safe to call when no file exists. After reset, the next mutator call lazy-creates a fresh state.

Return type:: None

mark_completed()[source]¶

Mark the migration as complete (success).

The file remains on disk after completion so the user can audit dest_snapshot_ids; a follow-up call to reset() removes it.

Return type:: None

is_committed(batch_idx)[source]¶

Return True if batch_idx was already committed.

Always False before any mark_committed() call.

Parameters:: batch_idx (int)
Return type:: bool

mark_committed(batch_idx, dest_snapshot_id)[source]¶

Record that batch_idx produced dest_snapshot_id and persist.

Lazily creates the checkpoint file if it does not exist yet - callers are not required to call start() first if they don’t care about source_snapshot_id.

Parameters:

batch_idx (int) – Zero-based batch index. Must be strictly greater than last_committed(); the streaming loop in DataMigrator commits in monotonically increasing order, and out-of-order calls indicate a bug.
dest_snapshot_id (int | list[int]) – Destination snapshot ID(s) the write of this batch created, in commit order. An upsert batch can create several snapshots, and a no-op upsert none (pass []). Every ID is appended to dest_snapshot_ids.

Raises:

ValueError – If batch_idx is not strictly greater than last_committed_batch.

Return type:

None

last_committed()[source]¶

Return the last committed batch index, or None if none.

Return type:: int | None

property path: Path¶: Filesystem path the checkpoint persists to.

exists()[source]¶

True if the checkpoint file is currently on disk.

Return type:: bool

property source_snapshot_id: int | None¶: Source snapshot recorded by the most recent start().

property job: dict[str, Any] | None¶

Job identity recorded by the most recent start().

Returns a copy, so mutating it does not change the store. None when the checkpoint was started without a job.

property started_at: str | None¶: ISO 8601 UTC timestamp of the most recent start() call.

property completed: bool¶: True once mark_completed() has flipped the flag.

property dest_snapshot_ids: list[int]¶: Snapshot IDs every committed batch produced, in commit order.

Migrator¶

iceberg_ops.data_migration.migrator¶

Streaming, schema-reconciling, resumable Iceberg-to-Iceberg data migration.

A DataMigrator reads from a source Iceberg table, projects each batch through a SchemaReconciler, and writes to a destination Iceberg table - appending, upserting, or partition- overwriting per write_mode. A CheckpointStore makes the loop resumable: a crashed migration restarts from the last successfully committed batch.

Algorithm¶

Pre-flight. Every check runs before the destination is touched:
- Refresh both tables and resolve the source snapshot: the checkpoint’s snapshot when resuming, else source_snapshot, else the source’s current snapshot.
- When resuming, validate the checkpoint (not completed, same job settings, source snapshot still present).
- Build a ReconciliationPlan from the source schema at that snapshot.
- Validate write-mode prerequisites (upsert_keys present in the projection, an upsert-compatible missing-column policy, the overwrite_partition limits below) and that pyiceberg-core is installed when the destination’s partition spec needs it.
Only then does pre-flight write, in this order:
- add any auto_add_columns to the destination schema (one schema commit, no snapshot);
- for write_mode="upsert" only, commit one empty append when the destination’s current snapshot predates its schema - a workaround for apache/iceberg-python#2467, without which the first upsert after a schema-only change fails;
- start the checkpoint file (fresh runs).
A dry run performs the checks but none of the writes.
Streaming loop:
- Open source.scan(row_filter, snapshot_id).to_arrow_batch_reader().
- Accumulate raw source batches up to batch_rows rows.
- Project each accumulated chunk through the reconciler.
- Write to dest via Table.append() / Table.upsert() / Table.dynamic_partition_overwrite() - one catalog commit per chunk.
- Mark the checkpoint after each write.
Post-flight: mark the checkpoint completed.

Memory¶

Each destination commit is bounded by batch_rows rows, but peak memory is not: PyIceberg’s batch reader decodes whole source data files at a time (apache/iceberg-python#2407), so peak memory grows with the number and size of the source data files. An upsert also loads the destination rows that match each chunk’s keys.

Write modes¶

append: One Table.append() per chunk.
upsert: One upsert keyed by upsert_keys per chunk. Two rows with the same key inside one chunk raise ValueError (earlier chunks stay committed); across chunks the later row wins, so the outcome for duplicate keys depends on batch_rows.
overwrite_partition: Table.dynamic_partition_overwrite(). In this release it needs a destination partitioned by identity transforms only, and the whole migration must fit in one commit (at most batch_rows source rows after the row filter): a second commit would delete the rows the first one wrote to the same partitions. Pre-flight rejects anything else - raise batch_rows or use append / upsert.

Resume¶

The checkpoint records the job’s identity (source and destination table UUIDs, row filter, batch_rows, write mode, upsert keys, reconciler settings) and the source snapshot. A resumed run reads at that snapshot and raises CheckpointError before writing if any setting changed, if the snapshot no longer exists, or if the checkpoint already records a completed migration (call reset() to run again).

Every destination snapshot a batch creates carries the job ID and batch index in its summary (iceberg-ops.migration-job / iceberg-ops.migration-batch); the empty schema-stamp snapshot an upsert may commit first carries only iceberg-ops.migration-job. On resume, a batch whose write landed just before a crash but was not yet checkpointed is found by these properties and recorded instead of being written twice. A resume also raises CheckpointError when a checkpointed batch’s snapshot is no longer in the destination’s current history (the destination was rolled back), because skipping that batch would lose its rows.

reset() forgets which batches are already in the destination. When batches were committed, every refusal says so and names the rollback target (the job’s dest_snapshot_before): roll the destination back before resetting, or an append re-run writes those rows again.

Verification¶

After DataMigrator.run(), call DataMigrator.verify() to confirm the destination matches the source. Three methods are supported - row_count (manifest record counts), partition_hash (thorough but materialises both sides), and sample (first-N rows; reliable only for small tables or an identical read order).

Out of scope (today)¶

PartitionStrategy.REPARTITION_BY_DEST_SPEC and FLATTEN - the only strategy supported today is PRESERVE: the destination’s own partition spec dictates the layout (PyIceberg’s writer applies it).
target_file_size_mb - controlled by the destination table’s write.target-file-size-bytes property; the migrator does not override it.
Non-identity partition transforms in verify() with method="partition_hash" (it raises NotImplementedError).

class iceberg_ops.data_migration.migrator.DataMigrator(source, dest, *, schema_reconciler=None, partition_strategy=PartitionStrategy.PRESERVE, row_filter=None, source_snapshot=None, batch_rows=50000, write_mode='append', upsert_keys=None, checkpoint=None, progress=None, dry_run=False)[source]¶

Bases: object

Streaming, schema-reconciling, resumable cross-table migrator.

See module docstring for the algorithm. The migrator is constructed with all configuration up-front; run() performs the migration.

Parameters:

source (Table) – Source Iceberg table.
dest (Table) – Destination Iceberg table. May be in a different catalog than source.
schema_reconciler (SchemaReconciler | None, default: None) – Declarative source -> dest schema mapping. Defaults to a strict reconciler (errors on any mismatch); passing one explicitly lets the user enable rename / null-fill / auto-add behaviour.
partition_strategy (PartitionStrategy, default: PartitionStrategy.PRESERVE) – Currently only PRESERVE is supported.
row_filter (str | BooleanExpression | None, default: None) – PyIceberg row filter applied to the source scan. Defaults to AlwaysTrue (no filter).
source_snapshot (int | None, default: None) – Pin the source read to a specific snapshot ID. Defaults to the source’s current snapshot when run() starts. The snapshot is recorded in the checkpoint; a resumed run reads at the recorded snapshot and refuses an explicit source_snapshot that differs from it.
batch_rows (int, default: 50000) – Target rows per destination commit. Higher -> fewer commits; lower -> more, smaller commits. Each batch is one catalog commit, which can create several snapshots (see MigrationResult.new_snapshots). It does not bound peak memory - see the module docstring. Default 50_000.
write_mode (Literal['append', 'upsert', 'overwrite_partition'], default: 'append') – How to write each chunk to the destination. upsert requires upsert_keys. overwrite_partition needs a destination partitioned by identity transforms only and a migration that fits in one commit (at most batch_rows source rows after the row filter); pre-flight rejects anything else.
upsert_keys (list[str] | None, default: None) – Required when write_mode="upsert" and rejected otherwise. Forwarded to Table.upsert() as join_cols; every key must be a column of the projected destination schema.
checkpoint (CheckpointStore | None, default: None) – Per-batch checkpoint for resumability. When None the migration is non-resumable (a crash means re-running from scratch). An existing checkpoint file is resumed only when it matches this migrator’s settings and is not completed; otherwise run() raises CheckpointError.
progress (Callable[[int, int | None], None] | None, default: None) – progress(rows_done, rows_total_or_none) called once after each successful commit. rows_total is None during live migration (not pre-counted) and the post-scan total during a dry-run.
dry_run (bool, default: False) – When True, scan the source through the reconciler but skip every destination write. Returns a MigrationResult populated with estimated_rows / estimated_bytes and an empty new_snapshots list. A dry run ignores the checkpoint.

Raises:

ValueError – If write_mode="upsert" without upsert_keys, if upsert_keys is given for another write mode or holds a blank or repeated name, if batch_rows <= 0, if the partition strategy is not PRESERVE, or if source and dest are the same table (same table UUID, even when loaded through different catalogs).

run()[source]¶

Execute the migration.

Return type:

MigrationResult

Raises:

CheckpointError – If an existing checkpoint records a completed migration, was started with different settings (or without a job fingerprint), reads a source snapshot that no longer exists or differs from source_snapshot, or records batches whose snapshots are no longer in the destination’s current history (the destination was rolled back). Raised before any destination write.
ValueError – If the reconciler cannot map the schemas, upsert_keys are not in the projection or have no source column (they would be filled with null), an upsert would drop destination columns (on_missing_in_source="skip"), or overwrite_partition limits are not met - all raised before any destination write. During streaming, an upsert batch that holds duplicate keys also raises it; earlier batches stay committed.
ImportError – If the destination’s partition spec needs pyiceberg-core and it is not installed (real runs only; a dry run reports it in plan.warnings).

verify(*, method='row_count', sample_size=100)[source]¶

Post-migration consistency check.

Three methods, ordered cheapest -> most thorough:

row_count (default): Count source rows (after the migrator’s row filter and snapshot pin) and destination rows. Compare. Uses the manifests’ per-file record counts; reads data files only where the row filter cannot be decided per file or delete files apply.
partition_hash: Group rows by destination partition value, sort each group’s rows by all columns, SHA256-hash, compare. Currently supports only identity-transform partition specs and unpartitioned tables; other specs raise NotImplementedError. Memory-intensive - materialises both sides in full.
sample: Take the first sample_size rows on each side, sort within the sample by all columns, hash, compare. Fast but only reliable when both tables fit within sample_size (degenerates to a full check) or when the read order is deterministic across both sides.

Parameters:

method (Literal['row_count', 'partition_hash', 'sample'], default: 'row_count') – Which verification to run.
sample_size (int, default: 100) – For method="sample" only. Default 100. Ignored for other methods.

Return type:

VerificationResult

Raises:

ValueError – If method is unknown, or sample_size <= 0 for method="sample".
NotImplementedError – For method="partition_hash" when the destination’s partition spec has a non-identity transform. It is a RuntimeError, so it is not reported as a mismatch.

Notes

The source is read at the explicit source_snapshot, else at the snapshot the last run() on this migrator read (including the checkpoint’s snapshot on a resumed run), else at the source’s current snapshot. Data appended to the source after the run therefore does not count as a mismatch.

Caveat¶

verify() compares source (filtered + snapshot- pinned per the migrator’s config) against dest (current state, no filter). Meaningful only when write_mode="append" into an empty destination, or when the user accounts for any pre-existing rows in their interpretation of the result.

class iceberg_ops.data_migration.migrator.MigrationResult(rows_moved=0, bytes_moved=0, new_snapshots=list(), schema_changes=list(), duration_seconds=0.0, checkpoint_path=None, dry_run=False, estimated_rows=None, estimated_bytes=None, plan=None, dest_snapshot_before=None, rows_inserted=None, rows_updated=None)[source]¶

Bases: object

Outcome of DataMigrator.run().

Variables:

rows_moved (int) – Number of rows actually written to the destination - i.e. excludes batches skipped because the checkpoint already had them. For dry-runs this is always 0; see estimated_rows. For upserts it counts incoming rows, not distinct keys; see rows_inserted / rows_updated.
bytes_moved (int) – Approximate bytes written, computed as sum(chunk.nbytes) across every chunk passed to the destination’s writer. The on-disk Parquet size will differ (compression, encoding) - this is an in-memory measure useful for progress reporting.
new_snapshots (list of int) – Every destination snapshot this run created, in commit order: the empty schema-stamp snapshot (upsert after a schema-only change, when needed), then every snapshot each batch’s commit created. One batch is one catalog commit but can create several snapshots - up to three for an upsert (three per sub-batch when a composite-key batch of more than 1,000 rows is split), two for overwrite_partition. Snapshots committed by other writers in the meantime are not included.
schema_changes (list of str) – Names of columns the migrator added to the destination via auto_add reconciliation.
duration_seconds (float) – Wall-clock elapsed time across pre-flight + streaming + post-flight.
checkpoint_path (pathlib.Path or None) – Filesystem path of the active CheckpointStore, or None if no checkpoint was used.
dry_run (bool) – True when this result describes a dry-run.
estimated_rows (int or None) – Total rows scanned (post-filter) during a dry-run. None for real runs.
estimated_bytes (int or None) – Total Arrow bytes scanned during a dry-run. None for real runs.
plan (ReconciliationPlan) – The plan the migrator used. Useful for post-mortem introspection (auto-add list, warnings).
dest_snapshot_before (int or None) – The destination’s current snapshot before the migration wrote anything - the target for rollback_to_snapshot() to undo it. For a resumed run it is the snapshot recorded when the job first started. None when the destination had no snapshot then, or for a dry-run.
rows_inserted (int or None) – Rows the upsert inserted, summed over this run’s batches. None unless write_mode="upsert" (and not a dry-run).
rows_updated (int or None) – Rows the upsert rewrote because a non-key value changed, summed over this run’s batches. None unless write_mode="upsert" (and not a dry-run).

Parameters:

rows_moved (int, default: 0)
bytes_moved (int, default: 0)
new_snapshots (list[int], default: list())
schema_changes (list[str], default: list())
duration_seconds (float, default: 0.0)
checkpoint_path (Path | None, default: None)
dry_run (bool, default: False)
estimated_rows (int | None, default: None)
estimated_bytes (int | None, default: None)
plan (ReconciliationPlan | None, default: None)
dest_snapshot_before (int | None, default: None)
rows_inserted (int | None, default: None)
rows_updated (int | None, default: None)

rows_moved: int = 0¶

bytes_moved: int = 0¶

new_snapshots: list[int]¶

schema_changes: list[str]¶

duration_seconds: float = 0.0¶

checkpoint_path: Path | None = None¶

dry_run: bool = False¶

estimated_rows: int | None = None¶

estimated_bytes: int | None = None¶

plan: ReconciliationPlan | None = None¶

dest_snapshot_before: int | None = None¶

rows_inserted: int | None = None¶

rows_updated: int | None = None¶

class iceberg_ops.data_migration.migrator.PartitionStrategy(*values)[source]¶

Bases: StrEnum

How the migrator should treat the destination partition spec.

Only PRESERVE is implemented today: the destination’s own spec dictates the layout. REPARTITION_BY_DEST_SPEC and FLATTEN are reserved for a future release.

PRESERVE = 'preserve'¶

class iceberg_ops.data_migration.migrator.VerificationResult(method, matches, source_count=None, dest_count=None, sample_size=None, mismatched_partitions=list(), details=list())[source]¶

Bases: object

Outcome of DataMigrator.verify().

A small structured report indicating whether the destination matches the source under the chosen verification method.

Variables:

method ("row_count" | "partition_hash" | "sample") – Method used to compute this result.
matches (bool) – True when source and destination agreed under the chosen method. For row_count this means equal counts; for partition_hash it means every partition’s content hash matched; for sample it means the sampled rows matched.
source_count (int or None) – Total source rows seen (after row filter + snapshot pin + reconciler projection). Populated for row_count and sample.
dest_count (int or None) – Total destination rows seen. Populated for row_count and sample.
sample_size (int or None) – For method="sample", the number of rows actually compared (capped at the smaller of source rows / dest rows / requested sample_size).
mismatched_partitions (list of str) – For method="partition_hash", the partition keys whose hashes differed (or that exist on only one side).
details (list of str) – Human-readable explanations for any mismatch. Empty when matches is True.
Caveat
------
source (All three methods compare the current state of)
against ((filtered + snapshot-pinned per the migrator's config))
When (the current state of dest (no filter).)
pre-existing (write_mode != "append" or the destination had)
- (rows, the source-vs-dest counts will not naturally agree)
accordingly. (interpret the result)

Parameters:

method (Literal['row_count', 'partition_hash', 'sample'])
matches (bool)
source_count (int | None, default: None)
dest_count (int | None, default: None)
sample_size (int | None, default: None)
mismatched_partitions (list[str], default: list())
details (list[str], default: list())

method: Literal['row_count', 'partition_hash', 'sample']¶

matches: bool¶

source_count: int | None = None¶

dest_count: int | None = None¶

sample_size: int | None = None¶

mismatched_partitions: list[str]¶

details: list[str]¶