Iceberg format versions: what v1, v2, v3 and v4 actually change

import json

path = spark.sql("SELECT file FROM ice.blog_format_versions.orders_v3.metadata_log_entries "
                 "ORDER BY timestamp DESC LIMIT 1").collect()[0][0]
md = json.loads(spark.read.text(path, wholetext=True).collect()[0][0])
print("format-version:", md["format-version"], "next-row-id:", md.get("next-row-id", "absent"))

"""Create an Iceberg table at every format version and compare what each one writes.

Run inside datalake-playground:
  docker cp format_versions.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[2]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/format_versions.py'
"""
import json

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("iceberg-format-versions")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql
NS = "ice.blog_format_versions"


def error(e):
    java = getattr(e, "java_exception", None)
    if java is not None:
        return f"{java.getClass().getSimpleName()}: {java.getMessage().splitlines()[0]}"
    return f"{type(e).__name__}: {str(e).splitlines()[0]}"


def current_metadata(table):
    """The table's current metadata.json, parsed."""
    path = sql(f"SELECT file FROM {table}.metadata_log_entries "
               f"ORDER BY timestamp DESC LIMIT 1").collect()[0][0]
    return json.loads(spark.read.text(path, wholetext=True).collect()[0][0])


def file_counts(table):
    # .files lists every live file of the current snapshot: data (0), position
    # deletes or deletion vectors (1) and equality deletes (2)
    rows = sql(f"SELECT content, count(*) AS n FROM {table}.files GROUP BY content").collect()
    by = {r.content: r.n for r in rows}
    return f"data={by.get(0, 0)} position/DV deletes={by.get(1, 0)} equality deletes={by.get(2, 0)}"


sql(f"CREATE NAMESPACE IF NOT EXISTS {NS}")

print("== 1. every format version ==")
for v in (1, 2, 3, 4, 5):
    t = f"{NS}.orders_v{v}"
    sql(f"DROP TABLE IF EXISTS {t} PURGE")
    try:
        sql(f"""CREATE TABLE {t} (order_id BIGINT, region STRING, amount DECIMAL(10,2))
                USING iceberg TBLPROPERTIES ('format-version' = '{v}')""")
    except Exception as e:
        print(f"format-version={v}: CREATE failed: {error(e)}")
        continue
    sql(f"INSERT INTO {t} VALUES (1001, 'eu', 120.50), (1002, 'us', 75.00), (1003, 'apac', 310.25)")
    md = current_metadata(t)
    print(f"format-version={v}: CREATE ok, rows={spark.table(t).count()}, "
          f"metadata format-version={md['format-version']}, next-row-id={md.get('next-row-id', 'absent')}")

print("\n== 2. the default ==")
sql(f"DROP TABLE IF EXISTS {NS}.orders_default PURGE")
sql(f"CREATE TABLE {NS}.orders_default (order_id BIGINT, region STRING) USING iceberg")
for r in sql(f"SHOW TBLPROPERTIES {NS}.orders_default").collect():
    if r.key in ("format-version", "write.parquet.compression-codec"):
        print(f"{r.key} = {r.value}")

print("\n== 3. a DELETE on each version (default copy-on-write) ==")
for v in (1, 2, 3, 4):
    t = f"{NS}.orders_v{v}"
    sql(f"DELETE FROM {t} WHERE order_id = 1002")
    op = sql(f"SELECT operation FROM {t}.snapshots ORDER BY committed_at DESC LIMIT 1").collect()[0][0]
    print(f"format-version={v} after DELETE: rows={spark.table(t).count()}, {file_counts(t)}, operation={op}")

print("\n== 4. merge-on-read requested on v1 ==")
sql(f"ALTER TABLE {NS}.orders_v1 SET TBLPROPERTIES ('write.delete.mode' = 'merge-on-read')")
try:
    sql(f"DELETE FROM {NS}.orders_v1 WHERE order_id = 1003")
    print(f"v1 merge-on-read DELETE: rows={spark.table(NS + '.orders_v1').count()}, {file_counts(NS + '.orders_v1')}")
except Exception as e:
    print("v1 merge-on-read DELETE:", error(e))

print("\n== 5. row lineage columns ==")
for v in (2, 3):
    t = f"{NS}.orders_v{v}"
    try:
        rows = sql(f"SELECT order_id, _row_id, _last_updated_sequence_number FROM {t} ORDER BY order_id").collect()
        print(f"v{v}:", [(r.order_id, r._row_id, r._last_updated_sequence_number) for r in rows])
    except Exception as e:
        print(f"v{v} lineage columns:", error(e))
sql(f"UPDATE {NS}.orders_v3 SET amount = 99.99 WHERE order_id = 1001")
rows = sql(f"SELECT order_id, _row_id, _last_updated_sequence_number FROM {NS}.orders_v3 ORDER BY order_id").collect()
print("v3 after UPDATE:", [(r.order_id, r._row_id, r._last_updated_sequence_number) for r in rows])
print("v3 next-row-id now:", current_metadata(f"{NS}.orders_v3").get("next-row-id"))

print("\n== 6. a column with a default ==")
for v in (2, 3):
    t = f"{NS}.orders_v{v}"
    try:
        sql(f"ALTER TABLE {t} ADD COLUMN channel STRING DEFAULT 'web'")
        print(f"v{v} ADD COLUMN ... DEFAULT: ok, existing rows read",
              [r.channel for r in sql(f"SELECT channel FROM {t}").collect()])
    except Exception as e:
        print(f"v{v} ADD COLUMN ... DEFAULT:", error(e))

spark.stop()

"""Compare delete-file accumulation on v2 and v3 merge-on-read tables.

Each table holds 5,000 orders in one data file and takes the same six row-level
commits: three DELETEs and three MERGEs. After every commit the script counts the
live delete files, their format and size, and the delete files ever written.

Three tables: v2 with Spark's default file-scoped deletes, v2 with
'write.delete.granularity' = 'partition', and v3, which uses deletion vectors.

Run inside datalake-playground:
  docker cp deletion_vectors.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[2]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/deletion_vectors.py'
"""
from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder.appName("iceberg-deletion-vectors")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql
NS = "ice.blog_format_versions"

orders = (spark.range(1, 5001)
               .select(F.col("id").alias("order_id"),
                       (F.col("id") % 500 + 1).alias("customer_id"),
                       F.element_at(F.array(F.lit("PLACED"), F.lit("PAID"), F.lit("SHIPPED")),
                                    (F.col("id") % 3 + 1).cast("int")).alias("status"),
                       (F.col("id") % 400 + 10).cast("decimal(10,2)").alias("amount"))
               .coalesce(1))


def state(t):
    live = sql(f"SELECT file_format, count(*) AS n, sum(file_size_in_bytes) AS bytes "
               f"FROM {t}.files WHERE content = 1 GROUP BY file_format").collect()
    ever = sql(f"SELECT count(DISTINCT file_path) FROM {t}.all_delete_files").collect()[0][0]
    data = sql(f"SELECT count(*) FROM {t}.files WHERE content = 0").collect()[0][0]
    # sizes to 0.1 KB: Parquet delete files vary by a few bytes from run to run
    fmt = ", ".join(f"{r.n} {r.file_format} ({r.bytes / 1024:.1f} KB)" for r in live) or "0"
    return f"data files={data}  live delete files={fmt}  delete files ever written={ever}  rows={spark.table(t).count()}"


CASES = [("v2", 2, "file"), ("v2_partition", 2, "partition"), ("v3", 3, "file")]
for name, v, granularity in CASES:
    t = f"{NS}.cdc_orders_{name}"
    sql(f"DROP TABLE IF EXISTS {t} PURGE")
    sql(f"""CREATE TABLE {t} (order_id BIGINT, customer_id BIGINT, status STRING, amount DECIMAL(10,2))
            USING iceberg TBLPROPERTIES (
              'format-version' = '{v}',
              'write.delete.mode' = 'merge-on-read',
              'write.update.mode' = 'merge-on-read',
              'write.merge.mode'  = 'merge-on-read',
              'write.delete.granularity' = '{granularity}')""")
    orders.writeTo(t).append()
    print(f"== format-version={v}, write.delete.granularity={granularity} ==")
    print("  after load     ", state(t))
    for i, lo in enumerate((100, 200, 300), 1):
        sql(f"DELETE FROM {t} WHERE order_id BETWEEN {lo} AND {lo + 9}")
        print(f"  after DELETE {i} ", state(t))
    for i, lo in enumerate((1000, 2000, 3000), 1):
        spark.range(lo, lo + 20).select(F.col("id").alias("order_id")).createOrReplaceTempView("changes")
        sql(f"""MERGE INTO {t} t USING changes c ON t.order_id = c.order_id
                WHEN MATCHED THEN UPDATE SET t.status = 'CANCELLED'""")
        print(f"  after MERGE {i}  ", state(t))
    cancelled = sql(f"SELECT count(*) FROM {t} WHERE status = 'CANCELLED'").collect()[0][0]
    print(f"  cancelled rows = {cancelled}")

spark.stop()

"""Upgrade an Iceberg table from v2 to v3 on a copy first, and see what changes.

Run inside datalake-playground:
  docker cp upgrade.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[2]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/upgrade.py'
"""
import json

from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder.appName("iceberg-format-upgrade")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql
NS = "ice.blog_format_versions"
SRC, COPY = f"{NS}.events", f"{NS}.events_v3_test"


def error(e):
    java = getattr(e, "java_exception", None)
    if java is not None:
        return f"{java.getClass().getSimpleName()}: {java.getMessage().splitlines()[0]}"
    return f"{type(e).__name__}: {str(e).splitlines()[0]}"


def metadata(table):
    path = sql(f"SELECT file FROM {table}.metadata_log_entries "
               f"ORDER BY timestamp DESC LIMIT 1").collect()[0][0]
    return path, json.loads(spark.read.text(path, wholetext=True).collect()[0][0])


def version(table):
    return [r.value for r in sql(f"SHOW TBLPROPERTIES {table} ('format-version')").collect()][0]


def with_row_id(table):
    # count(_row_id) is pushed down to Iceberg, which fails with "Cannot find field
    # '_row_id'"; reading the column and counting in Spark avoids the pushdown
    ids = [r[0] for r in sql(f"SELECT _row_id FROM {table}").collect()]
    return f"{sum(i is not None for i in ids)} of {len(ids)}"


def deletes(table):
    rows = sql(f"SELECT file_format, count(*) AS n FROM {table}.files WHERE content = 1 "
               f"GROUP BY file_format ORDER BY file_format").collect()
    return ", ".join(f"{r.n} {r.file_format}" for r in rows) or "none"


# The registered copy shares the source's files and directory, so it is dropped
# WITHOUT purge: DROP TABLE ... PURGE on the copy deletes the source's files too.
sql(f"DROP TABLE IF EXISTS {COPY}")
sql(f"DROP TABLE IF EXISTS {SRC} PURGE")

# A v2 merge-on-read table of 1,000 telemetry events that has already taken deletes
sql(f"""CREATE TABLE {SRC} (id BIGINT, device STRING, region STRING, reading DOUBLE)
        USING iceberg TBLPROPERTIES (
          'format-version' = '2',
          'write.delete.mode' = 'merge-on-read',
          'write.update.mode' = 'merge-on-read')""")
(spark.range(0, 1000)
      .select("id",
              F.concat(F.lit("sensor-"), (F.col("id") % 50).cast("string")).alias("device"),
              F.concat(F.lit("r"), (F.col("id") % 5).cast("string")).alias("region"),
              (F.col("id") % 97 / 10.0).alias("reading"))
      .coalesce(1).writeTo(SRC).append())
sql(f"DELETE FROM {SRC} WHERE id % 100 = 0")
print(f"source: format-version={version(SRC)}, rows={spark.table(SRC).count()}, delete files: {deletes(SRC)}")

print("\n== 1. copy with the snapshot procedure ==")
try:
    sql(f"CALL ice.system.snapshot('{SRC.split('.', 1)[1]}', '{COPY.split('.', 1)[1]}')").show()
except Exception as e:
    print("snapshot:", error(e))

print("\n== 2. copy by registering the same metadata file ==")
src_path, _ = metadata(SRC)
sql(f"CALL ice.system.register_table(table => '{COPY.split('.', 1)[1]}', "
    f"metadata_file => '{src_path}')").show(truncate=False)
print(f"copy: format-version={version(COPY)}, rows={spark.table(COPY).count()}")

print("\n== 3. raise the copy ==")
sql(f"ALTER TABLE {COPY} SET TBLPROPERTIES ('format-version' = '3')")
_, md = metadata(COPY)
print(f"copy: format-version={version(COPY)}, next-row-id={md.get('next-row-id')}; "
      f"source still format-version={version(SRC)}")
print("rows with a _row_id:", with_row_id(COPY))

print("\n== 4. exercise the new representations on the copy ==")
sql(f"DELETE FROM {COPY} WHERE id < 100")
print(f"after DELETE : rows={spark.table(COPY).count()}, delete files: {deletes(COPY)}")
sql(f"UPDATE {COPY} SET region = 'r9' WHERE id BETWEEN 200 AND 300")
print(f"after UPDATE : rows={spark.table(COPY).count()}, delete files: {deletes(COPY)}, "
      f"r9 rows={sql(f'SELECT count(*) FROM {COPY} WHERE region = ' + chr(39) + 'r9' + chr(39)).collect()[0][0]}")
_, md = metadata(COPY)
print(f"copy next-row-id={md.get('next-row-id')}; rows with a _row_id: {with_row_id(COPY)}")
print(f"source untouched: rows={spark.table(SRC).count()}, delete files: {deletes(SRC)}")

print("\n== 5. try to go back ==")
try:
    sql(f"ALTER TABLE {COPY} SET TBLPROPERTIES ('format-version' = '2')")
    print("downgrade: accepted?!")
except Exception as e:
    print("downgrade:", error(e))

print("\n== 6. the copy is not independent ==")
shared = sql(f"""SELECT count(*) FROM {SRC}.files s JOIN {COPY}.all_files c
                 ON s.file_path = c.file_path""").collect()[0][0]
print("source files also referenced by the copy:", shared)
loc = lambda t: sql(f"DESCRIBE TABLE EXTENDED {t}").where("col_name = 'Location'").collect()[0][1]
print("source location:", loc(SRC))
print("copy location  :", loc(COPY))
sql(f"DROP TABLE {COPY}")          # never PURGE a registered copy
print("after dropping the copy (no PURGE), source rows:", spark.table(SRC).count())

spark.stop()