# Expire snapshots older than 7 days
from pyspark.sql import SparkSession

spark.sql("""
    CALL spark_catalog.system.expire_snapshots(
        table => 'db.orders',
        older_than => TIMESTAMP '2024-01-01 00:00:00',
        retain_last => 3
    )
""")

# Or using the Python API
from pyiceberg.catalog import load_catalog

catalog = load_catalog("glue")
table = catalog.load_table("db.orders")
table.expire_snapshots().expire_older_than(1704067200000).commit()

-- View all snapshots of a table
SELECT * FROM db.orders.snapshots;

-- View history (human-readable)
SELECT * FROM db.orders.history;

-- View metadata files
SELECT * FROM db.orders.metadata_log_entries;

-- View data files in current snapshot
SELECT * FROM db.orders.files;

# Method 1: Using Spark SQL AS OF TIMESTAMP
spark.sql("""
    SELECT * FROM db.orders
    TIMESTAMP AS OF '2024-01-15 10:00:00'
""")

# Method 2: Using DataFrame API option
df = spark.read \
    .format("iceberg") \
    .option("as-of-timestamp", "1705312800000") \  # epoch milliseconds
    .load("db.orders")

df.show()

# Find available snapshots first
spark.sql("SELECT snapshot_id, committed_at, operation FROM db.orders.snapshots").show()

# Read by snapshot ID — SQL
spark.sql("""
    SELECT * FROM db.orders
    VERSION AS OF 8537401032498282350
""")

# Read by snapshot ID — DataFrame API
df = spark.read \
    .format("iceberg") \
    .option("snapshot-id", "8537401032498282350") \
    .load("db.orders")

df.show()

# Rollback to a specific snapshot ID
spark.sql("""
    CALL spark_catalog.system.rollback_to_snapshot(
        'db.orders',
        8537401032498282350
    )
""")

# Rollback to a timestamp
spark.sql("""
    CALL spark_catalog.system.rollback_to_timestamp(
        'db.orders',
        TIMESTAMP '2024-01-15 09:00:00'
    )
""")

-- Create table partitioned by month
CREATE TABLE db.orders (
    order_id    BIGINT,
    customer_id BIGINT,
    order_date  DATE,
    amount      DOUBLE
) USING iceberg
PARTITIONED BY (months(order_date));

-- After 1 year, data grew — switch to day partitioning
-- Old data stays in month partitions, new data uses day partitions
ALTER TABLE db.orders
REPLACE PARTITION FIELD months(order_date)
WITH days(order_date);

-- Add a second partition field (combined partitioning)
ALTER TABLE db.orders
ADD PARTITION FIELD region;

-- Drop a partition field
ALTER TABLE db.orders
DROP PARTITION FIELD region;

# Create table with bucket + day partitioning
spark.sql("""
    CREATE TABLE db.events (
        event_id    BIGINT,
        user_id     BIGINT,
        event_type  STRING,
        event_ts    TIMESTAMP,
        payload     STRING
    ) USING iceberg
    PARTITIONED BY (
        bucket(16, user_id),
        days(event_ts)
    )
""")

# Query — no need to specify partition columns explicitly
# Iceberg auto-prunes buckets and day partitions
spark.sql("""
    SELECT * FROM db.events
    WHERE user_id = 12345
      AND event_ts BETWEEN '2024-01-01' AND '2024-01-31'
""").show()

-- Add a new column
ALTER TABLE db.orders ADD COLUMN discount DOUBLE;

-- Add a nested column inside a struct
ALTER TABLE db.orders ADD COLUMN address.zip_code STRING;

-- Old files return NULL for the new column — safe!
SELECT order_id, discount FROM db.orders;
-- Old rows: discount = NULL, new rows: discount = actual value

-- Rename column (field-id stays the same internally)
ALTER TABLE db.orders RENAME COLUMN price TO unit_price;

-- Old files: column was written as "price" but has field-id=5
-- New files: column written as "unit_price" with same field-id=5
-- Queries using "unit_price" work on both old and new files!

-- Drop a column (data still exists in old files, just not read)
ALTER TABLE db.orders DROP COLUMN internal_notes;

-- Reorder columns (logical reordering, no data rewrite)
ALTER TABLE db.orders ALTER COLUMN discount
    AFTER order_date;

-- Change column type (with compatible promotion)
-- INT → LONG is safe (widening)
ALTER TABLE db.orders ALTER COLUMN quantity
    TYPE BIGINT;

-- View current schema
DESCRIBE TABLE db.orders;

# Append new rows to Iceberg table
df_new = spark.read.parquet("s3://bucket/new_orders/")

df_new.writeTo("db.orders").append()

# Or using SQL
spark.sql("""
    INSERT INTO db.orders
    SELECT order_id, customer_id, order_date, amount
    FROM staging.new_orders
""")

# Overwrite using dynamic partition overwrite
spark.sql("""
    INSERT OVERWRITE db.orders
    SELECT * FROM staging.new_orders
    WHERE order_date = '2024-01-15'
""")

-- Full upsert: update if exists, insert if not
MERGE INTO db.orders AS target
USING staging.orders_updates AS source
ON target.order_id = source.order_id

WHEN MATCHED AND source.op = 'DELETE' THEN DELETE

WHEN MATCHED AND source.op = 'UPDATE' THEN UPDATE SET
    target.amount     = source.amount,
    target.updated_at = source.updated_at

WHEN NOT MATCHED AND source.op = 'INSERT' THEN INSERT *;

from pyspark.sql.functions import expr

# Create source updates DataFrame
source_df = spark.read.table("staging.orders_updates")

# Register as temp view for SQL MERGE
source_df.createOrReplaceTempView("source")

spark.sql("""
    MERGE INTO db.orders t
    USING source s
    ON t.order_id = s.order_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

-- Update rows matching a condition
UPDATE db.orders
SET status = 'CANCELLED'
WHERE order_date < '2023-01-01' AND amount = 0;

-- Delete rows matching a condition (GDPR right to erasure!)
DELETE FROM db.orders
WHERE customer_id = 99999;

-- Delete a full partition efficiently
DELETE FROM db.orders
WHERE order_date = '2020-01-01';
-- Iceberg skips rewriting other partitions!

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("IcebergDemo") \
    .config("spark.jars.packages",
           "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0") \
    # Register Iceberg catalog — name it "spark_catalog" to be default
    .config("spark.sql.catalog.spark_catalog",
           "org.apache.iceberg.spark.SparkSessionCatalog") \
    .config("spark.sql.catalog.spark_catalog.type", "hive") \
    # Or use a separate catalog name
    .config("spark.sql.catalog.iceberg",
           "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.iceberg.type", "hadoop") \
    .config("spark.sql.catalog.iceberg.warehouse", "s3://my-bucket/warehouse") \
    .getOrCreate()

# Enable Iceberg extensions for DML (MERGE, UPDATE, DELETE)
spark.conf.set("spark.sql.extensions",
    "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")

from pyspark.sql.types import StructType, StructField, LongType, StringType, DoubleType, DateType
from pyspark.sql.functions import col, lit, current_date

# Create Iceberg table via SQL
spark.sql("""
    CREATE TABLE IF NOT EXISTS iceberg.db.orders (
        order_id    BIGINT NOT NULL,
        customer_id BIGINT,
        order_date  DATE,
        region      STRING,
        amount      DOUBLE
    ) USING iceberg
    PARTITIONED BY (months(order_date), identity(region))
    TBLPROPERTIES (
        'write.metadata.delete-after-commit.enabled' = 'true',
        'write.metadata.previous-versions-max' = '10'
    )
""")

# Write DataFrame to Iceberg table
df = spark.createDataFrame([
    (1, 101, "2024-01-15", "US", 250.0),
    (2, 102, "2024-01-16", "EU", 180.0),
], ["order_id", "customer_id", "order_date", "region", "amount"])

# Append
df.writeTo("iceberg.db.orders").append()

# Overwrite by partition filter
df.writeTo("iceberg.db.orders") \
  .overwritePartitions()

# Create and overwrite (replace) entire table
df.writeTo("iceberg.db.orders").createOrReplace()

# Standard read
df = spark.read.table("iceberg.db.orders")
df.show()

# SQL read
spark.sql("SELECT * FROM iceberg.db.orders WHERE region = 'US'").show()

# Incremental read — read only changes since a snapshot
spark.read \
    .format("iceberg") \
    .option("start-snapshot-id", "8537401032498282350") \
    .option("end-snapshot-id",   "9123456789012345678") \
    .load("iceberg.db.orders") \
    .show()

spark = SparkSession.builder \
    .appName("IcebergGlue") \
    .config("spark.jars.packages",
           "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0,"
           "software.amazon.awssdk:bundle:2.20.160") \
    .config("spark.sql.extensions",
           "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    # Glue Catalog
    .config("spark.sql.catalog.glue_catalog",
           "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.glue_catalog.warehouse",
           "s3://my-bucket/warehouse") \
    .config("spark.sql.catalog.glue_catalog.catalog-impl",
           "org.apache.iceberg.aws.glue.GlueCatalog") \
    .config("spark.sql.catalog.glue_catalog.io-impl",
           "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.sql.catalog.glue_catalog.aws.region", "us-east-1") \
    .getOrCreate()

# Now use the glue_catalog prefix in table references
spark.sql("CREATE DATABASE IF NOT EXISTS glue_catalog.analytics")
spark.sql("SHOW TABLES IN glue_catalog.analytics").show()

spark = SparkSession.builder \
    .config("spark.sql.catalog.rest",
           "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.rest.type", "rest") \
    .config("spark.sql.catalog.rest.uri",
           "http://iceberg-rest-catalog:8181") \
    .config("spark.sql.catalog.rest.warehouse",
           "s3://my-bucket/warehouse") \
    .config("spark.sql.catalog.rest.token", "my-auth-token") \
    .getOrCreate()

# Same table API — just different catalog prefix
spark.sql("SELECT * FROM rest.analytics.orders").show()

-- Basic compaction (merge small files)
CALL spark_catalog.system.rewrite_data_files('db.orders');

-- Compaction with target file size (256MB)
CALL spark_catalog.system.rewrite_data_files(
    table => 'db.orders',
    options => map(
        'target-file-size-bytes', '268435456',
        'min-file-size-bytes',    '67108864'
    )
);

-- Sort-order compaction (like ZORDER) — great for range queries
CALL spark_catalog.system.rewrite_data_files(
    table   => 'db.orders',
    strategy => 'sort',
    sort_order => 'order_date ASC NULLS LAST, customer_id ASC'
);

-- Expire snapshots older than 7 days, keep last 3
CALL spark_catalog.system.expire_snapshots(
    table        => 'db.orders',
    older_than   => TIMESTAMP '2024-01-08 00:00:00',
    retain_last  => 3
);

-- Also deletes orphan data files not referenced by kept snapshots

-- Rewrite and merge small manifest files
CALL spark_catalog.system.rewrite_manifests('db.orders');

-- Use native Avro writer (avoids Spark job overhead)
CALL spark_catalog.system.rewrite_manifests(
    table         => 'db.orders',
    use_caching   => 'true'
);

CALL spark_catalog.system.delete_orphan_files(
    table      => 'db.orders',
    older_than => TIMESTAMP '2024-01-01 00:00:00'
);

# EMR Bootstrap — add Iceberg jars via spark-defaults.conf
# or pass --packages to spark-submit

spark = SparkSession.builder \
    .appName("EMR_Iceberg") \
    .config("spark.sql.extensions",
           "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.glue",
           "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.glue.catalog-impl",
           "org.apache.iceberg.aws.glue.GlueCatalog") \
    .config("spark.sql.catalog.glue.warehouse",
           "s3://my-data-lake/iceberg/") \
    .config("spark.sql.catalog.glue.io-impl",
           "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.sql.catalog.glue.aws.region", "us-east-1") \
    .getOrCreate()

# Create Glue database
spark.sql("CREATE DATABASE IF NOT EXISTS glue.analytics")

# Create Iceberg table visible in Glue + Athena
spark.sql("""
    CREATE TABLE IF NOT EXISTS glue.analytics.sales (
        sale_id   BIGINT,
        sale_date DATE,
        product   STRING,
        revenue   DOUBLE
    ) USING iceberg
    PARTITIONED BY (months(sale_date))
""")

# Write data
df.writeTo("glue.analytics.sales").append()

-- Standard query
SELECT * FROM analytics.sales
WHERE sale_date >= DATE('2024-01-01');

-- Time travel in Athena
SELECT * FROM analytics.sales
FOR TIMESTAMP AS OF TIMESTAMP '2024-01-15 10:00:00 UTC';

-- MERGE in Athena (v3)
MERGE INTO analytics.sales t
USING updates u ON t.sale_id = u.sale_id
WHEN MATCHED THEN UPDATE SET revenue = u.revenue
WHEN NOT MATCHED THEN INSERT *;

aws emr-serverless start-job-run \
  --application-id app-123 \
  --execution-role-arn arn:aws:iam::123:role/emr-role \
  --job-driver '{
    "sparkSubmit": {
      "entryPoint": "s3://my-bucket/scripts/iceberg_job.py",
      "sparkSubmitParameters": "--conf spark.jars=/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar"
    }
  }'