iceberg-versioned-reads

name	iceberg-versioned-reads
description	Iceberg source primitives for incremental computation: Snapshot, Delta (with inserts AND deletes), Version, Schema. Maps Iceberg changelog to __weight model.

Iceberg Versioned Reads

Implements the four Source Prerequisites defined by incremental-computation for Apache Iceberg tables on Spark.

Version

SELECT snapshot_id FROM <table>.snapshots ORDER BY committed_at DESC LIMIT 1

Returns the current snapshot ID — used as the version identifier in profiles.

Schema

spark.table("catalog.db.table").schema

Or via metadata table:

SELECT * FROM <table>.schema

Snapshot(V)

Read all rows at a specific version. No __weight column.

spark.read.option("snapshot-id", version).table("catalog.db.table")

Delta(V1, V2)

Full Delta (changelog — handles inserts, deletes, updates)

changelog_df = spark.read \
    .option("start-snapshot-id", from_version) \
    .option("end-snapshot-id", to_version) \
    .table("catalog.db.table.changes")

Returns all data columns plus _change_type with values: INSERT, DELETE, UPDATE_BEFORE, UPDATE_AFTER.

Map to __weight:

from pyspark.sql import functions as F

delta_df = changelog_df.withColumn(
    "__weight",
    F.when(F.col("_change_type").isin("INSERT", "UPDATE_AFTER"), F.lit(1))
     .when(F.col("_change_type").isin("DELETE", "UPDATE_BEFORE"), F.lit(-1))
).drop("_change_type", "_change_ordinal", "_commit_snapshot_id")

Append-Only Fast Path (positiveOnly)

When delete_count == 0, skip changelog overhead:

append_df = spark.read \
    .option("start-snapshot-id", from_version) \
    .option("end-snapshot-id", to_version) \
    .table("catalog.db.table")

All rows carry implicit __weight = +1. No _change_type column.

Detecting delete_count

Iceberg has two delete modes with different summary keys:

IMPORTANT: Iceberg snapshot_id is NOT sequential. Use committed_at timestamps or the history table to filter snapshots in a version range.

Method 1: Using committed_at timestamps

First, get the timestamps for the version range:

from_timestamp = spark.sql(f"""
    SELECT committed_at FROM <table>.snapshots WHERE snapshot_id = {from_version}
""").collect()[0]['committed_at']

to_timestamp = spark.sql(f"""
    SELECT committed_at FROM <table>.snapshots WHERE snapshot_id = {to_version}
""").collect()[0]['committed_at']

Then query snapshots in range:

Copy-on-write (default): DELETE rewrites entire data files. Check operation and deleted-records:

SELECT s.snapshot_id,
       s.operation,
       s.summary['deleted-records']        AS deleted_records,
       s.summary['deleted-data-files']     AS deleted_data_files
FROM <table>.snapshots s
WHERE s.committed_at > timestamp'{from_timestamp}'
  AND s.committed_at <= timestamp'{to_timestamp}'

Merge-on-read: DELETE produces delete files. Check delete file counters:

SELECT s.snapshot_id,
       s.summary['added-delete-files']     AS added_deletes,
       s.summary['added-equality-deletes'] AS eq_deletes,
       s.summary['added-position-deletes'] AS pos_deletes
FROM <table>.snapshots s
WHERE s.committed_at > timestamp'{from_timestamp}'
  AND s.committed_at <= timestamp'{to_timestamp}'

Method 2: Using history table

The history table records snapshots in chronological order:

SELECT s.snapshot_id,
       s.operation,
       s.summary['deleted-records']        AS deleted_records,
       s.summary['deleted-data-files']     AS deleted_data_files,
       s.summary['added-delete-files']     AS added_deletes,
       s.summary['added-equality-deletes'] AS eq_deletes,
       s.summary['added-position-deletes'] AS pos_deletes
FROM <table>.snapshots s
WHERE s.snapshot_id IN (
    SELECT h.snapshot_id 
    FROM <table>.history h
    WHERE h.made_current_at > (
        SELECT made_current_at FROM <table>.history WHERE snapshot_id = :from_version
    )
    AND h.made_current_at <= (
        SELECT made_current_at FROM <table>.history WHERE snapshot_id = :to_version
    )
)

Unified detection logic (covers both modes):

• If any snapshot has operation IN (delete, overwrite) OR deleted-records > 0 OR any delete file counter > 0 → delete_count > 0 → use full changelog Delta.
• Otherwise → delete_count = 0 → use append-only fast path.

Integration with incremental-computation

When incremental-computation asks for source prerequisites, answer with:

Prerequisite	Iceberg Implementation
Version	SELECT snapshot_id FROM <table>.snapshots ORDER BY committed_at DESC LIMIT 1
Schema	spark.table(t).schema
Snapshot(V)	spark.read.option("snapshot-id", V).table(t)
Delta(V1,V2)	Changelog: .table(t + ".changes") → map _change_type to __weight
delete_count	Snapshot operation + deleted-records (CoW) or added-delete-files (MoR)