iceberg-ops =========== **Alembic-style schema migrations for Apache Iceberg.** Versioned, reviewable table changes: numbered ``upgrade()`` / ``downgrade()`` revisions, autogenerate, ``stamp``, and a migration history stored as an Iceberg table in your own catalog. Pure Python on PyIceberg: no Spark, no JVM. iceberg-ops brings the migration workflow that application teams use for relational databases to Apache Iceberg tables. Each schema change is a numbered Python file that you review in a pull request, apply with ``migrate`` and undo with ``rollback``. The status of each migration is recorded in a history table that lives next to your data. The CLI is ``iceberg-ops``; the Python package is ``iceberg_ops``. 60-second demo -------------- .. code-block:: bash pip install "iceberg-ops[sqlite]" mkdir demo && cd demo iceberg-ops init iceberg-ops revision -m "create orders" # then fill in upgrade() / downgrade() iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales migrate # Describe the next schema in contracts/orders.yaml, then let iceberg-ops write the migration: iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales \ revision -m "rename total, add status" --autogenerate \ --target-table orders --target-schema-yaml contracts/orders.yaml \ --rename-hint order_total:order_amount iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales plan iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales migrate iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales status iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales rollback :doc:`quickstart` walks through the same steps with the contents of the migration file and the schema contract. What iceberg-ops is, and isn't ------------------------------ **It is** a migration framework (files, history, rollback, ``stamp``, autogenerate) for Iceberg schemas, partition specs, sort orders and table properties, shipped as a Python library and CLI on top of PyIceberg. The Python API takes any PyIceberg catalog; the CLI opens a local SQLite catalog or any catalog defined in ``.pyiceberg.yaml`` or ``PYICEBERG_CATALOG__*`` environment variables. **It isn't:** * a query engine or a managed maintenance service; * a merge-on-read writer: the rewrites it performs are copy-on-write, and it does not write delete files or deletion vectors. Compaction does not support tables that already contain them and does not check for them, so do not run it on such tables; * a lock service: concurrent runs for one namespace are not guarded. A second concurrent run is detected when it records history (``ConcurrentMigrationError``), but its ``upgrade()`` may already have run. Run migrations for a namespace from a single job; * a partition-rewrite engine: ``migrate-data --write-mode overwrite_partition`` is refused unless the whole migration fits in one commit. Also included, secondary to the migration workflow: * **Cross-table data migration (experimental):** ``iceberg-ops migrate-data`` streams rows between Iceberg tables, in one catalog or across catalogs. See :doc:`data_migration`. * **Maintenance helpers (basic):** snapshot expiry, a health report, and compaction and orphan removal for tables on the local filesystem. See :doc:`maintenance`. * **Apache Polaris policy executor (experimental):** runs ``system.snapshot-expiry`` policies; it does not yet work against a released Polaris server. See :doc:`polaris`. :doc:`compatibility` lists the supported versions, catalogs and limits. .. toctree:: :maxdepth: 2 :caption: User Guide quickstart schema_migrations autogenerate architecture compatibility data_migration maintenance polaris cli .. toctree:: :maxdepth: 2 :caption: API Reference api/index .. toctree:: :maxdepth: 1 :caption: Project changelog Project links ------------- * `Source code `__ and `issues `__ * `Contributing `__ * `Roadmap `__ * `Security policy `__ * `Code of conduct `__ For coding agents: * `AGENTS.md `__ is the guide for agents that change the repository. * `agent-skills/iceberg-ops-migrations/SKILL.md `__ is a skill that teaches a coding agent to write, review and run iceberg-ops migrations. * `llms.txt `__ and `llms-full.txt `__ are a plain-text index of this site and the full text of every page. Indices ------- * :ref:`genindex` * :ref:`modindex` * :ref:`search` License and trademarks ---------------------- iceberg-ops is released under the `MIT License `__. Apache Iceberg, Iceberg, Apache Polaris and Apache are trademarks of the Apache Software Foundation. iceberg-ops is not affiliated with or endorsed by the ASF. Quick Start =========== This page builds a small project against a local SQLite catalog: one hand-written migration, one autogenerated migration, then ``migrate``, ``status`` and ``rollback``. Every command runs as shown. Install ------- .. code-block:: bash python -m venv .venv source .venv/bin/activate # Windows PowerShell: .venv\Scripts\Activate.ps1 pip install "iceberg-ops[sqlite]" The ``[sqlite]`` extra brings the local SQLite and in-memory catalogs. Other extras: .. code-block:: bash pip install "iceberg-ops[glue]" # AWS Glue pip install "iceberg-ops[hive]" # Hive Metastore pip install "iceberg-ops[s3tables]" # AWS S3 Tables (REST with SigV4 signing) pip install "iceberg-ops[transforms]" # writing to bucket / truncate / time-partitioned tables pip install "iceberg-ops[polaris]" # experimental Apache Polaris policy executor REST catalogs (Apache Polaris, Nessie, Lakekeeper and others) need no extra. :doc:`compatibility` lists the supported Python, PyIceberg and PyArrow versions. To work on iceberg-ops itself, see `CONTRIBUTING.md `__. Scaffold a project ------------------ .. code-block:: bash mkdir demo && cd demo iceberg-ops init This writes: * ``iceberg-ops.yaml`` -- a config file stub that selects the local SQLite catalog (``catalog: sqlite``) and the ``default`` namespace * ``migrations/__init__.py`` -- so the directory is a real Python package * a marked block in ``.gitignore`` that ignores ``local_data/``, where the SQLite catalog lives by default Write your first migration -------------------------- .. code-block:: bash iceberg-ops revision -m "create orders" This creates ``migrations/0001_create_orders.py``. Replace its contents with: .. code-block:: python # migrations/0001_create_orders.py import pyarrow as pa from iceberg_ops.migration import BaseMigration, MigrationContext class Migration(BaseMigration): version = "0001" description = "Create orders table" def upgrade(self, ctx: MigrationContext) -> None: ctx.create_table_if_not_exists( "orders", pa.schema( [ pa.field("order_id", pa.int64(), nullable=False), pa.field("order_total", pa.float64()), pa.field("created_at", pa.timestamp("us", tz="UTC")), ] ), ) def downgrade(self, ctx: MigrationContext) -> None: ctx.drop_table_if_exists("orders") Migration files are *namespace-agnostic*: pass bare table names to the ``MigrationContext`` helpers and the runner qualifies them with the active namespace at apply time, so the same file works for ``sales``, ``staging`` or ``finance``. ``version`` must match the file's ``NNNN`` prefix. Apply it to a local SQLite catalog (``demo.db``, data files under ``warehouse/``) in the ``sales`` namespace. Global options go before the subcommand: .. code-block:: bash iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales migrate Autogenerate the next migration ------------------------------- Describe the schema you want in ``contracts/orders.yaml``: .. code-block:: yaml # contracts/orders.yaml columns: - {name: order_id, type: long, required: true} - {name: order_amount, type: double} # renamed from order_total - {name: created_at, type: timestamptz} - {name: status, type: string} # new column Let iceberg-ops diff the live table against it and write the migration: .. code-block:: bash iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales \ revision -m "rename total, add status" --autogenerate \ --target-table orders --target-schema-yaml contracts/orders.yaml \ --rename-hint order_total:order_amount The command writes ``migrations/0002_rename_total_add_status.py``. Review it before running it. Its migration class reads: .. code-block:: python class Migration(BaseMigration): version = '0002' description = 'rename total, add status' def upgrade(self, ctx: MigrationContext) -> None: target_table = 'orders' ctx.enforce_rename(target_table, 'order_total', 'order_amount') ctx.enforce_column(target_table, 'status', ice_types.StringType()) def downgrade(self, ctx: MigrationContext) -> None: target_table = 'orders' ctx.delete_column(target_table, 'status') ctx.enforce_rename(target_table, 'order_amount', 'order_total') Without the ``--rename-hint``, the rename would appear as a drop plus an add. A step that cannot be undone, such as a column drop, would make the generated ``downgrade()`` raise ``NotImplementedError`` before any step runs. See :doc:`autogenerate`. Run migrations -- CLI --------------------- .. code-block:: bash iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales plan iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales migrate iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales status iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales rollback iceberg-ops --catalog sqlite --db-path demo.db --warehouse warehouse --namespace sales history ``rollback`` undoes one migration; ``--steps N`` or ``--target 0001`` goes further back. ``history`` prints the latest recorded status of each version. ``status``, ``plan`` and ``history`` only print, and exit 0 even when a migration is recorded as failed. To drop the flags, put them in ``iceberg-ops.yaml``. Relative paths in the file resolve against the directory that holds it: .. code-block:: yaml # iceberg-ops.yaml catalog: sqlite namespace: sales db_path: demo.db warehouse: warehouse Then ``iceberg-ops migrate``, ``iceberg-ops status`` and the other commands need no options. Without ``db_path`` and ``warehouse``, the SQLite catalog lives under ``local_data/`` next to the config file. Run migrations -- Python API ---------------------------- .. code-block:: python from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.runner import MigrationRunner catalog = CatalogFactory.build("sqlite", db_path="demo.db", warehouse_dir="warehouse") runner = MigrationRunner(catalog, namespace="sales", migrations_dir="migrations") runner.status() # show pending vs applied runner.migrate() # apply all pending migrations runner.rollback(steps=1) # undo the last migration runner.history() # latest recorded status of each version The runner methods print their report and return ``None``. Use your own catalog -------------------- ``--catalog-name`` (or ``catalog_name:`` in ``iceberg-ops.yaml``) opens a catalog defined in ``.pyiceberg.yaml`` or ``PYICEBERG_CATALOG____*`` environment variables: .. code-block:: bash iceberg-ops --catalog-name prod --namespace sales plan iceberg-ops --catalog-name prod --namespace sales migrate From Python, pass any PyIceberg catalog, for example ``pyiceberg.catalog.load_catalog("prod")``, to ``MigrationRunner``. Only the SQLite and in-memory catalogs are integration-tested; see :doc:`compatibility`. Adopt on an existing deployment ------------------------------- If your tables were created out-of-band, ``stamp`` records every discovered migration as applied without invoking ``upgrade()``: .. code-block:: bash iceberg-ops --catalog-name prod --namespace sales stamp iceberg-ops --catalog-name prod --namespace sales stamp --target 0002 # or stop at a version Later migrations apply normally on top. Where to next ------------- * :doc:`schema_migrations` -- migration authoring guide. * :doc:`autogenerate` -- generate migrations from schema diffs. * :doc:`architecture` -- how iceberg-ops is put together. * :doc:`compatibility` -- supported versions, catalogs and limits. * :doc:`data_migration` -- cross-table data movement (experimental). * :doc:`maintenance` -- basic maintenance helpers. * :doc:`polaris` -- Apache Polaris policy executor (experimental). * :doc:`cli` -- CLI reference. * :doc:`api/index` -- Python API reference. Schema Migrations ================= iceberg-ops takes the Alembic mental model -- versioned ``NNNN_*.py`` files driven by a ``MigrationContext`` -- and adapts it to PyIceberg's metadata-only DDL primitives. Each file has an ``upgrade()`` and a ``downgrade()``; the runner applies pending files in version order and records each version's status in a history table inside your own catalog. Migration file conventions -------------------------- .. list-table:: :header-rows: 1 :widths: 25 75 * - Rule - Detail * - File name - ``NNNN_.py`` (4-digit zero-padded version, an underscore, then a slug). Files that do not match are ignored. * - Class name - Must be exactly ``Migration``. A matching file without a ``Migration`` class is skipped with a warning. * - ``version`` - String equal to the file's ``NNNN`` prefix. A mismatch, or two files with the same version, is an error. * - ``description`` - Human-readable description; surfaces in ``status`` / ``history``. * - ``upgrade()`` - Must be idempotent (see `Concurrent runs`_). Prefer ``ctx.create_table_if_not_exists`` / ``enforce_column`` / ``enforce_rename`` over raw create / add / rename. * - ``downgrade()`` - Reverses ``upgrade()``. When a step cannot be undone, raise ``NotImplementedError`` before changing anything (see `Downgrades`_). * - Table identifiers - Pass bare table names (``"orders"``); the context qualifies them with the namespace the runner was given, so one file serves every namespace. A dotted name (``"sales.orders"``) is used unchanged and pins the migration to that table. * - Directory layout - Flat (``migrations/0001_*.py``) or one subdirectory per namespace (``migrations/sales/0001_*.py``); a subdirectory named after the runner's namespace is used when it exists. A missing migrations directory is an error (``FileNotFoundError``). Each ``ctx`` helper call is its own Iceberg commit, so a migration is not atomic: if ``upgrade()`` fails half-way, the earlier steps stay committed and the version is recorded as ``failed``. A failed version is pending again on the next ``migrate``, which is another reason to keep ``upgrade()`` idempotent. MigrationContext API -------------------- The context is injected into every ``upgrade()`` / ``downgrade()`` call. The core helpers are available as direct methods on the context; ``ctx.get_helper(table)`` returns a ``SchemaEvolutionHelper`` for its lower-level primitives. Types: column helpers take a PyIceberg type (``pyiceberg.types.DoubleType()``, ``StringType()``, ...) or a primitive PyArrow type (``pa.float64()``, ``pa.string()``), which is translated to its Iceberg equivalent. Nested types (struct / list / map) must be given as PyIceberg types. Table schemas take a ``pyarrow.Schema`` or a ``pyiceberg.schema.Schema``. **Table lifecycle** * ``ctx.create_table_if_not_exists(identifier, schema, *, partition_spec=None, sort_order=None, properties=None, location=None)`` -- create the table, or return the existing one unchanged. The keyword arguments apply only when the table is created. The parent namespace is created if needed (nested namespaces such as ``"sales.eu"`` work). ``partition_spec`` and ``sort_order`` reference source field ids, so pass them together with a ``pyiceberg.schema.Schema``; with a ``pyarrow.Schema``, create the table first and then call ``update_partition_spec`` / ``replace_sort_order``. * ``ctx.drop_table_if_exists(identifier)`` * ``ctx.load_table(identifier)`` * ``ctx.qualified(name)`` -- returns ``"."``; a name that already contains a dot is returned unchanged. **Column lifecycle** * ``ctx.add_column(identifier, name, field_type, *, doc=None)`` -- add a new column. Raises if it already exists. Dotted names (``"address.zip_code"``) add a field to an existing struct. * ``ctx.enforce_column(identifier, name, field_type, *, doc=None, contract=None, legacy_archival=False)`` -- ensure the column exists with the given type; see the `Behaviour matrix`_ below. * ``ctx.enforce_rename(identifier, old_name, new_name, *, doc=None, contract=None)`` -- rename idempotently. Both names must share a parent struct. When both names exist, the blocking column is archived first (see `Rename collisions`_). * ``ctx.delete_column(identifier, column_name)`` -- metadata-only delete. Data files still hold the values (reachable by time travel). **Partition spec evolution** * ``ctx.update_partition_spec(identifier, add_fields=None, remove_fields=None)`` -- add / remove partition fields in one commit. ``add_fields`` is a list of ``(source_column, transform, partition_name)`` tuples, where ``transform`` is a PyIceberg transform instance such as ``DayTransform()``; ``remove_fields`` is a list of partition field names. Writing to a table partitioned by a non-identity transform (bucket, truncate, year, month, day, hour) needs ``pyiceberg-core``: ``pip install "iceberg-ops[transforms]"``. **Sort order management** * ``ctx.replace_sort_order(identifier, sort_fields=None)`` -- replace the table's sort order. Each entry is ``(column, "asc" | "desc")`` with an identity transform (``asc`` sorts nulls first, ``desc`` nulls last). ``None`` or an empty list clears the sort order. **Table properties** * ``ctx.set_table_properties(identifier, properties)`` -- set one or more Iceberg table properties in one commit. * ``ctx.unset_table_properties(identifier, keys)`` -- remove properties by key; missing keys are ignored. **Convenience accessors** * ``ctx.set_compaction_target_size(identifier, target_mb)`` -- sets ``write.target-file-size-bytes`` (``target_mb`` MiB), which PyIceberg's writers use as their target file size. The iceberg-ops compaction helper does not read it; it takes an explicit ``target_size_mb``. * ``ctx.set_snapshot_retention(identifier, *, min_snapshots=None, max_age_hours=None)`` -- sets the standard ``history.expire.min-snapshots-to-keep`` / ``history.expire.max-snapshot-age-ms`` properties. :py:func:`iceberg_ops.maintenance.expire_snapshots` reads them when its own arguments are omitted. Do not use it on AWS S3 Tables, which manages snapshot retention itself; a warning is logged when the catalog is configured for S3 Tables. * ``ctx.set_clustering_keys(identifier, columns, *, direction="asc")`` -- sets the table sort order on ``columns`` (the same as ``replace_sort_order`` with one direction for every column). It writes no other properties. Lifecycle hooks --------------- ``BaseMigration`` carries four optional hooks: .. code-block:: text pre_upgrade() -> upgrade() -> post_upgrade() pre_downgrade() -> downgrade() -> post_downgrade() All four are no-ops by default. Error semantics: .. list-table:: :header-rows: 1 :widths: 25 75 * - Hook - If it raises * - ``pre_upgrade`` - The migration is not applied: it is recorded as ``failed``, the run stops, and ``migrate`` raises ``RuntimeError`` (the CLI exits 1). * - ``post_upgrade`` - The migration is already recorded as applied -- a warning is logged, nothing is rolled back. * - ``pre_downgrade`` - The rollback stops with an error (``RuntimeError``); the version stays applied. * - ``post_downgrade`` - The rollback is already recorded -- a warning is logged. Example: pre_upgrade quality gate --------------------------------- .. code-block:: python # migrations/0010_add_session_duration.py from pyiceberg.types import DoubleType from iceberg_ops.migration import BaseMigration, MigrationContext class Migration(BaseMigration): version = "0010" description = "Add session_duration with non-empty gate" def pre_upgrade(self, ctx: MigrationContext) -> None: tbl = ctx.load_table("events") if tbl.current_snapshot() is None: raise RuntimeError("Cannot migrate empty events table.") def upgrade(self, ctx: MigrationContext) -> None: ctx.enforce_column("events", "session_duration", DoubleType()) def downgrade(self, ctx: MigrationContext) -> None: ctx.delete_column("events", "session_duration") Downgrades ---------- ``downgrade()`` undoes ``upgrade()`` where the prior state is known: a column that ``upgrade()`` added can be deleted, a rename can be renamed back, a property can be restored to its old value. Some steps cannot be undone: * a column delete (re-adding it creates a new, empty column; the old values are reachable only by time travel); * a type promotion such as ``int -> long`` (Iceberg only widens types); * a legacy-archival type change (see below). For those, raise ``NotImplementedError`` as the first statement of ``downgrade()``, before any reversible step runs, so a rollback leaves the table untouched. The runner re-raises ``NotImplementedError`` (the CLI exits 1 with one ``Error:`` line), the version stays applied, and the rollback stops; versions already rolled back earlier in the same run stay rolled back. Files written by ``revision --autogenerate`` follow this rule (see :doc:`autogenerate`). Before any ``downgrade()`` runs, ``rollback`` checks that the migration file of every version it has to undo exists; if one is missing, nothing is rolled back. ``rollback(target="0000")`` (CLI: ``rollback --target 0000``) rolls back every applied migration. Legacy archival on incompatible type change ------------------------------------------- Iceberg's spec only allows *promoting* primitive types in place (``int -> long``, ``float -> double``, ``decimal`` precision widening at the same scale). Anything else -- ``int -> string``, ``string -> long``, ``string -> binary``, a decimal scale change, ``timestamp -> date`` -- is not a promotion. ``enforce_column`` checks this allow-list itself rather than relying on PyIceberg, which accepts some changes the spec forbids. Because Iceberg has no atomic "drop and re-add a column with the same name" primitive, the helper implements an **archive-and-recreate** flow guarded by a compensating transaction. Opting in is explicit: pass ``legacy_archival=True`` to ``ctx.enforce_column``. By default an incompatible type change raises ``pyiceberg.exceptions.ValidationError`` and the table is left untouched, so the migration author sees the conflict and decides what to do (cast in a backfill, write a data migration, drop the column, etc.). Behaviour matrix ~~~~~~~~~~~~~~~~ .. list-table:: :header-rows: 1 :widths: 35 65 * - Starting state - Action * - Column absent - Add a new column with the requested type. * - Column present, equivalent type (including an identical struct / list / map) - No-op; only a changed ``doc`` is written. * - Column present, promotable primitive type (e.g. ``int -> long``) - In-place ``update_column``. * - Column present, incompatible type, ``legacy_archival=False`` - Raise ``pyiceberg.exceptions.ValidationError``. Schema unchanged. * - Column present, incompatible type, ``legacy_archival=True`` - **Step 1**: rename the existing column to ``__legacy_N`` (and make it optional if it was required). **Step 2**: add a fresh column at the original name with the new type and prefix the legacy column's doc with ``[ARCHIVED]``. If step 2 fails, the helper renames the legacy column back, restores its required flag (compensating transaction) and re-raises the step-2 error. If that compensation also fails, a ``RuntimeError`` names the stranded column. Archival is refused with ``ValueError`` (table untouched) for a column that the partition spec, the sort order or the identifier fields reference, because the archived column keeps its field id and those references would silently move to it. Archival runs only on a genuine type incompatibility: commit errors and failing contracts propagate without archiving anything. The two steps are separate commits. If the process dies between them, the column exists only as ``__legacy_N``; re-running ``enforce_column`` then sees the column as absent and adds the fresh one. The new column starts empty: existing rows read it as null until you backfill it. Rename collisions ~~~~~~~~~~~~~~~~~ ``ctx.enforce_rename(table, old, new)`` uses the same archival path when both ``old`` and ``new`` exist: the blocking ``new`` column is renamed to ``__legacy_N`` (and made optional), then ``old`` is renamed to ``new`` and the archived column gets an ``[ARCHIVED]`` doc. The same layout check applies (``ValueError`` if the partition spec, sort order or identifier fields reference the blocking column). If the rename fails, the archive is renamed back (compensating transaction) and the error is re-raised. The ``__legacy_N`` suffix ~~~~~~~~~~~~~~~~~~~~~~~~~ ``N`` starts at ``1`` and increments. Every archival call scans the current schema (or, for nested fields, the current parent struct's children) for existing ``__legacy_`` siblings and picks ``max(N) + 1``. A column that has been re-archived three times will therefore have siblings ``v__legacy_1``, ``v__legacy_2``, and ``v__legacy_3`` all coexisting in the schema. Nested struct fields use dotted paths (``address.zip_code``); the archival is scoped to the parent struct, so the archived sibling is ``address.zip_code__legacy_1`` and never collides with a top-level ``zip_code`` column or a sibling struct's ``zip_code``. Example ~~~~~~~ .. code-block:: python # migrations/0042_retype_duration.py import pyarrow as pa from iceberg_ops.migration import BaseMigration, MigrationContext class Migration(BaseMigration): version = "0042" description = "Re-type events.duration from int to string" def upgrade(self, ctx: MigrationContext) -> None: # int -> string is not a promotion. Opt in to archival so the # historical int values are preserved as duration__legacy_1 # while a fresh string column takes the canonical name. ctx.enforce_column( "events", "duration", pa.string(), legacy_archival=True, ) def downgrade(self, ctx: MigrationContext) -> None: # Raised before any change: the archived column carries the # original data; rename it back by hand if needed. raise NotImplementedError( "Reversing a legacy_archival type change requires a manual " "rename of duration__legacy_1 back to duration." ) After ``upgrade`` runs, the schema contains both ``duration`` (string) and ``duration__legacy_1`` (int). The legacy column's doc carries the ``[ARCHIVED]`` prefix so it is obvious at a glance which fields are historical. Interaction with autogenerate ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ``iceberg-ops revision --autogenerate`` (see :doc:`autogenerate`) emits a type change only when it is an Iceberg promotion (``ctx.enforce_column(target_table, "", )``), and never adds ``legacy_archival=True``. For any other type change the generated ``upgrade()`` raises ``NotImplementedError`` before any step runs, with the generated calls kept as comments. The generated ``downgrade()`` of any type change raises as well, because a type cannot be narrowed back. To finish such a file, either: * write the step with ``legacy_archival=True`` to opt into the archive-and-recreate flow, or * replace it with a hand-written data migration (see :doc:`data_migration`) that copies the old values into a new column before dropping the original. Migration history ----------------- The runner records history in the ``_iceberg_ops.migration_history`` Iceberg table of the same catalog (the meta-namespace is configurable). It holds one row per ``(version, namespace)`` with the latest status (``applied``, ``rolled_back`` or ``failed``, plus the error text for a failure). Earlier states are not kept as rows; they stay readable by time travel on the history table until its snapshots expire. ``plan``, ``status`` and ``history`` only read the catalog: they do not create the namespace or the table, which are created on the first write. ``plan`` and ``status`` still import every migration file (see :ref:`migration-trust-model`). ``stamp`` records versions as applied without running ``upgrade()``, to adopt iceberg-ops on tables that already exist. It does not check that the live schema matches what the stamped migrations would have produced. Concurrent runs --------------- Concurrent runs are not guarded. Two ``migrate`` runs for one namespace can both execute ``upgrade()`` for the same version; the second run to record history raises ``iceberg_ops.history.ConcurrentMigrationError`` (also importable from ``iceberg_ops.runner``), and history keeps one row per ``(version, namespace)`` instead of a duplicate. That run's ``upgrade()`` may already have executed, so ``upgrade()`` must be idempotent. When a migration's own commit conflicts with a concurrent commit, the runner raises ``ConcurrentMigrationError`` and records nothing for that version, so the other run's history row is not overwritten. Run migrations for a namespace from one job at a time (for example a CI concurrency group), or put an external lock in front of the runner. Runners for different namespaces can share the history table. .. _migration-trust-model: Trust model ----------- Migration files are Python code, and the runner executes them. ``status``, ``current``, ``plan``, ``migrate``, ``rollback`` and ``stamp`` (and the matching ``MigrationRunner`` methods) import every ``NNNN_*.py`` file in the migrations directory, or in its ``/`` subdirectory when that exists, before they print or change anything. Module-level code in those files runs with the caller's catalog credentials, so ``plan`` and ``status`` write nothing to the catalog but are not safe to run on files you have not reviewed. ``history`` reads the history table without importing the files. ``revision --target-schema-arrow module:attr`` imports the module it names, looking in the current directory first. Review migration files, the ``migrations_dir`` setting of a cloned project's ``iceberg-ops.yaml``, and modules named by ``--target-schema-arrow`` as you would any other code. Do not run these commands on an untrusted branch with production credentials, for example in a CI job triggered by ``pull_request_target``. Running migrations from Python ------------------------------ The CLI (see :doc:`cli`) wraps ``MigrationRunner``. The same flow from Python, against a throwaway in-memory catalog: .. code-block:: python import tempfile from pathlib import Path from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.runner import MigrationRunner migrations_dir = Path(tempfile.mkdtemp()) / "migrations" migrations_dir.mkdir() (migrations_dir / "0001_create_orders.py").write_text( ''' import pyarrow as pa from iceberg_ops.migration import BaseMigration, MigrationContext class Migration(BaseMigration): version = "0001" description = "Create orders table" def upgrade(self, ctx: MigrationContext) -> None: ctx.create_table_if_not_exists( "orders", pa.schema([pa.field("order_id", pa.int64(), nullable=False)]), ) def downgrade(self, ctx: MigrationContext) -> None: ctx.drop_table_if_exists("orders") ''', encoding="utf-8", ) catalog = CatalogFactory.build("memory") # use your own catalog in practice runner = MigrationRunner(catalog, namespace="sales", migrations_dir=str(migrations_dir)) runner.plan() # writes nothing to the catalog, but imports every migration file runner.migrate() # applies 0001 to sales.orders runner.status() runner.rollback(steps=1) # runs 0001's downgrade() Schema contracts ---------------- Schema contracts encode "this is what the table must look like" and verify the live schema matches. They are useful as a CI gate or as the target of an autogenerate run (see :doc:`autogenerate`). Python API: .. code-block:: python import pyarrow as pa from pyiceberg import types as ice_types from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.core.schema import ColumnSpec, SchemaContract catalog = CatalogFactory.build("memory") catalog.create_namespace("sales") table = catalog.create_table( "sales.orders", pa.schema( [ pa.field("order_id", pa.int64(), nullable=False), pa.field("order_total", pa.float64()), ] ), ) contract = SchemaContract( columns=[ ColumnSpec("order_id", ice_types.LongType(), required=True), ColumnSpec("order_total", ice_types.DoubleType()), ], strict=False, ) contract.enforce(table) # raises SchemaContractViolation on drift ``required=True`` means the column must be present; it does not check nullability. Types are compared exactly, so a column that was promoted (``int -> long``) is reported if the contract still names the old type. ``strict=True`` also reports columns the contract does not list. YAML contracts (PyYAML is a core dependency): .. code-block:: yaml # contracts/orders.yaml strict: false columns: - name: order_id type: long required: true - name: order_total type: double - name: created_at type: timestamptz .. code-block:: python # illustrative: needs contracts/orders.yaml and a loaded table from iceberg_ops.core.schema import SchemaContract contract = SchemaContract.from_yaml("contracts/orders.yaml") contract.enforce(table) The file is read as UTF-8 and validated: ``columns`` must be a non-empty list with unique names, and ``strict`` / ``required`` must be booleans. Supported types (case-insensitive): ``boolean``, ``int`` (or ``integer``), ``long``, ``float``, ``double``, ``string``, ``binary``, ``date``, ``time``, ``timestamp``, ``timestamptz``, ``uuid``, ``decimal(P,S)``, ``fixed[N]``, and the other Iceberg primitive type strings such as ``timestamp_ns``. Nested types are not supported. See :doc:`api/migration`, :doc:`api/runner` and :doc:`api/schema` for the full reference. Autogenerate ============ ``iceberg-ops revision --autogenerate`` compares an existing Iceberg table against a target schema and writes a migration file whose ``upgrade()`` closes the gap and whose ``downgrade()`` undoes it where the prior state is known. * From the CLI, the target is a YAML ``SchemaContract`` or an importable ``pa.Schema`` (``module:attr``), and only columns are diffed. * From Python, the target can also be a ``pyiceberg.schema.Schema``, and ``spec_diff_to_ops`` / ``sort_order_diff_to_ops`` / ``properties_diff_to_ops`` diff the partition spec, sort order and table properties. Generated files always carry a "Review every line before running" warning: autogenerate is a *starting point*, not a finished implementation. Rename detection in particular requires explicit hints, and steps that cannot run or cannot be reversed make the generated method raise instead of running (see `Limitations`_). Generated files keep the table name exactly as you pass it. Pass a bare name (``orders``): the migration context qualifies it with the namespace given at ``migrate`` time, so one file serves every namespace. A dotted name (``sales.orders``) pins the file to that one table, and the CLI prints a warning. Diff functions -------------- The autogenerate machinery decomposes into four diff helpers, all exposed on the public API: * ``schema_diff_to_ops(*, current, target, rename_hints=None)`` emits ``add_column`` / ``rename_column`` / ``delete_column`` / ``type_change`` ops. Without ``rename_hints``, a rename is indistinguishable from a drop + add and the diff falls back to drop + add. With a ``pa.Schema`` or ``pyiceberg.schema.Schema`` target, table columns missing from the target are dropped. A non-strict ``SchemaContract`` target (the default) only reconciles the columns it lists and keeps the rest; a strict one drops unlisted columns. * ``spec_diff_to_ops(*, current, target, schema)`` emits ``add_partition_field`` / ``remove_partition_field`` ops for partition spec evolution, matching fields by partition name. The generated call passes the transform as ``transforms.(...)`` and the file imports ``from pyiceberg import transforms``. * ``sort_order_diff_to_ops(*, current, target, schema)`` emits a single ``set_sort_order`` op when the orders differ (replace semantics, not delta). * ``properties_diff_to_ops(*, current, target)`` emits ``set_property`` / ``remove_property`` ops, preserving the previous value in args so the generated downgrade can restore it. Renderer -------- ``render_migration_body(ops, *, target_table)`` returns ``(upgrade_body, downgrade_body)``; ``render_migration_file(*, version, description, ops, target_table)`` wraps them in the file template; ``write_autogenerated_revision(*, version, description, ops, target_table, out_path, overwrite=False)`` writes the file as UTF-8 with LF line endings. It raises ``FileExistsError`` when ``out_path`` exists unless ``overwrite=True``, and it does not create missing directories. Every name, value and description is written as a Python literal, so a generated file always compiles. Generated migrations call ``ctx.(target_table, ...)`` directly -- no ``helper = ctx.get_helper(target_table)`` boilerplate -- because ``MigrationContext`` exposes the column, partition, sort-order and property helpers as direct methods. Generated files read like hand-written ones. What each op renders to: .. list-table:: :header-rows: 1 :widths: 24 38 38 * - Op - ``upgrade()`` - ``downgrade()`` * - ``add_column`` - ``ctx.enforce_column(...)`` with the new type (and the target doc, if any) - ``ctx.delete_column(...)`` * - ``rename_column`` - ``ctx.enforce_rename(old, new)`` - ``ctx.enforce_rename(new, old)`` * - ``delete_column`` - ``ctx.delete_column(...)`` - raises (re-adding gives a new, empty column) * - ``type_change`` - ``ctx.enforce_column(...)`` with the wider type, for a promotion only; otherwise raises - raises (Iceberg only widens types) * - ``add_partition_field`` - ``ctx.update_partition_spec(add_fields=[...])`` - ``ctx.update_partition_spec(remove_fields=[...])`` * - ``remove_partition_field`` - ``ctx.update_partition_spec(remove_fields=[...])`` - raises (the old source column and transform are not recorded) * - ``set_sort_order`` - ``ctx.replace_sort_order(...)`` - raises (the previous order is not recorded) * - ``set_property`` - ``ctx.set_table_properties(...)`` - restores the old value, or unsets a new key * - ``remove_property`` - ``ctx.unset_table_properties(...)`` - restores the old value; raises when it was not recorded "Raises" means the whole method raises ``NotImplementedError`` right after ``target_table`` is assigned, before any ``ctx`` call, with every generated step kept below as a comment, in order. So rolling back a generated migration that drops a column fails before any of its reversible steps run, and the version stays applied. CLI usage --------- Given an existing ``sales.orders`` table with columns ``order_id`` (long, required) and ``amount`` (double), and this target contract: .. code-block:: yaml # contracts/orders.yaml strict: true columns: - name: order_id type: long required: true - name: order_amount type: double the command below writes the next ``migrations/NNNN_*.py`` file with one ``ctx.enforce_rename(target_table, 'amount', 'order_amount')`` step and its reverse: .. code-block:: bash iceberg-ops --namespace sales revision \ -m "align orders with v2 schema" \ --autogenerate \ --target-table orders \ --target-schema-yaml contracts/orders.yaml \ --rename-hint amount:order_amount ``revision --autogenerate`` needs a catalog (``--catalog``, ``--catalog-name`` or the config file) and ``--namespace`` to load the existing table. These are group options, so they go before ``revision``. The migrations directory must already exist (``iceberg-ops init`` creates it); a ``migrations//`` subdirectory is used when present. Flags: * ``--autogenerate / --no-autogenerate`` -- enable diff mode. * ``--target-table`` -- existing table to compare against. Use a bare name; it is loaded from ``--namespace`` now and written into the file as typed. * ``--target-schema-yaml PATH`` -- load a ``SchemaContract`` from YAML. Contracts are read as UTF-8 and validated (``strict`` / ``required`` must be booleans); types include ``decimal(P,S)``, ``fixed[N]`` and ``timestamp_ns`` (see :doc:`schema_migrations`). * ``--target-schema-arrow module:attr`` -- import a ``pa.Schema`` (the attr must resolve to a ``pa.Schema``). The module must be importable from the current directory or the environment. * ``--rename-hint old:new`` -- repeatable; tells the differ to emit a rename rather than drop + add. Hints are validated: an unknown column, an old name that is still in the target, a new name that already exists (swaps and chains), or two hints with the same name are usage errors (exit 2). Plain ``iceberg-ops revision`` errors when any autogen flag is set without ``--autogenerate`` so misuse is caught early. After writing the file, the command lists the steps that make ``upgrade()`` or ``downgrade()`` raise, the table columns a non-strict contract kept, and nullability or doc differences it did not change. An invalid contract or a bad hint exits 2 without writing a file. Python API ---------- .. code-block:: python import tempfile from pathlib import Path import pyarrow as pa from iceberg_ops.autogen import schema_diff_to_ops, write_autogenerated_revision from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.runner import MigrationRunner # An existing table to compare against. catalog = CatalogFactory.build("memory") catalog.create_namespace("sales") existing_table = catalog.create_table( "sales.orders", pa.schema([pa.field("order_id", pa.int64(), nullable=False), ("amount", pa.float64())]), ) target = pa.schema( [ pa.field("order_id", pa.int64(), nullable=False), pa.field("order_amount", pa.float64()), # was 'amount' ] ) ops = schema_diff_to_ops( current=existing_table.schema(), target=target, rename_hints={"amount": "order_amount"}, ) migrations_dir = Path(tempfile.mkdtemp()) / "migrations" migrations_dir.mkdir() write_autogenerated_revision( version="0001", description="rename amount -> order_amount", ops=ops, target_table="orders", # bare: follows the namespace at migrate time out_path=migrations_dir / "0001_rename_amount.py", ) runner = MigrationRunner(catalog, namespace="sales", migrations_dir=str(migrations_dir)) runner.migrate() # renames amount -> order_amount runner.rollback() # renames it back Limitations ----------- * Renames need an explicit ``--rename-hint`` (``rename_hints=`` in Python) -- the differ has no other way to disambiguate a rename from a drop + add. * Steps that cannot run as generated (a type change that is not a promotion, nested types) or cannot be reversed (column drop, type widening, partition-field removal, sort-order replacement, property removal without a recorded old value) make the affected method raise ``NotImplementedError`` before any step runs. The generated calls stay in the file as comments for manual completion. * Type changes are one-way. Only ``int -> long``, ``float -> double`` and decimal precision widening at the same scale are emitted; the generated downgrade of a type change always raises. * Nested types (``Struct`` / ``List`` / ``Map``) are not rendered: a nested add or type change blocks the ``upgrade()`` until you write that step by hand. With a ``pa.Schema`` target, nested columns that already exist are not compared (a warning is logged, and the CLI prints one naming them) and a new nested column is rejected with ``ValueError``. * A generated ``upgrade()`` is idempotent and adopts columns that already exist, but ``downgrade()`` still removes every column the migration lists, even in a namespace whose table already had it. Review ``downgrade()`` before rolling back a namespace whose table differed from the one that was diffed. * Dropping a column that the partition spec, sort order or identifier fields reference makes ``upgrade()`` raise before any step runs. The CLI passes the table's spec, sort order and format version to the diff; from Python, pass ``spec=``, ``sort_order=`` and ``format_version=`` to ``schema_diff_to_ops`` (identifier fields are always checked). A changed column whose name contains ``.`` raises ``ValueError``, and so does, when ``format_version`` is given, a type the table's format version cannot hold. * A non-strict contract (the default) keeps table columns it does not list; set ``strict: true`` to drop them. * Nullability and doc changes on existing columns are reported, not emitted. New columns are added as optional and keep the target's doc. * Each step commits on its own: if a step fails, the earlier steps stay applied and the version is recorded as ``failed``. * ``spec_diff_to_ops`` rejects a partition field that keeps its name but changes its source column or transform, and ``sort_order_diff_to_ops`` rejects sort fields with a non-identity transform or a non-default null order (``ValueError``); write those steps by hand. * The CLI diffs columns only; use the Python diff functions for partition specs, sort orders and properties. See :doc:`api/autogen` for the full API. Architecture ============ Where iceberg-ops fits ---------------------- Many tools can change or maintain Apache Iceberg tables without a JVM. They take different approaches (as of November 2026): * **Declarative tools**, such as the Apache Iceberg Terraform provider and desired-state planners for Iceberg metadata, describe where a table should end up and compute the change for you. * **Engines** such as Spark, Trino and Flink run ``ALTER TABLE`` DDL, but they do not version or track it. * **iceberg-ops** follows the model of Alembic for relational databases. Each change is a numbered migration file with an ``upgrade()`` and a ``downgrade()``. The runner records which changes were applied, in what order and in which namespace, in an Iceberg history table inside your catalog. Autogenerate writes the downgrade where a step can be undone and makes it raise where it cannot. The maintenance helpers and the cross-table data mover ship in the same package but are secondary to the migration workflow. Capability surface ------------------ .. list-table:: :header-rows: 1 :widths: 20 80 * - Module - Responsibility * - ``iceberg_ops.migration`` - ``BaseMigration`` + ``MigrationContext`` -- the per-migration API surface (create, drop, schema evolution, partition spec, sort order, table properties). * - ``iceberg_ops.runner`` - Migration runner: discover, apply, roll back, stamp, plan, status and history for one namespace at a time. * - ``iceberg_ops.history`` - Iceberg-backed migration history table inside an ``_iceberg_ops`` meta-namespace, and ``ConcurrentMigrationError``. * - ``iceberg_ops.autogen`` - Schema, partition-spec, sort-order and property diffs turned into migration ops, and a renderer that writes a migration file for review. * - ``iceberg_ops.scaffolding`` - ``init_project`` + ``create_revision`` -- writes the ``iceberg-ops.yaml`` stub and ``NNNN_*.py`` migration skeletons. * - ``iceberg_ops.cli`` - Click-based ``iceberg-ops`` entry point. * - ``iceberg_ops.core`` - Catalog factory, schema diff / contract / evolution helpers, Arrow <-> Iceberg type translation, snapshot / upsert / time- travel utilities. * - ``iceberg_ops.data_migration`` - Cross-table streaming migrator with schema reconciliation, checkpoint-based resume, and three verify methods (experimental). * - ``iceberg_ops.maintenance`` - Basic maintenance helpers for self-managed tables: ``compact_small_files_streaming``, ``expire_snapshots``, ``remove_orphan_files``, ``table_health_report``. * - ``iceberg_ops.polaris`` - Apache Polaris policy client and dispatcher (experimental; snapshot expiry only). * - ``iceberg_ops.testing`` - Public ``FakeSession`` / ``FakeResponse`` for downstream integrators driving the Polaris client without HTTP. Catalog backends ---------------- Every API that takes a catalog accepts any PyIceberg ``Catalog``. ``CatalogFactory.build`` is a convenience for six kinds: .. list-table:: :header-rows: 1 :widths: 18 30 52 * - Kind - Install - Notes * - ``"sqlite"`` - ``pip install "iceberg-ops[sqlite]"`` - Local file-backed catalog for development and CI. Without ``db_path`` / ``warehouse_dir`` it uses ``./local_data/iceberg_catalog.db`` and ``./local_data/iceberg_warehouse`` in the current working directory at call time. The CLI's ``--catalog sqlite`` uses ``local_data/`` next to the config file, or ``--db-path`` / ``--warehouse``. * - ``"memory"`` - ``pip install "iceberg-ops[sqlite]"`` - In-memory SQLite (``sqlite://``) for tests and throw-away jobs. Data files go to ``warehouse_dir`` or a fresh temporary directory. Python API only: the CLI rejects it because its state is gone when the command exits. * - ``"glue"`` - ``pip install "iceberg-ops[glue]"`` - AWS Glue Data Catalog. * - ``"rest"`` - ``pip install iceberg-ops`` - REST catalogs such as Apache Polaris, Nessie and Lakekeeper. PyIceberg's REST client is part of the core install. * - ``"hive"`` - ``pip install "iceberg-ops[hive]"`` - Hive Metastore. * - ``"s3tables"`` - ``pip install "iceberg-ops[s3tables]"`` - AWS S3 Tables (REST + SigV4, pre-set); pass ``uri``, ``warehouse`` (the table bucket ARN) and ``region``. The remote kinds (``glue``, ``rest``, ``hive``, ``s3tables``) are loaded through :func:`pyiceberg.catalog.load_catalog` under a catalog name that defaults to the kind, so ``.pyiceberg.yaml`` and ``PYICEBERG_CATALOG____*`` environment variables apply; explicit keyword arguments win. From the CLI, ``--catalog-name NAME`` opens any catalog PyIceberg can load. Only the SQLite and in-memory kinds are integration-tested; see :doc:`compatibility`. Backends are *lazy-imported*: every backend lives behind a ``_lazy_import`` helper so ``import iceberg_ops`` succeeds even when none of the optional extras are installed. Calling ``CatalogFactory.build`` for a backend whose extra is missing raises an ``ImportError`` that names the ``pip install 'iceberg-ops[...]'`` line to run. Self-managed metadata namespace ------------------------------- The runner stores migration history as a real Iceberg table inside the user's catalog under a meta-namespace (default ``_iceberg_ops``): .. code-block:: text _iceberg_ops.migration_history version, namespace, description, applied_at, status, error The table holds one row per ``(version, namespace)`` with the latest status (``applied``, ``rolled_back`` or ``failed``). Earlier states are readable by time travel on the history table until its snapshots expire. The meta-namespace and the table are created by the first write; ``plan``, ``status`` and ``history`` do not create them. Keeping history in the catalog means it moves with the data and needs no separate database to back up. The Python API can use another meta-namespace through ``MigrationRunner(meta_namespace=...)``; the CLI always uses ``_iceberg_ops``. Concurrent runs against one namespace are not guarded: run one migration job per namespace. If two ``migrate`` runs overlap, both can execute ``upgrade()``; the second one to record history raises ``ConcurrentMigrationError`` and the history keeps one row per version and namespace. See :doc:`compatibility`. Namespace-scoped runner ----------------------- .. code-block:: python from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.runner import MigrationRunner catalog = CatalogFactory.build("sqlite", db_path="catalog.db", warehouse_dir="warehouse") sales_runner = MigrationRunner(catalog, namespace="sales") finance_runner = MigrationRunner(catalog, namespace="finance") Each namespace has its own history rows (one shared history table keyed by namespace) and its own pending / applied view. Migration files are namespace-agnostic: use bare table names so the same files apply against ``sales`` and ``finance`` without edits. Files written by ``revision --autogenerate`` keep a bare ``--target-table`` bare, so they follow the same rule. When ``migrations//`` exists, the runner for that namespace uses it instead of the flat layout. Local-filesystem-only operations -------------------------------- Two of the maintenance helpers (``compact_small_files_streaming`` and ``remove_orphan_files``) support local ``file://`` table locations only. Tables on ``s3://`` / ``gs://`` / ``abfs://`` raise ``NotImplementedError`` with the scheme named in the error message. ``expire_snapshots`` and ``table_health_report`` only touch metadata, so they are expected to work on any filesystem PyIceberg supports, but they are tested on local tables only. The maintenance helpers get bug fixes only; the roadmap is to hand these jobs to PyIceberg's own maintenance APIs as they land. Compatibility and limits ======================== This page collects what iceberg-ops 0.1 supports, what it has been tested with, and where its boundaries are. The guides link here instead of repeating the details. Python and dependencies ----------------------- .. list-table:: :header-rows: 1 :widths: 18 22 60 * - Requirement - Range - Notes * - Python - 3.12 or later - CI tests 3.12, 3.13 and 3.14 on Linux and 3.12 on Windows, plus one run at the lowest declared dependency versions. Python 3.15 is not tested; at release time PyIceberg 0.12.0 published no Python 3.15 wheels. * - PyIceberg - ``>=0.12.0,<0.13`` - Every catalog extra uses the same range. * - PyArrow - ``>=18`` - iceberg-ops does not use NumPy, but PyArrow 26 and later refuse to import next to NumPy 1.x. In an environment pinned to NumPy 1.x, also pin ``pyarrow<26``. * - Click - ``>=8.2`` - The CLI. * - PyYAML - ``>=6.0.1`` - Reads ``iceberg-ops.yaml`` and YAML schema contracts. * - tzdata - Windows only - Windows ships no time-zone database; ``timestamptz`` columns need one. Extras ------ .. list-table:: :header-rows: 1 :widths: 22 33 45 * - Extra - Installs - Needed for * - ``sqlite`` - ``pyiceberg[sql-sqlite]`` - The local SQLite catalog and the in-memory catalog. * - ``glue`` - ``pyiceberg[glue]`` - AWS Glue catalogs. * - ``hive`` - ``pyiceberg[hive]`` - Hive Metastore catalogs. * - ``s3tables`` - ``pyiceberg[rest-sigv4]`` - AWS S3 Tables (REST with SigV4 signing; brings boto3). * - ``transforms`` - ``pyiceberg[pyiceberg-core]`` - Writing data to tables partitioned by bucket, truncate, year, month, day or hour transforms (for example with ``migrate-data``). Schema migrations do not need it. * - ``polaris`` - ``requests``, ``urllib3`` - The experimental Apache Polaris policy client. * - ``rest``, ``contracts`` - nothing beyond the core install - Accepted as aliases. PyIceberg's REST client and PyYAML are part of the core install. ``docs`` and ``dev`` are for contributors; see `CONTRIBUTING.md `__. Catalogs -------- The Python API accepts any PyIceberg ``Catalog`` instance, and ``CatalogFactory.build`` is a convenience for the kinds below. The CLI selects a catalog with ``--catalog`` or ``--catalog-name`` (see :doc:`cli`). .. list-table:: :header-rows: 1 :widths: 17 29 29 25 * - Catalog - Python API - CLI - Tested * - SQLite (local file) - ``CatalogFactory.build("sqlite", db_path=..., warehouse_dir=...)`` - ``--catalog sqlite``, with ``--db-path`` / ``--warehouse`` or the ``local_data/`` default - Integration-tested * - In-memory - ``CatalogFactory.build("memory")`` - Not accepted: its state is gone when the command exits - Integration-tested (the test suite's default catalog) * - REST (Apache Polaris, Nessie, Lakekeeper, ...) - ``load_catalog(name)`` or ``CatalogFactory.build("rest", ...)`` - ``--catalog-name NAME`` or ``--catalog rest`` - Through PyIceberg; not tested against a live server * - AWS Glue - ``load_catalog(name)`` or ``CatalogFactory.build("glue", ...)`` - ``--catalog-name NAME`` or ``--catalog glue`` - Through PyIceberg; not tested against the live service * - Hive Metastore - ``load_catalog(name)`` or ``CatalogFactory.build("hive", ...)`` - ``--catalog-name NAME`` or ``--catalog hive`` - Through PyIceberg; not tested against a live metastore * - AWS S3 Tables - ``CatalogFactory.build("s3tables", uri=..., warehouse=..., region=...)`` - ``--catalog-name NAME`` or ``--catalog s3tables`` - Through PyIceberg; not tested against the live service * - Other PyIceberg catalog types - ``load_catalog(name)`` or your own instance - ``--catalog-name NAME`` - ``--catalog-name`` is tested with SQL catalogs declared in ``.pyiceberg.yaml`` How the CLI finds a catalog: * ``--catalog-name NAME`` (or ``catalog_name:`` in ``iceberg-ops.yaml``) calls ``pyiceberg.catalog.load_catalog(NAME)``. Its properties come from ``.pyiceberg.yaml`` (PyIceberg looks in ``$PYICEBERG_HOME``, then the home directory, then the current directory, and uses the first file it finds) and from ``PYICEBERG_CATALOG____`` environment variables. In variable names ``_`` stands for ``-``, so the catalog ``my-cat`` reads ``PYICEBERG_CATALOG__MY_CAT__URI``. * ``--catalog glue``, ``rest``, ``hive`` or ``s3tables`` loads the catalog named after the kind the same way, so ``PYICEBERG_CATALOG__REST__URI`` configures ``--catalog rest``. ``--warehouse`` sets the ``warehouse`` property; ``--db-path`` applies to ``sqlite`` only. * ``migrate-data`` and ``run-policies`` take their own catalog options; the catalog settings in ``iceberg-ops.yaml`` do not apply to them. Migration history is a table in the ``_iceberg_ops`` namespace of the managed catalog, created on the first write. The Python API can use another namespace through ``MigrationRunner(meta_namespace=...)``; the CLI always uses ``_iceberg_ops``, so a catalog that rejects that namespace name needs the Python API. Filesystems ----------- Schema migrations are metadata operations and go through PyIceberg's file IO, so they work wherever PyIceberg can read and write the table's metadata. Two maintenance helpers walk or rewrite files themselves and support the local filesystem only: .. list-table:: :header-rows: 1 :widths: 40 20 40 * - Helper - ``file://`` - ``s3://``, ``gs://``, ``abfs://`` * - ``table_health_report`` - yes - not tested (reads metadata only) * - ``expire_snapshots`` - yes - not tested (metadata only; data files stay on storage) * - ``compact_small_files_streaming`` - yes - no (``NotImplementedError``) * - ``remove_orphan_files`` - yes - no (``NotImplementedError``) See :doc:`maintenance` for each helper's behaviour. Format versions, delete files and deletion vectors -------------------------------------------------- * Every write iceberg-ops performs (history rows, compaction, data migration, upserts) goes through PyIceberg and is copy-on-write. iceberg-ops never writes position-delete files, equality-delete files or deletion vectors. * ``compact_small_files_streaming`` does not support tables that already contain delete files or deletion vectors, and it does not check for them. Do not run it on such tables. * ``remove_orphan_files`` keeps every file the table metadata references, including delete files, deletion-vector Puffin files and statistics files. * ``table_health_report`` does not account for delete files. * Format version 3 features are handled only as far as PyIceberg 0.12 supports them. iceberg-ops adds no v3-specific handling and has no v3 tests. Concurrency ----------- Concurrent runs against one namespace are not guarded: run one migration job per namespace at a time, for example one CI job or a concurrency group in your orchestrator. If two ``migrate`` runs overlap, both can execute ``upgrade()``; the second one to record history raises ``ConcurrentMigrationError``, and the history keeps one row per version and namespace. Because that run's ``upgrade()`` may already have run, keep ``upgrade()`` idempotent. Runners for different namespaces can share the history table. See :doc:`schema_migrations` for the details. A data-migration checkpoint file belongs to one job; do not share it between concurrent runs. A concurrent append to a partition being compacted fails the run with ``ValidationException`` and the staged files are removed. PyIceberg retries the commit after appends to other partitions; those appends are preserved and counted in ``files_after``. Experimental and frozen modules ------------------------------- iceberg-ops is Alpha (0.x). Any minor release before 1.0 may change the Python API, CLI options and the history-table or checkpoint layouts; the changelog lists breaking changes. .. list-table:: :header-rows: 1 :widths: 30 20 50 * - Module - Status - Scope * - Migrations (``migration``, ``runner``, ``history``, ``autogen``, ``scaffolding``, the CLI migration commands) - Alpha - The focus of development. * - ``iceberg_ops.data_migration``, ``migrate-data`` - Experimental - Documented fences in :doc:`data_migration`: ``overwrite_partition`` only in one commit on identity-partitioned destinations, resume refuses changed settings, the destination table must exist, and memory is not bounded by ``batch_rows``. * - ``iceberg_ops.maintenance`` - Basic, frozen - Helpers for self-managed tables on the local filesystem; bug fixes only. See :doc:`maintenance`. * - ``iceberg_ops.polaris``, ``run-policies`` - Experimental - Executes ``system.snapshot-expiry`` only. Its requests do not yet match the released Polaris policy API, so fetching policies from a released Polaris server fails. See :doc:`polaris`. * - ``iceberg_ops.testing`` - May change - Test doubles for the Polaris client. Upstream PyIceberg issues ------------------------- .. list-table:: :header-rows: 1 :widths: 25 75 * - Issue - Effect and what iceberg-ops does * - `apache/iceberg-python#4194 `__ - On PyIceberg 0.12 a concurrent ``Table.upsert`` can be retried into a duplicate row. iceberg-ops writes history rows with an overwrite scoped to ``(version, namespace)`` instead, so a conflicting write raises ``ConcurrentMigrationError``. Duplicate rows left by earlier runs are read as their latest status and removed by the next write for that version. Upserts in your own code are not covered by this guard. * - `apache/iceberg-python#3508 `__ - A composite-key upsert with a few thousand keys can crash the interpreter. iceberg-ops splits larger composite-key batches into sub-batches of 1,000 keys inside one commit. * - `apache/iceberg-python#2467 `__ - The first upsert after a schema-only change fails. The data migrator and ``upsert_streaming`` commit one empty snapshot first. * - `apache/iceberg-python#2407 `__ - PyIceberg's batch reader decodes whole files, so peak memory in compaction and data migration grows with the number and size of source files. iceberg-ops does not work around this. Windows ------- * ``tzdata`` is installed automatically. * Write a local warehouse in ``.pyiceberg.yaml`` as ``file://C:/path/to/warehouse``; PyArrow rejects the ``file:///C:/...`` form. The built-in ``sqlite`` kind builds its warehouse URI in the ``file://C:...`` form for you. Other engines may expect a different spelling of a local Windows path, so a table written this way may not open elsewhere. Cross-Table Data Migration ========================== The ``iceberg_ops.data_migration`` subpackage moves rows between two existing Iceberg tables. It handles schema reconciliation (renames, type coercions, missing-column policies), checkpoint-based resume, and three post-migration verify methods. Source and destination can live in different catalogs: pass any two PyIceberg tables. The destination table must already exist (create it with a schema migration first); the migrator never creates tables. .. note:: Data migration is a companion to the schema-migration workflow and is early-stage. Read the write-mode, resume and verify limits on this page before you point it at data you care about. Its API may change before 1.0. Three composable pieces ----------------------- ``SchemaReconciler`` builds a ``ReconciliationPlan`` from a (source, dest) schema pair and projects each ``pyarrow.RecordBatch`` against it. Handles column renames via ``column_mapping`` (every key must be a source column), per-column ``type_coercions`` (for a column the destination already has, the override must equal its type or be a legal Iceberg promotion of it), and two missing-column policies: * ``on_missing_in_source`` -- ``"error"`` (default), ``"null"``, or ``"skip"``. * ``on_missing_in_dest`` -- ``"error"`` (default) or ``"auto_add"``. Static lossy-cast detection (narrowing integer / float casts, float to integer, ``timestamp`` -> ``date``) surfaces warnings without blocking; data-dependent failures still raise inside ``project_batch``. ``CheckpointStore`` persists per-batch progress to a JSON file with atomic writes (``tmp`` file, ``fsync``, ``os.replace``). Disk I/O is lazy. A corrupt, unreadable or unsupported-version file raises ``CheckpointError`` (a ``RuntimeError``) with a remediation message. Single-writer model; one checkpoint file per migration job. ``DataMigrator`` wires the two together with a streaming loop: it reads the source at one snapshot, accumulates up to ``batch_rows`` source rows, projects them, and writes them in one destination commit. Pre-flight runs every check before it writes anything to the destination (auto-added columns included). Write modes ----------- * ``"append"`` -- one ``Table.append`` per batch. * ``"upsert"`` -- requires ``upsert_keys``; uses PyIceberg's native ``Table.upsert`` (see `Upserts`_). * ``"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 (dry runs included) rejects anything else with ``ValueError`` before any write: raise ``batch_rows`` or use ``append`` / ``upsert``. Memory ~~~~~~ ``batch_rows`` bounds the rows in each commit, not peak memory. 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 batch's keys. Checkpoints and resume ---------------------- Attach a ``CheckpointStore`` to make a run resumable. A crashed run resumes from the last committed batch on the next invocation, reading the source at the snapshot stored in the checkpoint (not the source's current snapshot), so rows added to the source after the crash are neither skipped nor duplicated. The checkpoint records the job's settings: source and destination table UUIDs, row filter, ``batch_rows``, write mode, upsert keys and reconciler settings. A resume raises ``CheckpointError`` before writing when: * the checkpoint already records a **completed** migration (a completed checkpoint is terminal: call ``CheckpointStore.reset()`` or delete the file to run again); * any recorded setting differs (for example a changed ``batch_rows`` or row filter, or a recreated destination table), or the checkpoint has no recorded settings; * the recorded source snapshot no longer exists (expired), or an explicit ``source_snapshot`` differs from it; * the destination's current history no longer holds a checkpointed batch's snapshots (it was rolled back): resuming would skip those rows. ``CheckpointStore.reset()`` forgets which batches are already in the destination. When batches were committed, every refusal names the job's ``dest_snapshot_before``. Roll the destination back to it with ``rollback_to_snapshot`` (or delete the job's rows) before you reset and re-run, or an append writes those rows again. Every destination snapshot a batch creates carries ``iceberg-ops.migration-job`` and ``iceberg-ops.migration-batch`` summary properties; the empty schema-stamp snapshot an upsert may commit first carries only ``iceberg-ops.migration-job``. On resume, a batch that was committed just before a crash but not yet checkpointed is found through them and recorded instead of being written twice. Upserts ------- * Two rows with the same key inside one batch raise ``ValueError`` (earlier batches stay committed). Across batches the later row wins, so the outcome for duplicate keys depends on ``batch_rows``. * ``on_missing_in_source="skip"`` is rejected for upserts (an upsert writes every destination column). With ``"null"``, matched keys get those columns overwritten with null; the plan carries a warning. * ``upsert_keys`` must be non-blank, unique, present in the projected destination schema, and mapped from a source column (a key the plan would fill with null is rejected before any write); they are rejected with any other write mode. * Composite-key batches of more than 1,000 rows are split into sub-batches inside one transaction, so each batch is still one catalog commit (works around `apache/iceberg-python#3508 `_). * When the destination's current snapshot predates its schema (for example right after ``auto_add``), one empty snapshot is committed first so the upsert can run (workaround for `apache/iceberg-python#2467 `_). * One batch is one catalog commit but can create up to three destination snapshots (three per 1,000-key sub-batch when a composite-key batch is split), plus the optional empty snapshot above. ``MigrationResult.new_snapshots`` lists all of them, and ``rows_inserted`` / ``rows_updated`` count the upsert's effect. To undo the run, roll back to ``MigrationResult.dest_snapshot_before`` (see `Undoing a migration`_), not to the snapshot before the last batch. Partitioned destinations ------------------------ Writing to a destination partitioned by a bucket, truncate, year, month, day or hour transform needs ``pyiceberg-core``: .. code-block:: bash pip install "iceberg-ops[transforms]" Without it, pre-flight raises ``ImportError`` before any write (a dry run reports it as a plan warning). Identity-partitioned and unpartitioned destinations do not need the extra. Verify methods -------------- After the migration completes, ``DataMigrator.verify(method=...)`` runs one of three consistency checks. The source is read at the snapshot the run migrated (an explicit ``source_snapshot``, else the snapshot the last ``run()`` read), so rows appended to the source afterwards do not count as a mismatch. The destination is read in full, so the result is meaningful for an append into an empty destination; account for pre-existing rows or upserts yourself. * ``"row_count"`` (default) -- compares row counts from manifest metadata; reads data files only where the row filter cannot be decided per file or delete files apply. * ``"partition_hash"`` -- per-partition SHA256 of sorted rows. Supports unpartitioned tables and identity-transform partition specs; non-identity partitions raise ``NotImplementedError`` (the CLI prints ``Verify error:`` and exits 1). Reads both tables in full. * ``"sample"`` -- compares the first ``sample_size`` rows of each side after sorting. Reliable only when both tables fit in the sample or the read order is the same on both sides; otherwise it can report a false mismatch. Each method returns a ``VerificationResult`` with ``method``, ``matches``, ``source_count`` / ``dest_count`` (``row_count`` and ``sample``), ``sample_size``, ``mismatched_partitions`` (``partition_hash``) and ``details`` (list of strings). Undoing a migration ------------------- ``MigrationResult.dest_snapshot_before`` is the destination's current snapshot before the run wrote anything (for a resumed run, the one recorded when the job first started); it is ``None`` when the destination had no snapshot. Pass it to ``iceberg_ops.core.utils.rollback_to_snapshot(table, snapshot_id)`` to put the destination back. Columns that ``auto_add`` added stay in the schema. After a rollback, call ``CheckpointStore.reset()`` before you run the job again: a resume after a rollback raises ``CheckpointError``. CLI usage --------- .. code-block:: bash iceberg-ops migrate-data \ --src-catalog sqlite --src-db-path src.db --src-warehouse src_wh \ --src-namespace sales --src-table orders \ --dst-catalog sqlite --dst-db-path dst.db --dst-warehouse dst_wh \ --dst-namespace warehouse --dst-table orders \ --batch-rows 50000 \ --column-mapping order_total:total_amount \ --on-missing-in-source null \ --on-missing-in-dest auto_add \ --checkpoint orders.checkpoint.json \ --verify row_count To write into another catalog, name it with ``--dst-catalog-name``. The name is resolved through PyIceberg's ``load_catalog``, so it is read from ``.pyiceberg.yaml`` or ``PYICEBERG_CATALOG____*`` environment variables: .. code-block:: yaml # .pyiceberg.yaml catalog: dest: type: rest uri: https://nessie.example.com/iceberg .. code-block:: bash iceberg-ops migrate-data \ --src-catalog sqlite --src-db-path src.db --src-warehouse src_wh \ --src-namespace sales --src-table orders \ --dst-catalog-name dest --dst-namespace warehouse --dst-table orders Key flags: * ``--src-catalog KIND`` or ``--src-catalog-name NAME`` -- one of the two is required. ``--src-db-path`` / ``--src-warehouse`` configure a sqlite source. * ``--dst-catalog`` / ``--dst-catalog-name`` -- the destination catalog. When neither is given, the destination is the source catalog (useful for cross-namespace migration in one catalog), and the source's ``--src-db-path`` / ``--src-warehouse`` carry over only to that same catalog. * ``--dst-table`` -- defaults to ``--src-table``. The table must already exist. * ``--row-filter`` -- PyIceberg row filter expression on the source scan. * ``--source-snapshot`` -- pin the read to a specific source snapshot ID. * ``--batch-rows`` -- source rows per commit (default 50000, at least 1). Bounds rows per commit, not peak memory. * ``--write-mode`` -- ``append`` (default) / ``upsert`` / ``overwrite_partition`` (identity-partitioned destination, one commit of at most ``--batch-rows`` rows; see `Write modes`_). * ``--upsert-keys`` -- comma-separated; required when ``--write-mode=upsert``. * ``--checkpoint`` -- path to the JSON checkpoint file. A second run with the checkpoint of a completed migration exits 1. * ``--column-mapping SRC:DST`` -- repeatable. * ``--dry-run`` -- pre-flight only: runs the checks and counts rows + bytes, does not write. * ``--verify {row_count|partition_hash|sample}`` -- run a post-migration check (skipped on a dry run). A mismatch exits 2. Errors from the migrator (a refused resume, a missing extra, a missing destination table, ...) print one ``Error:`` line and exit 1; set ``ICEBERG_OPS_TRACEBACK=1`` to see the traceback. Python API ---------- .. code-block:: python import os import tempfile import pyarrow as pa from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.data_migration import CheckpointStore, DataMigrator, SchemaReconciler work = tempfile.mkdtemp() # Source: a SQLite catalog with a populated sales.orders table. src_catalog = CatalogFactory.build( "sqlite", db_path=os.path.join(work, "src.db"), warehouse_dir=os.path.join(work, "src_wh"), ) src_catalog.create_namespace("sales") src = src_catalog.create_table( "sales.orders", pa.schema([("order_id", pa.int64()), ("order_total", pa.float64()), ("region", pa.string())]), ) src.append( pa.table( {"order_id": [1, 2, 3], "order_total": [9.5, 20.0, 7.25], "region": ["eu", "us", "eu"]} ) ) # Destination: a second catalog. The table must already exist. dst_catalog = CatalogFactory.build( "sqlite", db_path=os.path.join(work, "dst.db"), warehouse_dir=os.path.join(work, "dst_wh"), ) dst_catalog.create_namespace("warehouse") dst = dst_catalog.create_table( "warehouse.orders", pa.schema([("order_id", pa.int64()), ("total_amount", pa.float64()), ("channel", pa.string())]), ) reconciler = SchemaReconciler( column_mapping={"order_total": "total_amount"}, on_missing_in_source="null", # 'channel' has no source column on_missing_in_dest="auto_add", # 'region' is added to the destination ) migrator = DataMigrator( source=src, dest=dst, schema_reconciler=reconciler, batch_rows=50_000, write_mode="append", checkpoint=CheckpointStore(os.path.join(work, "orders.checkpoint.json")), ) result = migrator.run() print(f"moved {result.rows_moved:,} rows in {len(result.new_snapshots)} snapshots") print(f"columns added to the destination: {result.schema_changes}") verification = migrator.verify(method="row_count") assert verification.matches, verification.details Any PyIceberg catalog works on either side. For example, to load the destination from a catalog defined in ``.pyiceberg.yaml``: .. code-block:: python # illustrative: needs a reachable catalog named "dest" from pyiceberg.catalog import load_catalog dst_catalog = load_catalog("dest") # e.g. Nessie at https://nessie.example.com/iceberg dst = dst_catalog.load_table("warehouse.orders") Testing seam ------------ The ``iceberg_ops.testing`` module ships ``FakeSession`` / ``FakeResponse`` for integrators driving the Polaris client without HTTP. They are not needed for ``DataMigrator`` itself -- the data-migration tests run against real PyIceberg catalogs. See :doc:`api/data_migration` for the full reference. Storage Maintenance =================== The maintenance package wraps PyIceberg's lower-level primitives in four operations. They are basic helpers for self-managed tables on the local filesystem; delete files and v3 deletion vectors are out of scope. Catalog services that run table maintenance themselves (for example AWS S3 Tables) do not need them. These APIs may change before 1.0. The four helpers ---------------- * :py:func:`iceberg_ops.maintenance.compact_small_files_streaming` -- per-partition compaction. Reads each qualifying partition via ``Table.scan().to_arrow_batch_reader``, writes pre-staged Parquet files of roughly ``target_size_mb``, then commits one PyIceberg transaction (overwrite per touched partition + ``add_files``). Identity-transformed partitions (and unpartitioned tables) only; other partitions are skipped with a warning. Peak memory is **not** bounded by ``target_size_mb``: PyIceberg's batch reader decodes whole source files (`apache/iceberg-python#2407 `_, `apache/iceberg-python#4028 `_), so memory grows with the number and size of files per partition. * :py:func:`iceberg_ops.maintenance.expire_snapshots` -- max-age + min-count snapshot expiry via PyIceberg's ``Table.maintenance.expire_snapshots()``. All arguments after ``table`` are keyword-only. Branch and tag heads (including the current snapshot) are never expired; those old enough to expire are reported in ``protected_ids``. Unset arguments fall back to the table's ``history.expire.max-snapshot-age-ms`` and ``history.expire.min-snapshots-to-keep`` properties (as written by ``MigrationContext.set_snapshot_retention``), then to the Iceberg defaults of 5 days and 1 snapshot. ``min_snapshots`` below 1 or a negative age raise ``ValueError``. Expiry is metadata-only: data files of expired snapshots stay on storage until orphan cleanup reclaims them (``file://`` only). * :py:func:`iceberg_ops.maintenance.remove_orphan_files` -- recursive ``os.walk`` over a configurable sweep list (defaults to ``data/`` + ``metadata/``). **Defaults to** ``dry_run=True``: it reports and deletes nothing unless you pass ``dry_run=False``. A file is kept if any table metadata references it: manifest lists, manifests, every live manifest entry of every snapshot (data files, position- and equality-delete files, deletion-vector Puffin files), table and partition statistics files, and every metadata JSON. Matching is by basename and is conservative (a file is kept if any referenced path shares its name). Names are compared case-insensitively after Unicode NFC normalisation, because Windows and default macOS filesystems resolve them that way. The walk does not follow directory symlinks or Windows junctions, and a file whose real path lies outside the table location is never deleted. Only files with known Iceberg suffixes (``.parquet`` / ``.avro`` / ``.metadata.json`` / ``.puffin`` / ``.stats``) whose mtime predates the cutoff are candidates. Tables with ``gc.enabled=false`` are refused with ``ValueError`` (in dry-run too), a negative ``older_than_hours`` is rejected, sweep entries must be relative and stay inside the table location, and deletions that fail are reported in ``failed_paths``. Never run it against a location shared with another table. * :py:func:`iceberg_ops.maintenance.table_health_report` -- read-only diagnostic. File sizes are aggregated with ``pyarrow.compute``, but the input comes from ``inspect.files()``, which builds one Python dict per file, so cost grows with the file count. Returns a ``TableHealthReport`` with file counts and sizes, snapshot count, oldest-snapshot age, small-file count, and a list of human-readable recommendations. The examples below each build a small local table first, so they run as written; with your own table, skip that part. Compaction example ------------------ .. code-block:: python import pyarrow as pa from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.maintenance import compact_small_files_streaming # A local table with six small data files (one per append). catalog = CatalogFactory.build("memory") catalog.create_namespace("sales") table = catalog.create_table("sales.orders", pa.schema([("order_id", pa.int64())])) for i in range(6): table.append(pa.table({"order_id": [i]})) result = compact_small_files_streaming( table, target_size_mb=512, min_file_count=5, sort_by_iceberg_order=True, dry_run=False, ) print( f"compacted {result.files_rewritten} files; " f"{result.bytes_before:,} -> {result.bytes_after:,} bytes" ) If the commit fails with ``CommitFailedException`` or ``ValidationException``, nothing was committed and the staged files are removed. Any other commit error (for example ``CommitStateUnknownException``) means the commit may have landed, so the staged files are kept and the error is re-raised: refresh the table and check. If the commit did not land, ``remove_orphan_files`` reclaims the staged files once they are older than ``older_than_hours``. Snapshot expiry example ----------------------- .. code-block:: python import pyarrow as pa from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.maintenance import expire_snapshots # A local table with three snapshots. catalog = CatalogFactory.build("memory") catalog.create_namespace("sales") table = catalog.create_table("sales.orders", pa.schema([("order_id", pa.int64())])) for i in range(3): table.append(pa.table({"order_id": [i]})) result = expire_snapshots( table, max_age_hours=120, min_snapshots=2, dry_run=False, ) print(f"expired {result.expired_count}/{result.total_snapshots} snapshots") print(f"kept because a branch or tag points at them: {result.protected_ids}") Omit ``max_age_hours`` / ``min_snapshots`` to use the table's ``history.expire.*`` properties. ``max_age_hours`` accepts fractional hours. Orphan cleanup example ---------------------- .. code-block:: python import pyarrow as pa from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.maintenance import remove_orphan_files # A local table to sweep. catalog = CatalogFactory.build("memory") catalog.create_namespace("sales") table = catalog.create_table("sales.orders", pa.schema([("order_id", pa.int64())])) table.append(pa.table({"order_id": [1, 2, 3]})) # Report first (the default is dry_run=True) ... preview = remove_orphan_files(table, older_than_hours=72) print(f"would delete {preview.orphan_count} orphan files: {preview.orphan_paths[:10]}") # ... then delete explicitly. result = remove_orphan_files( table, older_than_hours=72, sweep=("data", "metadata"), dry_run=False, ) print(f"found {result.orphan_count} orphan files; failed: {result.failed_paths}") ``older_than_hours`` below 24 logs a warning: files written by an in-flight writer that has not committed yet look like orphans. Health report example --------------------- .. code-block:: python import pyarrow as pa from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.maintenance import table_health_report # A local table with a few small files. catalog = CatalogFactory.build("memory") catalog.create_namespace("sales") table = catalog.create_table("sales.orders", pa.schema([("order_id", pa.int64())])) for i in range(3): table.append(pa.table({"order_id": [i]})) report = table_health_report(table, small_file_threshold_mb=100) print(f"snapshots: {report.snapshot_count}") print(f"small files: {report.small_file_count}/{report.total_files}") for rec in report.recommendations: print(f" - {rec}") Known limitations ----------------- * A compaction run writes ``2N + 1`` snapshots for ``N`` touched partitions (a delete and an empty append per partition, then one append). Incremental readers see the rewritten rows as new data. * Qualifying partitions are rewritten whole, including files that are already large. * Compacted files use PyArrow's default codec (Snappy), not the table's configured ``write.parquet.compression-codec``. * A concurrent append to a partition being compacted fails the run with ``ValidationException`` and the staged files are removed. PyIceberg retries the commit after appends to other partitions; those appends are preserved and counted in ``files_after``. * Compaction supports Identity-transformed partitions (and unpartitioned tables) only; other partitions are skipped. * Snapshot expiry is metadata-only. It also commits without asserting the table's refs (a PyIceberg limitation), so do not create tags while an expiry runs. * Delete files and v3 deletion vectors are out of scope. Compaction and the health report read ``inspect.files()``, which lists delete files next to data files, and neither checks for them: do not run compaction on a table that has position or equality delete files or deletion vectors. Orphan cleanup keeps every delete file the metadata references. Filesystem support matrix ------------------------- .. list-table:: :header-rows: 1 :widths: 35 22 22 21 * - Helper - ``file://`` - ``s3://`` - ``gs://`` / ``abfs://`` * - ``table_health_report`` - yes - not tested - not tested * - ``expire_snapshots`` - yes - not tested (metadata only) - not tested (metadata only) * - ``compact_small_files_streaming`` - yes - no - no * - ``remove_orphan_files`` - yes - no - no The metadata-only helpers (``table_health_report`` and ``expire_snapshots``) only go through PyIceberg's metadata APIs, never listing or deleting objects themselves, so they are expected to work wherever PyIceberg can read the table; the test suite runs them on local tables only. The two helpers that *do* touch storage raise ``NotImplementedError`` with the scheme named in the message when the table location is not local. Plain paths, including Windows drive-letter paths such as a ``.pyiceberg.yaml`` warehouse of ``C:\warehouse``, count as local. See :doc:`api/maintenance` for the full reference. Polaris Policy Executor (experimental) ====================================== .. warning:: Experimental. Not verified against a live Apache Polaris server. In this release ``PolarisClient`` does not call the Polaris 1.x policy endpoints (``/api/catalog/polaris/v1/{prefix}/...``), so fetching from a real server fails with ``PolarisError``. ``dispatch_policy`` works on ``Policy`` objects you build yourself. Only ``system.snapshot-expiry`` is executed. The API may change in any release. Apache Polaris 1.x has a *policy framework*: a way to declare table-level maintenance rules (snapshot expiry, compaction, orphan-file removal) at the catalog level via REST endpoints. Polaris stores and resolves the policies; an external worker has to fetch the policies attached to a table and run the matching maintenance operations against it. The ``iceberg_ops.polaris`` subpackage is a prototype of that worker. Today it executes ``system.snapshot-expiry`` only, through :py:func:`iceberg_ops.maintenance.expire_snapshots`. Pieces ------ * :py:class:`iceberg_ops.polaris.PolarisClient` -- thin HTTP client for Polaris policy endpoints (paths unverified, see the warning above). Lazy-imports ``requests`` (``pip install "iceberg-ops[polaris]"``); supports OAuth2 client-credentials auth or a pre-fetched static bearer token. * :py:exc:`iceberg_ops.polaris.PolarisError` -- raised on HTTP failures or unparseable responses. * :py:class:`iceberg_ops.polaris.Policy` -- frozen dataclass with a parsed view of one Polaris policy. * :py:func:`iceberg_ops.polaris.parse_policy` -- decodes ``content`` (a JSON string per the Polaris policy API; an already-decoded object is accepted too). * :py:func:`iceberg_ops.polaris.fetch_table_policies` -- convenience wrapper around ``list_applicable_policies_for_table`` that parses the response into a list of ``Policy``. * :py:func:`iceberg_ops.polaris.dispatch_policy` / :py:func:`iceberg_ops.polaris.dispatch_policies` -- map one or many policies to the matching maintenance op against an Iceberg table. * :py:class:`iceberg_ops.polaris.DispatchResult` -- per-policy outcome (``policy_name``, ``policy_type``, ``op``, ``success``, ``skipped``, ``skip_reason``, ``details``, ``error``). Authentication -------------- * The bearer token -- a static ``token=`` or one fetched with ``oauth2_credentials=`` -- is fetched once and cached for the client's lifetime. There is no refresh on ``401`` or on ``expires_in``. * Polaris's default token lifetime is one hour (``PT1H``), so create one client per run. * With ``oauth2_credentials=`` the token URL is ``base_url + "/api/catalog/v1/oauth/tokens"``. That is Polaris's internal authentication endpoint: still served by Polaris 1.8.0, but deprecated in the Iceberg REST spec. * For external / OIDC identity providers, fetch the token yourself and pass it via ``token=`` (or ``--polaris-token`` on the CLI). * Token-endpoint errors never include the response's values: a response without ``access_token`` is reported by its key names and its ``error`` / ``error_description`` fields only. * Surrounding whitespace in ``token=`` (for example a trailing CR/LF from a token file) is stripped. A token that is empty or still contains whitespace or control characters raises ``ValueError`` without echoing it. * A request that fails before a response arrives raises ``PolarisError`` naming the exception type and URL; the bearer token and the client secret are replaced with ``***`` in the message. Policy types ------------ Every ``system.*`` policy that would run requires ``"enable": true`` (the JSON boolean). A missing ``enable``, ``false``, or a non-boolean such as ``"true"`` or ``1`` skips the policy. .. list-table:: :header-rows: 1 :widths: 30 70 * - Policy type - Behaviour * - ``system.snapshot-expiry`` - Runs ``expire_snapshots``. See the content keys below. * - ``system.metadata-compaction`` - Skipped: manifest rewrite is not implemented, and iceberg-ops no longer runs a data-file rewrite for this type. * - ``system.data-compaction`` - Skipped: not executed from policies in this release (compaction supports ``file://`` tables only). * - ``system.orphan-file-removal`` - Skipped: never executed from policy automation in this release. * - anything else - Skipped as unsupported, unless a custom handler is registered. Skipped policies are returned with ``DispatchResult(skipped=True, op=None, skip_reason=...)`` so a batch run can still process the rest. Snapshot-expiry content ~~~~~~~~~~~~~~~~~~~~~~~ The content follows the Polaris ``system.snapshot-expiry`` schema (version ``2025-02-03``): .. code-block:: json { "version": "2025-02-03", "enable": true, "config": { "max_snapshot_age_days": 2, "min_snapshot_to_keep": 1, "max_ref_age_days": 3 } } * Only ``version``, ``enable`` and ``config`` are allowed at the top level. Any other key (for example a legacy ``max_snapshot_age_ms``) fails the dispatch instead of running on defaults. A ``version`` other than ``2025-02-03`` fails too. * ``config.max_snapshot_age_days`` -- maximum snapshot age in days (fractional days allowed); passed as ``max_age_hours = days * 24``. * ``config.min_snapshot_to_keep`` -- integer >= 1; passed as ``min_snapshots``. * Both accept an int, a float, or a numeric string. Booleans, NaN, infinities and negative values fail the dispatch. * ``config.max_ref_age_days`` is ignored with a note: branch and tag heads are never expired. * Other ``config`` keys are ignored with a note in ``DispatchResult.details["notes"]``. * Missing keys are not passed, so ``expire_snapshots`` uses the table's ``history.expire.*`` properties, then its defaults (5 days, 1 snapshot). Custom policy types ------------------- Polaris cannot store custom policy types, so the ``handlers`` kwarg is for other policy sources, or for overriding the built-in behaviour of a type: .. code-block:: python import pyarrow as pa from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.polaris import Policy, dispatch_policy # A local table for the handler to act on. catalog = CatalogFactory.build("memory") catalog.create_namespace("sales") table = catalog.create_table("sales.orders", pa.schema([("order_id", pa.int64())])) table.append(pa.table({"order_id": [1, 2, 3]})) def handle_row_audit(table, policy, *, dry_run): # Do the work; the returned dict becomes DispatchResult.details. return {"rows": table.scan().count(), "dry_run": dry_run} policy = Policy(name="audit", policy_type="my.org.row-audit", content={}) result = dispatch_policy( table, policy, handlers={"my.org.row-audit": handle_row_audit}, ) print(result.op, result.success, result.details) The handler receives ``(table, policy, *, dry_run)`` and returns a dict of details (``None`` counts as ``{}``; any other non-dict fails the dispatch) or raises on failure. ``handlers`` keys take precedence over the built-in handling, including the always-skipped system types, but a handler for a ``system.*`` type still runs only with ``"enable": true``. CLI usage --------- .. code-block:: bash # illustrative: the policy fetch fails against a released Polaris server export ICEBERG_OPS_POLARIS_CLIENT_ID="$POLARIS_ID" export ICEBERG_OPS_POLARIS_CLIENT_SECRET="$POLARIS_SECRET" iceberg-ops run-policies \ --polaris-url https://polaris.example.com \ --polaris-catalog mycat \ --target-namespace sales --target-table orders \ --catalog sqlite --db-path catalog.db --warehouse warehouse \ --dry-run The command is experimental for the same reason as the client: against a released Polaris server the policy fetch fails. It prints a warning saying so on every run. * Credentials: provide exactly one of a bearer token (``ICEBERG_OPS_POLARIS_TOKEN`` or ``--polaris-token``) or the OAuth2 client-credentials pair (``ICEBERG_OPS_POLARIS_CLIENT_ID`` + ``ICEBERG_OPS_POLARIS_CLIENT_SECRET``, or ``--polaris-client-id`` + ``--polaris-client-secret``). Prefer the environment variables: command-line arguments are visible in process listings and CI logs. Auth validation runs before any HTTP or catalog work; a missing or doubled credential is a usage error (exit 2). * A ``--polaris-url`` that uses plain ``http`` to a host other than ``localhost`` / ``127.0.0.1`` / ``::1`` prints a warning, because credentials are then sent unencrypted. * The table the policies act on is loaded with ``--catalog`` (plus ``--db-path`` / ``--warehouse`` for sqlite) or with ``--catalog-name NAME`` for a catalog defined in ``.pyiceberg.yaml`` or ``PYICEBERG_CATALOG____*`` environment variables. One of the two is required. * ``--dry-run`` forwards ``dry_run=True`` to every maintenance op so the table isn't mutated. * Exit codes: 1 when the Polaris request fails, 2 when any policy dispatch fails (skipped policies do not count), 0 otherwise. Python API ---------- Build the policies yourself and dispatch them: .. code-block:: python import pyarrow as pa from iceberg_ops.core.catalog import CatalogFactory from iceberg_ops.polaris import Policy, dispatch_policies # A local table with a few snapshots. catalog = CatalogFactory.build("memory") catalog.create_namespace("sales") table = catalog.create_table("sales.orders", pa.schema([("order_id", pa.int64())])) for i in range(3): table.append(pa.table({"order_id": [i]})) policies = [ Policy( name="expire", policy_type="system.snapshot-expiry", content={ "enable": True, "config": {"max_snapshot_age_days": 7, "min_snapshot_to_keep": 5}, }, ), Policy( name="compact", policy_type="system.data-compaction", content={"enable": True}, ), # skipped: not executed in this release ] results = dispatch_policies(table, policies, dry_run=True) for r in results: if r.success: print(f"[ok] {r.policy_name} -> {r.op}") elif r.skipped: print(f"[skip] {r.policy_name}: {r.skip_reason}") else: print(f"[x] {r.policy_name}: {r.error}") Testing seam ------------ Callers driving custom Polaris workflows can plug a deterministic stub in via the ``session=`` kwarg: .. code-block:: python from iceberg_ops.polaris import PolarisClient from iceberg_ops.testing import FakeResponse, FakeSession sess = FakeSession() sess.register("GET", "/policies", FakeResponse(json_body={"policies": []})) client = PolarisClient( "https://polaris.example.com", "mycat", token="t", session=sess, ) client.list_policies("sales") # no real HTTP call print(sess.calls[0]["method"], sess.calls[0]["url"]) The ``FakeSession`` records every call in ``sess.calls`` and raises ``AssertionError`` on unmatched requests, so tests fail loudly rather than silently producing surprising output. See :doc:`api/polaris` for the full reference and :doc:`api/testing` for the public test seam. CLI Reference ============= iceberg-ops ships one console script, ``iceberg-ops``. Global options (catalog, namespace, paths, config file) go before the command; each command's own options go after it: .. code-block:: bash iceberg-ops [GLOBAL OPTIONS] COMMAND [OPTIONS] iceberg-ops --catalog sqlite --namespace sales migrate --target 0002 iceberg-ops --version The :ref:`command reference ` at the end of this page is generated from the installed Click commands, so it shows the same options and help text as ``iceberg-ops COMMAND --help``. The commands fall into three groups: * **Project commands** -- ``status``, ``current``, ``migrate``, ``rollback``, ``stamp``, ``history``, ``plan``. They need a catalog and ``--namespace``, from flags or the config file. * **Scaffolding** -- ``init`` and ``revision`` need no catalog. ``revision --autogenerate`` does, because it loads the existing table (see :doc:`autogenerate`). * **Standalone commands** -- ``migrate-data`` (see :doc:`data_migration`) and ``run-policies`` (experimental, see :doc:`polaris`) take their own catalog options; the global catalog options and the config file's catalog settings do not apply to them. Every project command except ``history`` imports and executes the migration files it discovers, with your catalog credentials, so ``plan`` and ``status`` are not safe on files you have not reviewed. ``revision --target-schema-arrow`` imports the module it names. See :ref:`migration-trust-model`. Choosing a catalog ------------------ ``--catalog sqlite`` A local SQLite catalog. Without ``--db-path`` / ``--warehouse`` it lives in ``local_data/iceberg_catalog.db`` and ``local_data/iceberg_warehouse`` next to the config file, or in the current directory when there is no config file. ``--catalog-name NAME`` A catalog defined in ``.pyiceberg.yaml`` or in ``PYICEBERG_CATALOG____*`` environment variables, opened with :func:`pyiceberg.catalog.load_catalog`. ``--catalog glue|rest|hive|s3tables`` Loads the catalog named after the kind in the same way, so ``PYICEBERG_CATALOG__REST__URI`` configures ``--catalog rest``. Combined with ``--catalog-name``, it opens that named catalog as that kind. The CLI does not accept the ``memory`` kind: its state is gone when the command exits. ``--db-path`` applies to ``sqlite`` only, and a named catalog cannot use the ``sqlite`` kind. Configuration file ------------------ By default the CLI reads ``iceberg-ops.yaml`` from the current directory; ``--config PATH`` reads another file. A ``--config`` path that does not exist is a usage error, except for ``init``, which creates it. ``iceberg-ops init`` writes a commented starter file; a minimal one looks like this: .. code-block:: yaml catalog: sqlite namespace: sales Rules: * The keys are ``catalog``, ``catalog_name``, ``namespace``, ``db_path``, ``warehouse`` and ``migrations_dir``. Every value must be a string, so quote numbers (``namespace: "2024"``). Unknown keys print a warning and are ignored. * Command-line flags override the file. A ``--catalog-name`` flag replaces the file's ``catalog`` and ``db_path``; a ``--catalog`` flag replaces the file's ``catalog_name``. The file's ``db_path`` and ``warehouse`` apply only when the flags select the same catalog as the file. * Setting both ``catalog: sqlite`` and ``catalog_name`` in the file is a usage error. To use a named catalog, delete the ``catalog: sqlite`` line that ``init`` writes. * Relative ``db_path``, ``migrations_dir`` and (for ``sqlite``) ``warehouse`` values resolve against the directory that holds the file. Relative paths given as flags resolve against the current directory. ``migrations/`` defaults to sitting next to the file. * The file is read as UTF-8; a byte-order mark is allowed. Exit codes ---------- .. list-table:: :header-rows: 1 :widths: 10 90 * - Code - Meaning * - ``0`` - Success. * - ``1`` - A runtime failure, reported as one ``Error:`` line instead of a traceback: the catalog cannot be opened (missing extra, missing property, connection, permission or server error), a table or namespace is missing, a commit conflicts, a migration fails or cannot be loaded, a ``downgrade()`` raises ``NotImplementedError``, a checkpoint is corrupt or already completed, a revision file already exists, or ``migrate-data --verify partition_hash`` runs on a destination with non-identity partitions (``Verify error:``). ``run-policies`` prints ``Polaris error:`` when the Polaris request fails. Set ``ICEBERG_OPS_TRACEBACK=1`` to see the original traceback instead of the ``Error:`` line. * - ``2`` - A usage error: an unknown or malformed option, a bad config file, a missing catalog or ``--namespace``, a missing migrations directory, a ``--target`` that names no discovered migration, ``rollback --steps`` combined with ``--target``, a source and destination that are the same table in ``migrate-data``, an invalid Polaris token in ``run-policies``, or, for ``revision --autogenerate``, a target table that does not exist or a target schema that cannot be read or compared. Also returned when ``migrate-data --verify`` reports a mismatch and when ``run-policies`` has at least one failed policy. ``--target`` takes a migration version with or without leading zeros (``2`` means ``0002``); ``rollback --target 0`` rolls back every applied migration. ``rollback`` undoes one migration by default; ``--steps`` must be at least 1. .. _cli-command-reference: Command reference ----------------- .. click:: iceberg_ops.cli:main :prog: iceberg-ops :nested: full API Reference ============= Reference for the importable Python API, generated from the NumPy-style docstrings. iceberg-ops is alpha software: until 1.0, any release may change these APIs. The schema-migration modules are the core of the project. The data migrator and the Apache Polaris policy executor are experimental, and the maintenance helpers are basic and frozen in scope; their pages say what each one does not cover. Schema migrations ----------------- .. toctree:: :maxdepth: 2 migration runner history scaffolding autogen cli Core helpers ------------ .. toctree:: :maxdepth: 2 catalog schema types utils Experimental ------------ .. toctree:: :maxdepth: 2 data_migration maintenance polaris Testing ------- .. toctree:: :maxdepth: 2 testing Migration Context ================= .. automodule:: iceberg_ops.migration :members: :show-inheritance: Migration Runner ================ .. automodule:: iceberg_ops.runner :members: :show-inheritance: Migration History ================= .. automodule:: iceberg_ops.history :members: :show-inheritance: Scaffolding =========== .. automodule:: iceberg_ops.scaffolding :members: Autogenerate ============ .. automodule:: iceberg_ops.autogen :members: :show-inheritance: CLI Module ========== The Click-based command-line entry point. The :doc:`/cli` page is the user-facing reference; this page documents the importable Python surface. .. automodule:: iceberg_ops.cli :members: Catalog Factory =============== .. automodule:: iceberg_ops.core.catalog :members: :show-inheritance: Schema Helpers ============== .. automodule:: iceberg_ops.core.schema :members: :show-inheritance: Arrow / Iceberg Type Bridge =========================== .. automodule:: iceberg_ops.core.types :members: Core Utilities ============== .. automodule:: iceberg_ops.core.utils :members: 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 :doc:`/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 ------- .. automodule:: iceberg_ops.data_migration :no-members: Schema Reconciler ----------------- .. automodule:: iceberg_ops.data_migration.reconciler :members: :show-inheritance: Checkpoint Store ---------------- .. automodule:: iceberg_ops.data_migration.checkpoint :members: :show-inheritance: Migrator -------- .. automodule:: iceberg_ops.data_migration.migrator :members: :show-inheritance: Maintenance =========== .. note:: **Basic, frozen in scope.** These are basic helpers for self-managed tables on the local filesystem: compaction and orphan cleanup support local (``file://``) table locations only, and delete files and v3 deletion vectors are out of scope. The helpers receive bug fixes only, and, like the rest of the Alpha API, may change in any release before 1.0. See :doc:`/maintenance` for the known limitations. Storage maintenance operations for Iceberg tables. Only needed when your catalog does not provide managed maintenance. The package re-exports its public functions and result dataclasses from the four submodules below; tests and downstream code can import the names from either the package (``from iceberg_ops.maintenance import compact_small_files_streaming``) or the submodule -- both bindings refer to the same object. Package ------- .. automodule:: iceberg_ops.maintenance :no-members: Compaction ---------- .. automodule:: iceberg_ops.maintenance.compaction :members: :undoc-members: Snapshot Expiry --------------- .. automodule:: iceberg_ops.maintenance.snapshots :members: :undoc-members: Orphan Cleanup -------------- .. automodule:: iceberg_ops.maintenance.orphans :members: :undoc-members: Health Report ------------- .. automodule:: iceberg_ops.maintenance.advisor :members: :undoc-members: Polaris ======= .. note:: **Experimental.** The Apache Polaris policy executor has not been verified against a live Polaris server. :class:`PolarisClient ` does not yet call the Polaris 1.x policy endpoints, so fetching policies from a released Polaris server fails. Of the built-in policy types only ``system.snapshot-expiry`` runs, and only with ``"enable": true``; the others are skipped unless you register a custom handler. The API may change in any release. The ``iceberg_ops.polaris`` package re-exports its public surface from the three submodules below. Tests and downstream code can import the names from either the package (``from iceberg_ops.polaris import PolarisClient``) or the submodule -- both bindings refer to the same object. See :doc:`/polaris` for the user-facing guide. Package ------- .. automodule:: iceberg_ops.polaris :no-members: Client ------ .. automodule:: iceberg_ops.polaris.client :members: :show-inheritance: Policies -------- .. automodule:: iceberg_ops.polaris.policies :members: :show-inheritance: Dispatcher ---------- .. automodule:: iceberg_ops.polaris.dispatcher :members: :show-inheritance: Testing ======= Public test helpers. ``FakeSession`` / ``FakeResponse`` are the recommended seam for downstream integrators driving the Polaris client without real HTTP. .. automodule:: iceberg_ops.testing :members: :show-inheritance: .. The changelog is maintained in CHANGELOG.md at the repository root and rendered here through the MyST parser, so the page title and per-release headings come from that file. # Changelog All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] ### Added - Project logo. The docs show it in the sidebar, in a light and a dark version, and use it as a favicon that follows the browser's light or dark theme. Links to the docs get a preview image (Open Graph), and the README shows the logo next to its introduction. ## [0.1.0] - 2026-11-08 First public release of `iceberg-ops`: Alembic-style schema migrations for Apache Iceberg, plus a cross-table data mover, basic maintenance helpers and an experimental Apache Polaris policy executor. Development builds from May 2026 were never published. The Changed and Fixed lists below describe how this release differs from them, for anyone who ran those builds from source. ### Added Schema migrations: - `BaseMigration` and `MigrationContext` for `migrations/NNNN_*.py` files with `upgrade()` / `downgrade()` and the `pre_upgrade`, `post_upgrade`, `pre_downgrade` and `post_downgrade` hooks. Context helpers: `create_table_if_not_exists` (with `partition_spec`, `sort_order`, `properties` and `location`, applied on create), `drop_table_if_exists`, `load_table`, `add_column`, `enforce_column` (with opt-in `legacy_archival`), `enforce_rename`, `delete_column`, `update_partition_spec`, `replace_sort_order`, `set_clustering_keys`, `set_table_properties`, `unset_table_properties`, `set_snapshot_retention` and `set_compaction_target_size`. Helpers take bare table names; the runner adds the namespace. - `MigrationRunner` with `migrate`, `rollback`, `stamp`, `plan`, `status` and `history` for one namespace at a time. Migration history is an Iceberg table, `_iceberg_ops.migration_history`, in the managed catalog; it keeps the latest status of each version per namespace. - `iceberg_ops.runner.normalize_version`: `"2"`, `"0002"` and `2` name the same version. - The `iceberg-ops` CLI: `init`, `revision`, `migrate`, `rollback`, `stamp`, `plan`, `status`, `current`, `history`, `migrate-data` and `run-policies`, plus `--version`. Settings come from flags or from `iceberg-ops.yaml` (keys `catalog`, `catalog_name`, `namespace`, `db_path`, `warehouse`, `migrations_dir`). - `--catalog-name NAME` and the config key `catalog_name` open any catalog defined in `.pyiceberg.yaml` or `PYICEBERG_CATALOG____*` environment variables. `migrate-data` has `--src-catalog-name` / `--dst-catalog-name` and `run-policies` has `--catalog-name`. - Autogenerate (`iceberg_ops.autogen`): `schema_diff_to_ops`, `spec_diff_to_ops`, `sort_order_diff_to_ops`, `properties_diff_to_ops`, `render_migration_body`, `render_migration_file` and `write_autogenerated_revision`, plus `iceberg-ops revision --autogenerate --target-table T` with `--target-schema-yaml FILE` or `--target-schema-arrow module:attr` and repeatable `--rename-hint old:new`. The CLI diffs columns; the Python API also diffs partition spec, sort order and table properties. Rename hints are validated, and new columns keep their target doc. - YAML schema contracts (`SchemaContract.from_yaml`) accept the Iceberg primitive types, including `decimal(P,S)`, `fixed[N]` and `timestamp_ns`, and are validated with clear errors. - Core helpers (`iceberg_ops.core`): `CatalogFactory.build` for the `sqlite`, `memory`, `glue`, `rest`, `hive` and `s3tables` kinds, `default_local_paths()`, Arrow-to-Iceberg type translation with `is_allowed_promotion(old, new)` (Arrow time, uuid, fixed-size binary, string_view, binary_view and dictionary types are mapped), schema diff, `SchemaContract`, `SchemaEvolutionHelper`, and snapshot, upsert, time-travel and field-history utilities. Companion modules: - Cross-table data migration, experimental (`iceberg_ops.data_migration`, `iceberg-ops migrate-data`): `SchemaReconciler`, `CheckpointStore` and `DataMigrator` with `append`, `upsert` and `overwrite_partition` write modes, checkpoint resume, dry runs and `row_count` / `partition_hash` / `sample` verification. `MigrationResult` reports `dest_snapshot_before` (a rollback target), `rows_inserted` and `rows_updated`; `CheckpointStore.job` exposes the recorded job settings. Every snapshot the migrator writes carries `iceberg-ops.migration-job` and `iceberg-ops.migration-batch` summary properties. - Basic maintenance helpers for self-managed tables (`iceberg_ops.maintenance`): `compact_small_files_streaming`, `expire_snapshots`, `remove_orphan_files` and `table_health_report`. `SnapshotExpiryResult.protected_ids` lists snapshots kept because a branch or tag points at them; `OrphanCleanupResult.failed_paths` lists orphans whose deletion failed. - Apache Polaris policy executor, experimental (`iceberg_ops.polaris`, `iceberg-ops run-policies`): `PolarisClient`, `Policy`, `parse_policy`, `fetch_table_policies`, `dispatch_policy`, `dispatch_policies` and a `handlers={...}` hook for custom policy types. Only `system.snapshot-expiry` is executed. - `iceberg_ops.testing`: `FakeResponse` and `FakeSession` test doubles for driving the Polaris client without HTTP. They may change before 1.0. Packaging and project: - Extras: `sqlite`, `glue`, `hive`, `s3tables`, `transforms` (needed to write to tables partitioned by bucket, truncate or time transforms) and `polaris`. `rest` and `contracts` are accepted but add nothing to the core install. - A `py.typed` marker, so type checkers use the package's annotations. - Documentation at https://namitkewat.github.io/iceberg-ops/, including a compatibility page and plain-text `llms.txt` / `llms-full.txt` exports of the docs. - `CONTRIBUTING.md`, `CODE_OF_CONDUCT.md` (Contributor Covenant 3.0), `SECURITY.md` (GitHub private vulnerability reporting), `ROADMAP.md`, issue forms and a pull request template. - `AGENTS.md`, a guide for coding agents working on the repository, and `agent-skills/iceberg-ops-migrations/SKILL.md`, a skill that teaches a coding agent to write, review and run iceberg-ops migrations. ### Changed Packaging and CLI: - The command-line tool is `iceberg-ops` only, and its config file is `iceberg-ops.yaml`. The earlier short script name and its config file name were dropped: that name belongs to an unrelated project. - PyIceberg is pinned to `>=0.12.0,<0.13` (was `>=0.11.0,<0.12`) and PyArrow to `>=18`. Python 3.14 is tested in CI. PyYAML is a core dependency, and `tzdata` is installed on Windows. - The `[maintenance]` extra was removed: it only installed boto3, which no module imported. `[s3tables]` installs `pyiceberg[rest-sigv4]`; `[rest]` and `[polaris]` no longer request a `pyiceberg[rest]` extra, which PyIceberg does not define. - Development status is Alpha. - Default local state: without `db_path` / `warehouse_dir`, `CatalogFactory.build("sqlite")` uses `./local_data/` in the current working directory at call time, and the CLI's sqlite catalog uses `local_data/` next to the config file (or in the current directory when there is none). `CatalogFactory.build("memory")` without `warehouse_dir` writes data files to a fresh temporary directory. - The `glue`, `rest`, `hive` and `s3tables` kinds are loaded through `pyiceberg.catalog.load_catalog`, so `.pyiceberg.yaml` and `PYICEBERG_CATALOG____*` variables apply; explicit arguments win. `db_path` with a remote kind raises `ValueError`, and the local kinds reject `uri` / `warehouse` keyword arguments. - The CLI no longer accepts the memory catalog kind, which forgot applied migrations after each command. Catalog-kind options are case-insensitive, and a typo is a usage error. - Relative paths in the config file (`db_path`, the sqlite `warehouse`, `migrations_dir`) resolve against the config file's directory, and `migrations/` defaults to sitting next to it. Paths given as flags stay relative to the current directory. - CLI exit codes: 2 for usage errors (bad flags or config, missing catalog, namespace or migrations directory, invalid `--target` / `--steps`); 1 for runtime failures, with a single `Error:` line. Set `ICEBERG_OPS_TRACEBACK=1` to see the traceback. - A missing migrations directory is an error (`FileNotFoundError` from the runner, a usage error in the CLI) instead of reporting that all migrations are up to date. An explicit `--config` path that does not exist is a usage error, except for `init`. Migrations: - A migration's `version` must equal its file's `NNNN` prefix, and two files with the same version are rejected. `revision` numbers new files by the same `NNNN_.py` rule the runner uses. - `plan`, `status` and `history` no longer create the `_iceberg_ops` namespace or the history table; the first write creates them. - `--target` / `target=` is compared as a version number: `2` means `0002`, and an unknown or malformed target raises `ValueError` (CLI: exit 2) before anything is applied. `rollback(target="0000")` rolls back every migration. `rollback --steps` must be 1 or more, defaults to 1 and cannot be combined with `--target`. - `enforce_column` adds a missing column, does nothing for an equivalent type (only a changed doc is written), updates a promotable primitive type in place, and otherwise raises `pyiceberg.exceptions.ValidationError` unless `legacy_archival=True`. Promotions follow Iceberg's rules, so string-to-binary and decimal scale changes are rejected. - `set_clustering_keys` sets only the Iceberg sort order, in one commit. It no longer writes `snowflake.cluster-by` or `delta.clustering` table properties. - `set_snapshot_retention` logs a warning when the catalog is AWS S3 Tables, which manages snapshot retention itself. - `ConcurrentMigrationError` is defined in `iceberg_ops.history` and is still importable from `iceberg_ops.runner`. - `revision` never overwrites an existing migration file (exit 1). `write_autogenerated_revision` refuses to overwrite one unless `overwrite=True`, and always writes LF line endings. Autogenerate: - A bare `--target-table` stays bare in the generated file, so the migration applies to whichever `--namespace` it runs against. A dotted name pins the file to that table and prints a warning. - A generated `upgrade()` or `downgrade()` raises `NotImplementedError` before any step runs when one of its steps cannot run or cannot be undone (a column drop, a type widening, a partition-field removal, a sort-order replacement, a property removal with an unknown old value). The generated calls are kept as comments. - Type-change downgrades are no longer generated as narrowing calls, since Iceberg only widens types; they raise `NotImplementedError`. - A non-strict `SchemaContract` (the default) keeps the table columns it does not list, and the CLI reports them. Set `strict: true` to drop them. Nullability and doc differences on existing columns are reported, not emitted. Data migration and maintenance: - `write_mode="overwrite_partition"` needs a destination partitioned by identity transforms only, and the migration must fit in one commit (at most `batch_rows` source rows after the row filter). Anything else is rejected before any write. - Checkpoints record the job's settings (source and destination table UUIDs, row filter, `batch_rows`, write mode, upsert keys and reconciler). Resuming with different settings, a recreated destination or a checkpoint without these settings raises `CheckpointError`, which is now a `RuntimeError`. A completed checkpoint is terminal: call `CheckpointStore.reset()` to run again. - `MigrationResult.new_snapshots` lists every destination snapshot the run created (up to three per upsert batch, or per 1,000-key sub-batch for composite keys), and `CheckpointStore.mark_committed` accepts a list of snapshot IDs. - `DataMigrator` rejects blank or repeated `upsert_keys`, `upsert_keys` with a write mode other than upsert, and upsert keys missing from the projected destination schema. An upsert with `on_missing_in_source="skip"` that would drop destination columns is rejected; with `"null"`, the plan warns that matched keys get those values overwritten with null. `SchemaReconciler.compute_plan` reports a `column_mapping` key that is not a source column. - `migrate-data --batch-rows` and `--sample-size` must be at least 1, `--upsert-keys` ignores empty entries, and a missing destination table explains that `migrate-data` does not create it. - `remove_orphan_files` defaults to `dry_run=True`; pass `dry_run=False` to delete. It refuses tables with `gc.enabled=false` (in a dry run too), rejects a negative `older_than_hours` and `sweep` entries that are absolute or escape the table location, and warns when `older_than_hours` is below 24. `deleted` is True only when at least one file was removed. - All `expire_snapshots` arguments after `table` are keyword-only, and `max_age_hours` accepts fractional hours. Omitted arguments fall back to the table's `history.expire.max-snapshot-age-ms` / `history.expire.min-snapshots-to-keep` properties (as written by `set_snapshot_retention`), then to the Iceberg defaults of 5 days and 1 snapshot. Invalid values raise `ValueError`. The fallback to PyIceberg's internal `ExpireSnapshots` class for releases older than 0.11 was removed. - `compact_small_files_streaming` lost its unused `batch_rows` parameter. Polaris (experimental): - Only `system.snapshot-expiry` is executed, and only with `"enable": true` (a JSON boolean). `system.metadata-compaction` no longer triggers a data-file rewrite; it, `system.data-compaction` and `system.orphan-file-removal` are reported as skipped. - `system.snapshot-expiry` reads the Polaris schema (`config.max_snapshot_age_days`, `config.min_snapshot_to_keep`; `max_ref_age_days` is ignored with a note). Legacy top-level `*_ms` keys, unknown top-level keys, unknown schema versions and invalid values fail the dispatch instead of running on defaults. - A custom handler returning `None` yields empty details; any other non-dict return marks the dispatch as failed. - `run-policies` is marked `[experimental]` and prints a warning when run, because its requests do not yet match the released Polaris policy API. ### Fixed Migrations and CLI: - Two concurrent `migrate` runs for the same namespace no longer leave duplicate history rows on PyIceberg 0.12 (apache/iceberg-python#4194). The run that records second raises `ConcurrentMigrationError`, and history keeps one row per version and namespace. Existing duplicate rows are read as their latest status and removed by the next write for that version. - When a migration's own commit conflicts with a concurrent commit, it raises `ConcurrentMigrationError` and records nothing, instead of overwriting the other run's history with `failed`. - `migrate --target 2` no longer applies every migration, `stamp --target 2` no longer stamps all of them, `rollback --target 1` is no longer a no-op, and `"2"` no longer also selects `0010`. - `rollback(steps=0)` and a negative `steps` raise `ValueError` instead of rolling back an unexpected number of migrations. - `rollback` checks that every migration file it needs exists before running any `downgrade()`. `status` lists history entries whose migration file is missing. - `enforce_column` is idempotent for struct, list and map columns. - `enforce_column(..., legacy_archival=True)` archives only on a real type incompatibility. Commit errors and failing contracts propagate instead of archiving the column and adding an empty replacement. - Archival, in `enforce_column` and in the `enforce_rename` collision path, refuses columns used by the partition spec, sort order or identifier fields. An archived required column becomes optional, so later appends no longer fail. If archival or the rename fails part way, the column is restored with its original doc and required flag; if that restore also fails, a `RuntimeError` is raised. - `enforce_rename(name, name)` is a no-op, and renaming across different parent structs raises `ValueError`. - `enforce_column` and `add_column` accept primitive PyArrow types such as `pa.float64()` and `pa.string()`. - Migration files that define a `@dataclass` (with `from __future__ import annotations`) load. Same-named files in different namespace directories no longer collide. A broken migration file raises an error that names the file, and the CLI prints it as one `Error:` line. - `create_table_if_not_exists` works for nested namespaces such as `sales.eu`, and returns the existing table if another process creates it first. - `set_table_properties` accepts a property key named `properties`. - `revision --autogenerate` honours `--catalog-name`, the config's `catalog_name` and the config-relative sqlite defaults. - Missing extras, missing tables, missing catalog properties, connection failures, permission or server errors, commit conflicts and corrupt checkpoints print a single `Error:` line and exit 1 instead of a traceback. - Config file problems (invalid YAML, a non-mapping top level, non-string values such as `namespace: 123`) are usage errors. Unknown keys print a warning. - Config and generated files are read and written as UTF-8 (a BOM is tolerated). `init` appends its block to `.gitignore` without rewriting the existing content, and honours `--config` and `--migrations-dir`. - Using the `s3tables` kind without boto3 says to install `iceberg-ops[s3tables]`. - `CatalogFactory.build(..., reset=True)` refuses to delete a filesystem root, the home directory, the working directory or one of its parents, or a database file that is not SQLite. - The `sqlite` and `memory` kinds reject a warehouse that looks like a URI and local paths containing `#` or `?`, which PyIceberg would misread, before anything is written. - The history table is created with `commit.retry.num-retries=20`, so runners for different namespaces sharing it no longer lose history writes to commit contention. - When recording a failure in history itself fails, the migration's own error is still the one raised; the history error is logged. - `history` sizes its columns to their contents, so full ISO-8601 timestamps no longer push DESCRIPTION out of line, and a multi-line error is printed on one line. - `create_table_if_not_exists` raises `ValueError` before creating anything when a `pa.Schema` is combined with a non-empty `partition_spec` or `sort_order`. - `SchemaContract.validate` compares struct, list and map types without regard to field IDs, and `enforce_rename` checks the contract when neither column exists. Autogenerate: - `revision --autogenerate` passes the table's partition spec, sort order and format version to the diff. Dropping a column that the spec, the sort order or the identifier fields use makes the generated `upgrade()` raise before any step runs, and a type the table's format version cannot hold is a usage error. Column names containing `.`, decimals with a scale outside `0..precision` or more than 38 digits, and `fixed[0]` are rejected too. Existing nested columns of a `--target-schema-arrow` target that were not compared are named in a warning instead of reporting no differences. - Partition and sort fields sourced from a nested column render as the full dotted path and bind to the nested field. - Autogenerate no longer emits type changes that Iceberg does not allow, such as a decimal scale change, string to long, or binary to string; such a step makes `upgrade()` raise before anything runs. - A generated `downgrade()` with an irreversible step no longer rolls back part of the migration and then fails. - Nested type changes (struct, list, map) in a `pyiceberg.schema.Schema` target (Python API) produce a file that compiles, with the step blocked for manual completion, instead of a syntax error. - Invalid YAML contracts, unknown types and bad rename hints make `revision --autogenerate` exit with a usage error. - Partition fields with non-identity transforms (day, bucket, ...) import their transform and run. Sort-order diffs compare transform and null order, and reject orders the generated call cannot express. - `--target-schema-arrow` modules are importable from the current directory when the CLI runs from the installed console script. Data migration: - `overwrite_partition` could delete rows written by an earlier batch of the same migration; it is now limited to one commit (see Changed). - A resume reads the source snapshot recorded in the checkpoint, so rows added to the source after a crash are neither skipped nor duplicated. `verify()` after a run compares against the snapshot that was migrated. - Re-running with a checkpoint that records a completed migration raises `CheckpointError` instead of duplicating and dropping rows. - A batch committed just before a crash, but not yet checkpointed, is not written twice on resume. - Pre-flight runs every check before it changes the destination, so auto-added columns are no longer committed when a later check fails. - Writing to a destination partitioned by bucket, truncate, year, month, day or hour without `pyiceberg-core` fails before any write with an `ImportError` naming `pip install 'iceberg-ops[transforms]'`. - The first upsert after a schema-only change (for example an auto-added column) no longer fails; an empty snapshot is committed first, as a workaround for apache/iceberg-python#2467. - Composite-key upserts with more than a few thousand keys per batch no longer crash the interpreter (apache/iceberg-python#3508). Large batches are split into sub-batches of 1,000 keys within one commit. - When the source schema changed after the pinned snapshot, the plan uses the schema at that snapshot. - `verify(method="partition_hash")` works when a partition field's name differs from its source column, and raises `NotImplementedError` for non-identity partitions instead of reporting a mismatch (the CLI prints `Verify error:` and exits 1). - A resume raises `CheckpointError` when the destination was rolled back past a checkpointed batch, instead of skipping that batch's rows. Every refusal that follows committed batches names the snapshot to roll back to before `reset()`. - Upsert keys that the plan would fill with null are rejected before any write. - `DataMigrator` and `migrate-data` refuse a source and destination that are the same table. - Checkpoint writes are fsync-ed before the atomic rename. Maintenance: - `remove_orphan_files` no longer deletes live position-delete, equality-delete and deletion-vector files, or registered table and partition statistics files. - `compact_small_files_streaming` no longer deletes its staged files when the commit outcome is unknown (for example `CommitStateUnknownException`), which could make a committed table unreadable. - `expire_snapshots` skips branch and tag heads (including the current snapshot after a rollback) instead of aborting, and dry runs select the same snapshots as real runs. A max age older than the epoch expires nothing instead of overflowing. - Compaction and orphan removal accept Windows drive-letter table locations. ### Security - Generated migration files cannot run injected code: the `-m` message and column, partition and property names and values are written as Python literals, and docstring text is escaped. - Polaris OAuth2 token-endpoint errors no longer echo the response payload, which could leak tokens to logs or stderr. The client strips surrounding whitespace from a token, rejects an empty token or one with inner whitespace or control characters (`run-policies` exits 2), and keeps the token and client secret out of connection-error messages. - A failed migration's history row holds only `": "`; the traceback is logged locally at `DEBUG` level, since everyone with catalog access can read the history table. - `remove_orphan_files` matches file names after Unicode NFC normalisation and case folding, does not follow directory symlinks or Windows junctions, and never deletes a file whose real path is outside the table root. - Compaction escapes identity-partition values the way PyIceberg does and refuses to stage files outside `/data`, so a partition value such as `../../x` cannot write elsewhere. - `run-policies` reads `ICEBERG_OPS_POLARIS_TOKEN`, `ICEBERG_OPS_POLARIS_CLIENT_ID` and `ICEBERG_OPS_POLARIS_CLIENT_SECRET` (preferred over flags, which show up in process listings) and warns when `--polaris-url` uses plain http to a non-loopback host. ### Known limitations - Concurrent runs for one namespace are not guarded, only detected: a second run can execute `upgrade()` before it fails to record history. Run one migration job per namespace at a time and keep `upgrade()` idempotent. - The history table keeps the latest status of each version per namespace, not one row per attempt. Earlier states are readable by time travel on the history table until its snapshots expire. - `status`, `current`, `plan` and `history` print only and exit 0 even when a migration is recorded as failed; there is no `check` command yet. The runner methods print their reports and return `None`. - CLI autogenerate diffs columns only. Autogenerate does not emit nullability or doc changes, blocks nested type changes from a `pyiceberg.schema.Schema` target, and does not compare existing nested columns of a `pyarrow.Schema` target (they are treated as unchanged; the CLI names them in a warning). Each generated step commits on its own, so a failing step leaves earlier steps applied. - Only SQLite and in-memory catalogs are integration-tested. Glue, REST, Hive and S3 Tables catalogs are reached through PyIceberg but have not been tested against live services. - Data migration: `overwrite_partition` is limited to one commit and identity-partitioned destinations; the destination table must already exist. `batch_rows` bounds the rows per commit, not peak memory, which grows with the number and size of source data files (apache/iceberg-python#2407); an upsert also loads the matched destination rows. Duplicate keys within one batch raise `ValueError`, and across batches the later row wins. `verify="sample"` can report a false mismatch on larger tables, and `partition_hash` needs an identity-partitioned or unpartitioned destination and reads both tables in full. - Maintenance: compaction and orphan removal support the local filesystem only. Compaction supports identity partitions and unpartitioned tables (other partitions are skipped), rewrites qualifying partitions whole, writes `2N + 1` snapshots for `N` partitions, uses PyArrow's default codec, fails when a concurrent append lands in a partition being compacted, and its memory is not bounded by `target_size_mb` (apache/iceberg-python#2407). Delete files and deletion vectors are out of scope for compaction (there is no guard) and the health report. Snapshot expiry is metadata-only. - The Polaris executor does not call the Polaris 1.x policy endpoints, so fetching policies from a released Polaris server fails. It has not been validated against a live server. - Python 3.15 is not tested. At release time PyIceberg 0.12.0 published no Python 3.15 wheels. - PyArrow 26 and later refuse to import next to NumPy 1.x. iceberg-ops does not use NumPy; in an environment pinned to NumPy 1.x, also pin `pyarrow<26`. - On Windows, write a local warehouse in `.pyiceberg.yaml` as `file://C:/path`; PyArrow rejects the `file:///C:/path` form. [Unreleased]: https://github.com/namitkewat/iceberg-ops/compare/v0.1.0...HEAD [0.1.0]: https://github.com/namitkewat/iceberg-ops/releases/tag/v0.1.0 :parser: myst_parser.sphinx_