Schema Helpers¶

iceberg_ops.core.schema¶

Schema evolution, diffing, and contract validation for PyIceberg tables.

Public API¶

SchemaEvolutionHelper: Idempotent column enforcement and renaming with collision handling. Uses Python logging, compensating transactions on step 2 failures, and optional per-call SchemaContract checks.
diff_schemas(schema_a, schema_b) -> SchemaDiff: Compare two Iceberg schemas by stable field ID and return a structured diff (added / removed / renamed / type-changed / doc-changed fields).
SchemaContract: Declarative column spec. Call contract.enforce(table) before every append to catch schema drift early in the pipeline.
ColumnSpec, FieldChange, SchemaDiff, SchemaContractViolation: Supporting data-classes and exceptions.

class iceberg_ops.core.schema.FieldChange(field_id, name, change_type, old_name=None, old_type=None, new_type=None, old_doc=None, new_doc=None)[source]¶

Bases: object

One field-level difference between two schema versions.

Parameters:

field_id (int | None)
name (str)
change_type (Literal['added', 'removed', 'type_changed', 'renamed', 'doc_changed'])
old_name (str | None, default: None)
old_type (str | None, default: None)
new_type (str | None, default: None)
old_doc (str | None, default: None)
new_doc (str | None, default: None)

field_id: int | None¶

name: str¶

change_type: Literal['added', 'removed', 'type_changed', 'renamed', 'doc_changed']¶

old_name: str | None = None¶

old_type: str | None = None¶

new_type: str | None = None¶

old_doc: str | None = None¶

new_doc: str | None = None¶

class iceberg_ops.core.schema.SchemaDiff(added=list(), removed=list(), changed=list())[source]¶

Bases: object

The result of comparing two Iceberg schemas.

Variables:

added (list[FieldChange]) – Fields new in schema_b (not present in schema_a).
removed (list[FieldChange]) – Fields dropped from schema_a (not present in schema_b).
changed (list[FieldChange]) – Fields present in both with a different name, type, or doc.

Parameters:

added (list[FieldChange], default: list())
removed (list[FieldChange], default: list())
changed (list[FieldChange], default: list())

added: list[FieldChange]¶

removed: list[FieldChange]¶

changed: list[FieldChange]¶

property is_empty: bool¶: True when both schemas are structurally identical.

summary()[source]¶

Return a multi-line human-readable diff report.

Returns:: "Schemas are identical." for an empty diff, otherwise one Added / Removed / Changed section per non-empty list, one line per field change.
Return type:: str

iceberg_ops.core.schema.diff_schemas(schema_a, schema_b)[source]¶

Compare two Iceberg schemas and return a structured SchemaDiff.

Comparison uses Iceberg field IDs (not column names) as the stable identity key. This correctly identifies renames (same ID, different name) rather than treating them as a remove + add pair.

Parameters:

schema_a (Schema) – The “before” / baseline schema.
schema_b (Schema) – The “after” / target schema.

Returns:

Structured diff with added, removed, and changed lists.

Return type:

SchemaDiff

Example

from iceberg_ops.core.schema import diff_schemas
diff = diff_schemas(table.metadata.schemas[0], table.schema())
print(diff.summary())

class iceberg_ops.core.schema.ColumnSpec(name, type, required=False, doc=None)[source]¶

Bases: object

Declaration of one column in a SchemaContract.

Variables:

name (str) – Top-level column name.
type (ice_types.IcebergType) – Expected Iceberg type.
required (bool) – For SchemaContract.validate(), the column must be present. Autogenerate maps it to the field’s required flag for the diff, but never adds a NOT NULL constraint to an existing table.
doc (str or None) – Column description. Autogenerate sets it on newly added columns.

Parameters:

name (str)
type (IcebergType)
required (bool, default: False)
doc (str | None, default: None)

name: str¶

type: IcebergType¶

required: bool = False¶

doc: str | None = None¶

exception iceberg_ops.core.schema.SchemaContractViolation(violations)[source]¶

Bases: Exception

Raised when a table does not conform to its declared contract.

Parameters:: violations (list[str])

class iceberg_ops.core.schema.SchemaContract(columns, strict=False)[source]¶

Bases: object

Declares the expected schema for an Iceberg table.

Validate before every table.append() to detect schema drift in the CI/CD pipeline rather than at query time.

Parameters:

columns (list[ColumnSpec]) – The expected column definitions.
strict (bool, default: False) – True → the table must have exactly these columns and no others; used as an autogenerate target, columns missing from the contract are dropped (delete_column). False (default) → extra columns in the table are silently allowed (useful when the schema has legacy-archived columns); used as an autogenerate target, columns missing from the contract are kept and only the listed columns are reconciled.

Example

contract = SchemaContract([
    ColumnSpec("order_id",    ice_types.LongType(),   required=True),
    ColumnSpec("order_total", ice_types.DoubleType(), required=False),
])
contract.enforce(table)   # raises SchemaContractViolation on mismatch

Type comparison¶

Types are compared structurally: field ids, field order and nested docs are ignored, so a struct, list or map column matches a contract built with different field ids. A type that would only be reached by a promotion (int where the contract says long) is still reported as a violation.

validate(table)[source]¶

Check table’s current schema against this contract.

Parameters:: table (Table) – The table to check; it is refreshed first.
Returns:: Violation messages. Empty list means the table conforms.
Return type:: list[str]

enforce(table)[source]¶

Like validate(), but raises SchemaContractViolation on failure.

Call this immediately before table.append(data) to make contract violations a hard error rather than a silent drift.

Parameters:: table (Table) – The table to check; it is refreshed first.
Raises:: SchemaContractViolation – If the table does not conform; its violations attribute lists every issue.
Return type:: None

classmethod from_yaml(path)[source]¶

Load a schema contract from a YAML file.

The file must hold a mapping with a non-empty columns list and an optional strict boolean. Each column entry is a mapping with string name and type keys, an optional boolean required and an optional string doc. Column names must be unique. The file is read as UTF-8 (a leading byte-order mark is ignored).

type is case-insensitive and accepts boolean, int / integer, long, float, double, string, binary, date, time, timestamp, timestamptz, uuid and the parameterised Iceberg primitives decimal(P,S) (1 <= P <= 38, 0 <= S <= P) and fixed[N] (N >= 1), plus the other Iceberg primitive type strings such as timestamp_ns. Format-version 3 types (timestamp_ns, timestamptz_ns, unknown, geometry, geography) are parsed here; autogenerate refuses them for a format-version 2 table. Nested types are not supported.

Parameters:

path (str) – Path to the YAML contract file.

Returns:

A contract instance ready for enforce() or validate().

Return type:

SchemaContract

Raises:

ImportError – If pyyaml is not installed.
OSError – If the YAML file cannot be read (FileNotFoundError when it does not exist).
ValueError – If the file is not valid YAML, does not have the layout described above, or names an unknown or nested column type.

Example

contract = SchemaContract.from_yaml("contracts/orders.yaml")
contract.enforce(table)

class iceberg_ops.core.schema.SchemaEvolutionHelper(table)[source]¶

Bases: object

Safe, idempotent schema-evolution wrapper around a PyIceberg Table.

Key behaviours¶

Uses Python logging instead of print().
enforce_column() adds a compensating transaction. If step 2 (add new column) fails after step 1 (rename to legacy) has committed, the helper attempts to rename back automatically.
enforce_rename() uses the same compensating-transaction pattern when it has to archive a colliding column first.
Both methods accept an optional contract argument to run a SchemaContract.enforce() check after the operation completes.

Known gap¶

PyIceberg (as of 0.12) does not support rename + add-with-same-name in a single update_schema() commit, so the two-step approach remains. The compensating transaction significantly reduces (but cannot eliminate) the risk of a half-migrated table after a hard crash.

param table:: The active PyIceberg table to evolve.
type table:: Table

get_arrow_schema()[source]¶

Return the current Iceberg schema as a PyArrow schema.

Returns:: The table’s current schema, read after a refresh.
Return type:: Schema

get_iceberg_schema()[source]¶

Return the current Iceberg schema object.

Returns:: The table’s current schema, read after a refresh.
Return type:: Schema

enforce_column(col_name, new_type, doc=None, contract=None, *, legacy_archival=False)[source]¶

Ensure col_name exists with new_type.

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

Behaviour matrix¶

State	Action
Column absent	Add new column.
Column exists, equivalent type (including identical struct / list / map types)	No-op (only a changed doc is written). Struct field ids, field order and nested docs are ignored.
Column exists, promotable primitive type	`update_column` in place (Iceberg spec promotions only: `int` to `long`, `float` to `double`, wider decimal precision).
Column exists, incompatible type	If `legacy_archival=False` (default), `ValidationError` is raised and the table is left untouched. If `legacy_archival=True`, step 1 renames the existing column to `<col>__legacy_N` and step 2 adds a fresh one. A step 2 failure reverses step 1 (compensating transaction).

Archival runs only on a genuine type incompatibility: commit errors and contract violations propagate without archiving. The archived column keeps its field id and data, is made optional (so writers no longer have to supply it) and gets an [ARCHIVED] doc. A column referenced by the current partition spec, the sort order or the identifier fields is never archived, because those references would silently move to the legacy column.

param col_name:: The canonical column name to enforce. Use dotted paths for nested struct fields (e.g. "address.state").
type col_name:: str
param new_type:: The intended data type. A primitive PyArrow type (e.g. pa.float64()) is translated to its Iceberg equivalent.
type new_type:: IcebergType | DataType
param doc:: Human-readable description stored in the Iceberg schema catalog.
type doc:: str | None, default: None
param contract:: If provided, contract.enforce(table) is called once after the operation completes to verify the full schema still conforms.
type contract:: SchemaContract | None, default: None
param legacy_archival:: Opt-in to the archive-and-recreate flow on incompatible type change. Default False.
type legacy_archival:: bool, default: False
raises pyiceberg.exceptions.ValidationError:: If the column exists with an incompatible type and legacy_archival is False. Nested types cannot change in place; evolve their sub-fields with dotted paths.
raises ValueError:: If col_name is a dotted path and the parent struct does not exist, or archival is requested for a column referenced by the partition spec, sort order or identifier fields.
raises SchemaContractViolation:: If contract is given and the table does not conform.
raises RuntimeError:: If the compensating transaction also fails after a step 2 error (only possible when legacy_archival=True).

Return type:: None

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

Rename old_name → new_name idempotently.

Supports dotted paths for nested struct fields. Both names must share the same parent struct (e.g. "address.zip_code" → "address.postal_code").

Behaviour matrix¶

State	Action
old_name == new_name	No-op.
Neither column exists	No-op (warning logged); contract is still checked.
Only new_name exists	Already done — optionally update doc.
Only old_name exists	Simple rename.
Both exist (collision)	Archive new_name → `__legacy_N`, then rename old_name → new_name. If the rename fails, the archive is renamed back (compensating transaction).

On a collision the archived column keeps its field id and data, is made optional (so writers no longer have to supply it) and gets an [ARCHIVED] doc, as with enforce_column() archival. A blocking new_name that the current partition spec, the sort order or the identifier fields reference is never archived, because those references would silently move to the legacy column.

param old_name:: Current column name. Use dotted paths for nested struct fields.
type old_name:: str
param new_name:: Desired column name. Use dotted paths for nested struct fields.
type new_name:: str
param doc:: Documentation string to attach to the renamed column.
type doc:: str | None, default: None
param contract:: Post-operation contract check.
type contract:: SchemaContract | None, default: None
raises ValueError:: If the names do not share a parent struct, the parent struct of a dotted old_name does not exist, or a colliding new_name is referenced by the partition spec, sort order or identifier fields. The table is left untouched.
raises RuntimeError:: If the rename fails after a collision archive and renaming the archive back also fails.
raises SchemaContractViolation:: If contract is given and the table does not conform.

Return type:: None