io.IcebergSink

Stores a node in an Iceberg table, exactly once per state:

Usage

io.IcebergSink(
    catalog,
    identifier,
    *,
    app_id=None,
    name=None,
    location=None,
    partition_by=None,
    properties=None,
    compact_every=None,
    snapshot_retention=None,
    min_snapshots=1,
    reproducible=None
)

node.write(IcebergSink(catalog, identifier)), written by every refresh.

After the first write, each state is one snapshot the size of the change: position delete files for the rows that went (one copy each, so duplicates are exact) and a data file of the rows that came. In a partitioned table, the deletes and the new rows go to the partitions the change touches, and the other partitions keep their files. A new sink on an existing table, or a table that does not hold the rows the sink wrote, is overwritten. Every snapshot carries the contract’s plan hash, the epoch (sum of source versions) and the app id; a state whose epoch the last snapshot of this app already has is skipped, unless another plan wrote it: then the table is overwritten, schema included. The contract is also the table property sverify.contract, and columns the view proves NOT NULL are required. name is the table’s name in the plans that read the stored node (default: the last part of identifier); app_id defaults to sverify:<name>.

The node write returns reads the table as an IcebergSource and builds on the contract the stored node proved. A restarted job that builds the same nodes resumes from the table on its first refresh. The refresh’s report includes what each write did (SinkWrite). It raises ValueError if the table holds this state’s epoch from other contents of a StaticSource.

The first write creates the table with location, the partition fields of partition_by and the table properties beside the sink’s own, in one commit (#468). A catalog without a default location, such as Unity Catalog’s Iceberg REST catalog on a schema without a managed storage root, needs location. They only apply to a table the sink creates: an overwrite keeps the table’s location, spec and properties, and a sink on an existing table that differs from them warns once (IcebergSink.create_options_ignored).

Every write adds a snapshot and small files: a refresh that deletes rows adds a position delete file and a data file, which readers open and the sink’s next write searches. compact_every compacts the table (maintain) after every so many writes: in each partition, it rewrites the data files with position deletes, and the small files where there are two or more, as data files of their live rows, and drops the delete files no remaining data file reads, in one replace snapshot. With snapshot_retention, it then expires the snapshots older than that (keeping the min_snapshots newest and the sink’s own last snapshot) and deletes the files only they referenced. A compaction snapshot carries the contract, and a restarted job resumes past it, as it does past another writer’s compaction. pyiceberg rewrites the table’s metadata file on every commit; the table properties write.metadata.delete-after-commit.enabled and write.metadata.previous-versions-max (properties) bound those.

Parameters

catalog: Catalog: A pyiceberg catalog.
identifier: str: The sink table’s identifier; the first write creates it.
app_id: str | None = None: The application id recorded in each snapshot. Default: "sverify:<name>".
name: str | None = None: The table’s name in the plans that read the stored node. Default: the last part of identifier.
location: str | None = None: The location of the table the first write creates. Default: None, the catalog’s default.
partition_by: Sequence[PartitionField] | None = None: The partition fields of the table the first write creates, each (column, transform) or (column, transform, name); a transform is "identity", "year", "month", "day", "hour", "bucket[N]", "truncate[W]" or a pyiceberg Transform. Default: None, unpartitioned.
properties: Mapping[str, str] | None = None: Table properties of the table the first write creates, such as "write.target-file-size-bytes", besides the sink’s own (sverify.*). Default: None.
compact_every: int | None = None: Compact the table after every this many writes of this process (maintain). Default: None, never.
snapshot_retention: Duration | None = None: With compact_every, then expire the snapshots older than this and delete the files only they referenced (maintain(snapshot_retention=...)). Nodes reading the stored node from an expired snapshot read the whole table, so keep it longer than the gap between their refreshes. Default: None, no expiry.
min_snapshots: int = 1: The newest snapshots of the table’s history an expiry keeps, however old. Default: 1.
reproducible: Observation | None = None: What a rerun of a state must reproduce (sv.equivalent(...), #444 D20). Default: rows as a multiset, doubles within the plan’s bound (sv.equivalent(floats="plan")), other values exact; sv.equivalent() asks for the same bits.

Raises

ValueError: A transform of partition_by is unknown, or properties sets a sverify.* property, or compact_every or min_snapshots is below 1.

Attributes

catalog: The catalog.
identifier: The sink table’s identifier.
name: str: The table’s name in the plans that read the stored node.
table_location: location, or None.
partition_by: The partition fields, as (column, Transform, name or None).
properties: properties, or an empty dict.
compact_every: Writes between compactions, or None.
snapshot_retention: The expiry’s retention after a compaction, or None.
min_snapshots: The newest snapshots an expiry keeps.

Examples

A sink partitioned by day, at a location of its own:

IcebergSink(
    catalog,
    "lake.daily",
    location="s3://lake/daily",
    partition_by=[("day", "day"), ("host", "identity")],
    properties={"write.target-file-size-bytes": str(64 << 20)},
)

Attributes

Name	Description
create_options_ignored	The start of the warning a sink on an existing table gives when the

create_options_ignored

The start of the warning a sink on an existing table gives when the

create_options_ignored="create options ignored"

table differs from its location, partition_by or properties.

Methods

Name	Description
maintain()	Compact the table, and with `snapshot_retention`, expire the
source()	The table as a source, for the nodes that read the stored node.

maintain()

Compact the table, and with snapshot_retention, expire the

Usage

Source

maintain(*, snapshot_retention=None, min_snapshots=None, expire=None)

snapshots older than that and delete the files only they referenced. The compaction rewrites, in each partition of the current spec, the data files with position deletes, and the files under write.target-file-size-bytes where there are two or more, as data files of their live rows, and drops the delete files that no remaining data file reads, in one replace snapshot. It records the sverify properties of the sink’s last snapshot, if no other writer changed rows since, so a restarted job resumes past it. The expiry keeps the min_snapshots newest snapshots, the sink’s last snapshot and the ones after it, and branch and tag heads.

Parameters

snapshot_retention: Duration | None = None: How old a snapshot may be and still be read, a timedelta or an ISO 8601 duration ("P7D"), as the constructor’s snapshot_retention. Nodes reading the stored node from an older snapshot read the whole table. Default: None, no expiry.
min_snapshots: int | None = None: The newest snapshots the expiry keeps. Default: the sink’s min_snapshots.
expire: timedelta | None = None: Deprecated name of snapshot_retention.

Returns

dict[str, int]

The counts of data files the compaction removed and added

("data_files_removed", "data_files_added") and of delete

files it removed ("delete_files_removed"), and with

snapshot_retention,

of snapshots expired ("snapshots_expired") and files deleted

("files_deleted"); empty if the table does not exist.

Raises

CommitFailedException: Another writer committed to the table during the compaction (pyiceberg’s).

source()

The table as a source, for the nodes that read the stored node.

Usage

Source

source()