Migration Context¶

iceberg_ops.migration¶

BaseMigration and MigrationContext — the two objects every migration file works with.

Authoring a migration¶

Create migrations/NNNN_<description>.py and define:

from iceberg_ops.migration import BaseMigration, MigrationContext
import pyarrow as pa

class Migration(BaseMigration):
    version     = "0001"
    description = "Create orders table"

    def upgrade(self, ctx: MigrationContext) -> None:
        ctx.create_table_if_not_exists(
            ctx.qualified("orders"),          # -> "<namespace>.orders"
            pa.schema([pa.field("id", pa.int64())]),
        )

    def downgrade(self, ctx: MigrationContext) -> None:
        ctx.drop_table_if_exists(ctx.qualified("orders"))

Conventions¶

version — zero-padded 4-digit string matching the file’s NNNN prefix.
upgrade() — must be idempotent.
downgrade() — exact reverse of upgrade; raise NotImplementedError when a full undo is impossible (e.g. column removal requires a table rewrite).

class iceberg_ops.migration.MigrationContext(catalog, namespace, version, description)[source]¶

Bases: object

Runtime environment injected into every upgrade() / downgrade() call.

Variables:

catalog (Catalog) – The live PyIceberg catalog being managed.
namespace (str) – The target namespace this migration operates on (e.g. "sales", "finance"). Use ctx.qualified(name) to build fully-qualified table identifiers.
version (str) – Version string of the migration currently executing.
description (str) – Human-readable description of the migration.

Parameters:

catalog (Any)
namespace (str)
version (str)
description (str)

catalog: Any¶

namespace: str¶

version: str¶

description: str¶

qualified(table_name)[source]¶

Return a fully-qualified table identifier.

If table_name already contains a dot ("sales.orders"), it is returned unchanged. Otherwise <namespace>.<table_name> is built.

Parameters:: table_name (str) – Bare ("orders") or qualified ("sales.orders") table name.
Returns:: The fully-qualified identifier.
Return type:: str

Example

# ctx.namespace = "sales"
ctx.qualified("orders")         # -> "sales.orders"
ctx.qualified("sales.orders")   # -> "sales.orders" (unchanged)

load_table(identifier)[source]¶

Load a table by its (relative or fully-qualified) identifier.

Parameters:: identifier (str) – Table name — relative ("orders", qualified with ctx.namespace) or qualified ("sales.orders").
Returns:: The loaded PyIceberg table.
Return type:: Table
Raises:: NoSuchTableError – If the table does not exist.

create_table_if_not_exists(identifier, schema, *, partition_spec=None, sort_order=None, properties=None, location=None)[source]¶

Create an Iceberg table or silently return the existing one.

The parent namespace (everything before the last dot, so nested namespaces such as "sales.eu" work) is created automatically if it does not exist. If another process creates the table between the existence check and the create, the existing table is returned.

Parameters:

identifier (str) – Table name — relative ("orders") or qualified ("sales.orders"). Relative names are qualified with ctx.namespace.
schema (Schema | Schema) – Schema used when creating a new table.
partition_spec (PartitionSpec | None, default: None) – Partition spec for a new table. Default: unpartitioned. Its fields reference source field ids, so it needs a pyiceberg.schema.Schema; with a pa.Schema, create the table first and then call update_partition_spec().
sort_order (SortOrder | None, default: None) – Sort order for a new table. Default: unsorted. Like partition_spec, it needs a pyiceberg.schema.Schema; with a pa.Schema, call replace_sort_order() after creating the table.
properties (dict[str, str] | None, default: None) – Table properties for a new table.
location (str | None, default: None) – Table location for a new table. Default: chosen by the catalog.

Returns:

The created table, or the existing one unchanged. The keyword arguments apply only when the table is created; an existing table is never altered to match them.

Return type:

Table

Raises:

ValueError – If schema is a pa.Schema and partition_spec or sort_order has fields. Raised before anything (namespace or table) is created.

drop_table_if_exists(identifier)[source]¶

Drop a table, silently succeeding if it does not exist.

Parameters:: identifier (str) – Table name — relative or fully qualified.
Return type:: None

get_helper(identifier)[source]¶

Return a SchemaEvolutionHelper bound to identifier.

Power-user escape hatch — the most common operations (add, enforce, rename, delete columns) are exposed directly on this context. Reach for the helper when you need its lower-level primitives or the contract= kwarg.

Parameters:: identifier (str) – Table name — relative or qualified.
Returns:: A helper bound to the freshly loaded table.
Return type:: SchemaEvolutionHelper
Raises:: NoSuchTableError – If the table does not exist.

add_column(identifier, name, field_type, *, doc=None)[source]¶

Add a column to a table.

Non-idempotent — raises if the column already exists. Use enforce_column() for the idempotent variant.

Supports dotted paths for nested struct fields (e.g. "address.zip_code"); the parent struct must already exist.

Parameters:

identifier (str) – Table name — relative or qualified.
name (str) – New column name. Dotted paths target nested struct fields.
field_type (IcebergType | DataType) – The column’s type. A primitive PyArrow type (e.g. pa.string()) is translated to its Iceberg equivalent.
doc (str | None, default: None) – Description stored in the Iceberg schema catalog.

Raises:

NoSuchTableError – If the table does not exist.
ValueError – If the column already exists, the parent struct of a dotted name does not exist, or a PyArrow field_type has no Iceberg primitive equivalent.

Return type:

None

enforce_column(identifier, name, field_type, *, doc=None, contract=None, legacy_archival=False)[source]¶

Idempotent column enforcement (delegates to SchemaEvolutionHelper.enforce_column()).

A column that already has an equivalent type (including an identical struct, list or map) is left alone, apart from a doc update. A promotion the Iceberg spec allows (int to long, float to double, wider decimal precision at the same scale) is applied in place; any other change, including ones PyIceberg itself would accept such as string to binary, is incompatible. With legacy_archival=False (default), an incompatible type change raises pyiceberg.exceptions.ValidationError and leaves the table untouched. With legacy_archival=True, the existing column is renamed to <name>__legacy_N and a fresh column is added; a step 2 failure reverses step 1 (compensating transaction). Archival happens only on a genuine type incompatibility, never because of a commit error or a failing contract.

Parameters:

identifier (str) – Table name — relative or qualified.
name (str) – Canonical column name. Dotted paths target nested struct fields.
field_type (IcebergType | DataType) – Intended type. A primitive PyArrow type (e.g. pa.float64()) is translated to its Iceberg equivalent.
doc (str | None, default: None) – Description stored in the Iceberg schema catalog.
contract (SchemaContract | None, default: None) – If provided, enforced once after the operation completes.
legacy_archival (bool, default: False) – Opt-in to the archive-and-recreate flow on incompatible type. Default False.

Raises:

pyiceberg.exceptions.ValidationError – If the type is incompatible and legacy_archival is False.
ValueError – If archival is requested for a column referenced by the partition spec, sort order or identifier fields.
SchemaContractViolation – If contract is given and the table does not conform.

Return type:

None

enforce_rename(identifier, old_name, new_name, *, doc=None, contract=None)[source]¶

Idempotent rename (delegates to SchemaEvolutionHelper.enforce_rename()).

Behaviour depends on which of old_name / new_name exists in the current schema; see the helper docstring for the matrix. Supports dotted paths for nested struct fields. When both exist, the blocking new_name is archived to <new_name>__legacy_N (made optional, [ARCHIVED] doc) unless the partition spec, sort order or identifier fields reference it, in which case ValueError is raised and the table is left untouched.

Parameters:

identifier (str) – Table name — relative or qualified.
old_name (str) – Current column name. Dotted paths target nested struct fields.
new_name (str) – Desired column name; must share old_name’s parent struct.
doc (str | None, default: None) – Description to set on the renamed column.
contract (SchemaContract | None, default: None) – If provided, enforced once after the operation completes (also when the rename is a no-op).

Raises:

NoSuchTableError – If the table does not exist.
ValueError – If the names do not share a parent struct, the parent struct does not exist, or a colliding new_name is referenced by the partition spec, sort order or identifier fields.
SchemaContractViolation – If contract is given and the table does not conform.

Return type:

None

delete_column(identifier, column_name)[source]¶

Delete a column from the table schema (metadata-only).

Old data files still contain the column but it becomes invisible to queries after this operation.

Parameters:

identifier (str) – Table name — relative or qualified.
column_name (str) – Name of the column to delete.

Raises:

NoSuchTableError – If the table does not exist.
ValueError – If the column does not exist.

Return type:

None

update_partition_spec(identifier, add_fields=None, remove_fields=None)[source]¶

Evolve the partition spec of a table.

Uses PyIceberg’s table.update_spec() context manager to add or remove partition fields in a single atomic commit.

Parameters:

identifier (str) – Table name — relative or qualified.
add_fields (list[tuple[str, Any, str]] | None, default: None) –
Each tuple contains:
- column_name — source column in the table schema.
- transform — a PyIceberg Transform instance (e.g. DayTransform(), IdentityTransform()).
- partition_name — the name for the new partition field.
remove_fields (list[str] | None, default: None) – Names of partition fields to remove (void-out).

Return type:

None

Example

from pyiceberg.transforms import DayTransform
ctx.update_partition_spec("orders", add_fields=[
    ("created_at", DayTransform(), "created_day"),
])

replace_sort_order(identifier, sort_fields=None)[source]¶

Replace the sort order of a table.

Builds a new sort order from the provided column / direction pairs and commits it via table.update_sort_order().

Parameters:

identifier (str) – Table name — relative or qualified.
sort_fields (list[tuple[str, str]] | None, default: None) – Each tuple is (column_name, "asc" | "desc"). If None or empty, the sort order is cleared.

Raises:

ValueError – If direction is not "asc" or "desc".

Return type:

None

Example

ctx.replace_sort_order("orders", sort_fields=[
    ("order_id", "asc"),
    ("created_at", "desc"),
])

set_table_properties(identifier, properties)[source]¶

Set one or more Iceberg table properties.

Uses a PyIceberg transaction so all properties land in a single atomic commit.

Parameters:

identifier (str) – Table name — relative or qualified.
properties (dict[str, str]) – Properties to set. Any key is allowed, including keys such as "properties".

Return type:

None

Example

ctx.set_table_properties("orders", {
    "write.parquet.compression-codec": "zstd",
})

unset_table_properties(identifier, keys)[source]¶

Remove one or more Iceberg table properties.

Uses a PyIceberg transaction so all keys are removed atomically. Keys that do not exist are silently ignored.

Parameters:

identifier (str) – Table name — relative or qualified.
keys (list[str]) – Property keys to remove.

Return type:

None

Example

ctx.unset_table_properties("orders", [
    "write.parquet.compression-codec",
])

set_compaction_target_size(identifier, target_mb)[source]¶

Set the target data file size for writers.

Sets the standard Iceberg write.target-file-size-bytes property. PyIceberg’s writers use it as their bin-packing target. iceberg-ops’ compaction helpers take an explicit target_size_mb (default 512) and do not read it; managed services such as S3 Tables configure compaction targets through their own maintenance settings.

Parameters:

identifier (str) – Table name — relative or qualified.
target_mb (int) – Target file size in megabytes. Stored as target_mb * 1024 * 1024.

Return type:

None

Example

ctx.set_compaction_target_size("orders", target_mb=256)

set_snapshot_retention(identifier, *, min_snapshots=None, max_age_hours=None)[source]¶

Configure snapshot retention via standard Iceberg properties.

Sets one or both of:

history.expire.min-snapshots-to-keep — minimum number of snapshots Iceberg must retain regardless of age (Iceberg default: 1).
history.expire.max-snapshot-age-ms — snapshots older than this become eligible for expiration (Iceberg default: 5 days).

Both conditions must be satisfied for a snapshot to be expired. Calling with both arguments None is a no-op.

Parameters:

identifier (str) – Table name — relative or qualified.
min_snapshots (int | None, default: None) – Minimum snapshot count to retain.
max_age_hours (int | None, default: None) – Maximum snapshot age in hours. Stored as max_age_hours * 3600 * 1000 milliseconds.

Return type:

None

Notes

These are the standard Iceberg properties read by engines’ expire-snapshots procedures. Do not set them on AWS S3 Tables: S3 Tables manages retention through its own maintenance configuration, and AWS documents that these properties make its snapshot management fail; a warning is logged when the catalog is configured for S3 Tables. expire_snapshots() reads these properties when its max_age_hours / min_snapshots arguments are omitted.

Example

ctx.set_snapshot_retention("orders", min_snapshots=5, max_age_hours=72)

set_clustering_keys(identifier, columns, *, direction='asc')[source]¶

Set the table’s Iceberg sort order on columns with a uniform direction.

The sort order is the only portable, spec-defined clustering mechanism; iceberg-ops writes no vendor-specific properties. Whether an engine or managed service uses the sort order when writing or compacting depends on that engine. This is a single replace_sort_order() commit.

Parameters:

identifier (str) – Table name — relative or qualified.
columns (list[str]) – One or more column names to cluster by, in priority order. Must be non-empty.
direction (str, default: 'asc') – "asc" (default) or "desc". Applied uniformly to every column. For mixed directions, call replace_sort_order() directly.

Raises:

ValueError – If columns is empty, or if direction is not "asc" / "desc".

Return type:

None

class iceberg_ops.migration.BaseMigration[source]¶

Bases: object

Abstract base class for all migration files.

Subclass this and name the subclass Migration exactly — the runner discovers it by that name.

The runner calls methods in this order:

pre_upgrade()    -> upgrade()    -> post_upgrade()
pre_downgrade()  -> downgrade()  -> post_downgrade()

All four hook methods (pre_upgrade, post_upgrade, pre_downgrade, post_downgrade) are optional no-ops by default. Override them to add precondition checks, notifications, or cleanup logic around the core migration.

Class attributes¶

versionstr: Zero-padded 4-digit string ("0001"). Must be unique within the project.
descriptionstr: Short summary stored in the migration history table.

version: str = ''¶

description: str = ''¶

upgrade(ctx)[source]¶

Apply this migration. Must be idempotent.

Parameters:: ctx (MigrationContext) – Runtime context: catalog, target namespace and table helpers.
Raises:: NotImplementedError – Unless overridden.
Return type:: None

downgrade(ctx)[source]¶

Reverse this migration.

Raise NotImplementedError when a full undo is not possible (e.g. column removal requires a full table rewrite in Iceberg).

Parameters:: ctx (MigrationContext) – Runtime context: catalog, target namespace and table helpers.
Raises:: NotImplementedError – Unless overridden.
Return type:: None

pre_upgrade(ctx)[source]¶

Override to validate preconditions before upgrade().

If this method raises, upgrade() does not run and the migration is recorded as failed.

Parameters:: ctx (MigrationContext) – Runtime context: catalog, target namespace and table helpers.
Return type:: None

post_upgrade(ctx)[source]¶

Override to perform post-migration actions after upgrade().

If this method raises, the migration is still recorded as applied (the schema change already committed); a warning is logged.

Parameters:: ctx (MigrationContext) – Runtime context: catalog, target namespace and table helpers.
Return type:: None

pre_downgrade(ctx)[source]¶

Override to validate preconditions before downgrade().

If this method raises, downgrade() does not run and the rollback fails.

Parameters:: ctx (MigrationContext) – Runtime context: catalog, target namespace and table helpers.
Return type:: None

post_downgrade(ctx)[source]¶

Override to perform post-rollback cleanup.

If this method raises, the rollback is still recorded as completed; a warning is logged.

Parameters:: ctx (MigrationContext) – Runtime context: catalog, target namespace and table helpers.
Return type:: None