Schema evolution in Iceberg: why a rename costs nothing

"""Schema evolution on an Iceberg events table: what each change costs, and field ids.

Run inside datalake-playground:
  docker cp schema_evolution.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/schema_evolution.py'
"""
import json
from datetime import datetime, timedelta
from decimal import Decimal

from pyspark.sql import SparkSession
from pyspark.sql.types import (DecimalType, IntegerType, StringType, StructField,
                               StructType, TimestampType)

spark = (
    SparkSession.builder.appName("iceberg-schema-evolution")
    .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
TABLE = "ice.blog_schema_evolution.events"
REGIONS = ["us-east", "us-west", "eu-central"]

SCHEMA = StructType([
    StructField("id", IntegerType(), False),
    StructField("event_ts", TimestampType(), False),
    StructField("region", StringType(), False),
    StructField("amount", DecimalType(10, 2), False),
])


def batch(start):
    """3,000 purchase events, one data file."""
    t0 = datetime(2026, 9, 1)
    rows = [(i, t0 + timedelta(minutes=i), REGIONS[i % 3], Decimal(i % 500) + Decimal("0.99"))
            for i in range(start, start + 3000)]
    spark.createDataFrame(rows, SCHEMA).coalesce(1).writeTo(TABLE).append()


def columns():
    return spark.table(TABLE).columns


def data_files():
    return sql(f"SELECT count(*) FROM {TABLE}.files").collect()[0][0]


def metadata_files():
    return sql(f"SELECT count(*) FROM {TABLE}.metadata_log_entries").collect()[0][0]


def current_metadata():
    """Read the table's current metadata.json straight from MinIO."""
    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 print_fields(fields, depth=0):
    for f in fields:
        t = f["type"]
        type_name = t["type"] if isinstance(t, dict) else t
        print("  " * depth + "field id=%-3s name=%-12s type=%s" % (f["id"], f["name"], type_name))
        if isinstance(t, dict) and t.get("type") == "struct":
            print_fields(t["fields"], depth + 1)


def current_schema_fields():
    md = current_metadata()
    schema = next(s for s in md["schemas"] if s["schema-id"] == md["current-schema-id"])
    return schema["fields"]


sql("CREATE NAMESPACE IF NOT EXISTS ice.blog_schema_evolution")
sql(f"DROP TABLE IF EXISTS {TABLE} PURGE")
sql(f"""CREATE TABLE {TABLE} (id INT, event_ts TIMESTAMP, region STRING, amount DECIMAL(10,2))
        USING iceberg""")
for start in (0, 3000, 6000):
    batch(start)
first_snapshot = sql(f"SELECT snapshot_id FROM {TABLE}.history "
                     f"ORDER BY made_current_at LIMIT 1").collect()[0][0]

print("== 1. three changes, measured ==")
print("before:", columns())
files_before, metadata_before = data_files(), metadata_files()
sql(f"ALTER TABLE {TABLE} ADD COLUMN currency STRING")
sql(f"ALTER TABLE {TABLE} RENAME COLUMN amount TO amount_usd")
sql(f"ALTER TABLE {TABLE} ALTER COLUMN id TYPE BIGINT")
print("after :", columns())
print("rows still readable            =", spark.table(TABLE).count())
print("old rows have NULL currency    =",
      sql(f"SELECT count(*) FROM {TABLE} WHERE currency IS NULL").collect()[0][0])
print("data files before / after      =", files_before, "/", data_files())
print("metadata.json files before / after =", metadata_before, "/", metadata_files())

print("\n== 2. field ids after the rename ==")
print_fields(current_schema_fields())

print("\n== 3. dropping and reordering on a Hive catalog ==")


def metadata_json_on_storage():
    """Count metadata.json files in the table's metadata directory on MinIO."""
    location = sql(f"DESCRIBE TABLE EXTENDED {TABLE}").where("col_name = 'Location'").collect()[0][1]
    jvm = spark._jvm
    path = jvm.org.apache.hadoop.fs.Path(location + "/metadata")
    fs = path.getFileSystem(spark._jsc.hadoopConfiguration())
    return sum(1 for st in fs.listStatus(path) if st.getPath().getName().endswith(".metadata.json"))


def attempt(label, statement):
    try:
        sql(statement)
        print(f"{label}: accepted")
    except Exception as e:
        java = getattr(e, "java_exception", None)
        name = java.getClass().getName().split(".")[-1] if java is not None else type(e).__name__
        msg = (java.getMessage() if java is not None else str(e)).splitlines()
        print(f"{label}: {name} - {msg[0]} {msg[1] if len(msg) > 1 else ''}".rstrip())


attempt("DROP COLUMN region (not last)", f"ALTER TABLE {TABLE} DROP COLUMN region")
print("columns after the failed drop:", columns())
print("metadata.json in the table log / on storage:", metadata_files(), "/", metadata_json_on_storage())
attempt("ALTER COLUMN currency FIRST", f"ALTER TABLE {TABLE} ALTER COLUMN currency FIRST")
attempt("ADD COLUMN channel AFTER id", f"ALTER TABLE {TABLE} ADD COLUMN channel STRING AFTER id")

print("\n-- drop the last column, then add a new one --")
sql(f"ALTER TABLE {TABLE} DROP COLUMN currency")
sql(f"ALTER TABLE {TABLE} ADD COLUMN country STRING")
print_fields(current_schema_fields())
print("country values on old rows:",
      [r[0] for r in sql(f"SELECT DISTINCT country FROM {TABLE}").collect()])
print("last-column-id:", current_metadata()["last-column-id"])

print("\n== 4. nested fields ==")
sql(f"ALTER TABLE {TABLE} ADD COLUMN payload struct<source: string, version: int>")
sql(f"ALTER TABLE {TABLE} ADD COLUMN payload.trace_id string")
try:
    sql(f"ALTER TABLE {TABLE} RENAME COLUMN payload.source TO payload.origin")
    print("RENAME ... TO payload.origin: accepted")
except Exception as e:
    print("RENAME ... TO payload.origin:", type(e).__name__, "-", " ".join(str(e).split())[:90])
sql(f"ALTER TABLE {TABLE} RENAME COLUMN payload.source TO origin")
print_fields(current_schema_fields())
print("payload on old rows:",
      [r[0] for r in sql(f"SELECT DISTINCT payload FROM {TABLE}").collect()])
print("data files now:", data_files())

print("\n== 5. old snapshots keep their schema ==")
old = spark.read.option("snapshot-id", str(first_snapshot)).format("iceberg").load(TABLE)
print("columns at the first snapshot:", old.columns)
print("columns now                  :", columns())
try:
    sql(f"SELECT snapshot_id, schema_id FROM {TABLE}.snapshots").collect()
    print("snapshots.schema_id: exists")
except Exception as e:
    print("snapshots.schema_id:", type(e).__name__, "-", str(e).splitlines()[0][:60])
print("snapshots columns:", spark.table(f"{TABLE}.snapshots").columns)
sql(f"""SELECT latest_snapshot_id, latest_schema_id
        FROM {TABLE}.metadata_log_entries ORDER BY timestamp""").show(20, truncate=False)
md = current_metadata()
print("schema-id per snapshot (from metadata.json):",
      [s.get("schema-id") for s in md["snapshots"]])
print("schemas in metadata.json:", len(md["schemas"]),
      "| current-schema-id:", md["current-schema-id"])

spark.stop()

"""Type promotion rules and the add-migrate-drop pattern on an Iceberg table.

`ice` is the stack's Hive Metastore catalog. `hdp` is a Hadoop catalog on the same
MinIO bucket, used to show the pattern on a catalog without the metastore's
positional column check.

Run inside datalake-playground:
  docker cp type_promotion.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/type_promotion.py'
"""
import re
from decimal import Decimal

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("iceberg-type-promotion")
    .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.sql.catalog.hdp", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.hdp.type", "hadoop")
    .config("spark.sql.catalog.hdp.warehouse", "s3a://warehouse/blog_schema_evolution_hadoop")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql


def attempt(label, statement):
    try:
        sql(statement)
        print(f"  {label:<44} accepted")
    except Exception as e:
        java = getattr(e, "java_exception", None)
        name = java.getClass().getName().split(".")[-1] if java is not None else type(e).__name__
        msg = " ".join((java.getMessage() if java is not None else str(e)).split())
        types = re.findall(r'with type "([^"]+)"', msg)
        code = re.match(r"\[([A-Z_]+)\]", msg)
        if code and len(types) == 2:
            msg = f"[{code.group(1)}] {types[0]} -> {types[1]}"
        elif "respective positions" in msg:
            cols = re.search(r"respective positions : ([A-Za-z0-9_,]+)", msg).group(1)
            msg = "columns incompatible in their positions: " + cols
        print(f"  {label:<44} {name}: {msg[:110]}")


def orders(table):
    """Load 3,000 orders as one data file."""
    rows = [(i, Decimal(i % 900) + Decimal("0.49"), float(i % 97) / 7, i % 5, f"C{i % 40:03d}")
            for i in range(3000)]
    (spark.createDataFrame(rows, "order_id BIGINT, amount DECIMAL(10,2), weight FLOAT, "
                                 "qty INT, customer STRING")
          .coalesce(1).writeTo(table).append())


def files(table):
    return sql(f"SELECT count(*) FROM {table}.files").collect()[0][0]


# ------------------------------------------------------------------ promotions
T = "ice.blog_schema_evolution.orders"
sql("CREATE NAMESPACE IF NOT EXISTS ice.blog_schema_evolution")
sql(f"DROP TABLE IF EXISTS {T} PURGE")
sql(f"""CREATE TABLE {T} (order_id BIGINT, amount DECIMAL(10,2), weight FLOAT, qty INT,
        customer STRING) USING iceberg""")
orders(T)

print("== 1. type promotion ==")
attempt("qty int -> bigint", f"ALTER TABLE {T} ALTER COLUMN qty TYPE BIGINT")
attempt("weight float -> double", f"ALTER TABLE {T} ALTER COLUMN weight TYPE DOUBLE")
attempt("amount decimal(10,2) -> decimal(18,2)", f"ALTER TABLE {T} ALTER COLUMN amount TYPE DECIMAL(18,2)")
attempt("amount decimal(18,2) -> decimal(18,4)", f"ALTER TABLE {T} ALTER COLUMN amount TYPE DECIMAL(18,4)")
attempt("order_id bigint -> int", f"ALTER TABLE {T} ALTER COLUMN order_id TYPE INT")
attempt("weight double -> float", f"ALTER TABLE {T} ALTER COLUMN weight TYPE FLOAT")
attempt("customer string -> int", f"ALTER TABLE {T} ALTER COLUMN customer TYPE INT")
print("  schema now:", [(f.name, f.dataType.simpleString()) for f in spark.table(T).schema])
print("  data files:", files(T), "| rows:", spark.table(T).count())

# ------------------------------------------- add-migrate-drop on the Hive catalog
print("\n== 2. add-migrate-drop on the Hive catalog ==")
sql(f"ALTER TABLE {T} ADD COLUMN amount_small FLOAT")
sql(f"UPDATE {T} SET amount_small = CAST(amount AS FLOAT)")
summary = sql(f"SELECT operation, summary FROM {T}.snapshots ORDER BY committed_at DESC LIMIT 1").collect()[0]
print("  UPDATE commit:", summary["operation"], "| added-data-files =", summary["summary"]["added-data-files"],
      "| deleted-data-files =", summary["summary"]["deleted-data-files"])
print("  rows where the cast lost a value:",
      sql(f"SELECT count(*) FROM {T} WHERE amount_small IS NULL AND amount IS NOT NULL").collect()[0][0])
attempt("DROP COLUMN amount", f"ALTER TABLE {T} DROP COLUMN amount")
print("  columns:", spark.table(T).columns)
sql(f"DROP TABLE {T} PURGE")

# ------------------------------------------ the same pattern on a Hadoop catalog
print("\n== 3. the same pattern on a Hadoop catalog ==")
H = "hdp.db.orders"
sql(f"DROP TABLE IF EXISTS {H} PURGE")
sql(f"""CREATE TABLE {H} (order_id BIGINT, amount DECIMAL(10,2), weight FLOAT, qty INT,
        customer STRING) USING iceberg""")
orders(H)
sql(f"ALTER TABLE {H} ADD COLUMN amount_small FLOAT")
sql(f"UPDATE {H} SET amount_small = CAST(amount AS FLOAT)")
print("  rows where the cast lost a value:",
      sql(f"SELECT count(*) FROM {H} WHERE amount_small IS NULL AND amount IS NOT NULL").collect()[0][0])
before_drop = sql(f"SELECT snapshot_id FROM {H}.refs WHERE name = 'main'").collect()[0][0]
attempt("DROP COLUMN amount", f"ALTER TABLE {H} DROP COLUMN amount")
print("  columns:", spark.table(H).columns)

data_file = sql(f"SELECT file_path FROM {H}.files LIMIT 1").collect()[0][0]
print("  columns physically in a current data file:", spark.read.parquet(data_file).columns)

result = sql(f"CALL hdp.system.rewrite_data_files(table => 'db.orders', "
             f"options => map('rewrite-all', 'true'))").collect()[0]
print("  rewrite_data_files rewrote", result["rewritten_data_files_count"], "file(s) into",
      result["added_data_files_count"])
new_file = sql(f"SELECT file_path FROM {H}.files LIMIT 1").collect()[0][0]
print("  columns physically in the rewritten file:", spark.read.parquet(new_file).columns)
old = spark.read.option("snapshot-id", str(before_drop)).format("iceberg").load(H)
print("  time travel to before the drop, after the rewrite:", old.columns,
      "| sum(amount) =", old.agg({"amount": "sum"}).collect()[0][0])

sql(f"CALL hdp.system.expire_snapshots(table => 'db.orders', "
    f"older_than => TIMESTAMP '2099-01-01 00:00:00', retain_last => 1)").collect()
try:
    spark.read.option("snapshot-id", str(before_drop)).format("iceberg").load(H).count()
    print("  time travel after expire_snapshots: still readable")
except Exception as e:
    print("  time travel after expire_snapshots:", type(e).__name__, "-", str(e).splitlines()[0][:90])

sql(f"DROP TABLE {H} PURGE")
spark.stop()