# Default: checkpoint every 10 commits
# You can change this per table property

spark.sql("""
  ALTER TABLE delta.`/tmp/my_table`
  SET TBLPROPERTIES ('delta.checkpointInterval' = '5')
""")
# Now a checkpoint is created every 5 commits (more frequent = faster reads)

# Manually trigger a checkpoint
from delta.tables import DeltaTable
dt = DeltaTable.forPath(spark, "/tmp/my_table")
dt.toDF()  # just accessing it may trigger checkpoint

# Inspect the _last_checkpoint file (tells Delta where to start)
# Location: _delta_log/_last_checkpoint  (a tiny JSON)
last_chk = spark.read.json("/tmp/my_table/_delta_log/_last_checkpoint")
last_chk.show()
# Output: {"version": 10, "size": 42}
# "version" = latest checkpoint version
# "size" = number of actions in that checkpoint

# Two jobs writing to the same Delta table simultaneously
# Job A: appending new sales records
# Job B: appending new user records
# Delta handles this gracefully — both succeed if they don't conflict

# Job A
df_sales = spark.createDataFrame([(101, "sale", 500)], ["id", "type", "amount"])
df_sales.write.format("delta").mode("append").save("/tmp/events")

# Job B (running at the same time as Job A)
# Delta detects no conflict (different files) → both commits succeed
df_users = spark.createDataFrame([(102, "user", 0)], ["id", "type", "amount"])
df_users.write.format("delta").mode("append").save("/tmp/events")

# When does OCC FAIL? When two writers modify the SAME files.
# Example: Two concurrent UPDATE statements on overlapping rows
# Delta raises: ConcurrentModificationException
# The losing writer must retry with fresh data

# Configure isolation level (Serializable = strictest, WriteSerializable = default)
# WriteSerializable: allows concurrent reads with writes
# Serializable: full isolation, no concurrent reads during writes
spark.sql("""
  ALTER TABLE delta.`/tmp/events`
  SET TBLPROPERTIES ('delta.isolationLevel' = 'Serializable')
""")

# Every read pins to a specific snapshot (version)

# Read the CURRENT snapshot (latest committed version)
df_current = spark.read.format("delta").load("/tmp/people_delta")
df_current.show()

# Read a specific HISTORICAL snapshot → Time Travel!
# By version number:
df_v0 = spark.read.format("delta") \
    .option("versionAsOf", 0) \
    .load("/tmp/people_delta")
df_v0.show()

# By timestamp:
df_ts = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-15 10:00:00") \
    .load("/tmp/people_delta")

# SQL syntax for time travel:
spark.sql("SELECT * FROM people VERSION AS OF 0").show()
spark.sql("SELECT * FROM people TIMESTAMP AS OF '2024-01-15'").show()

# This is possible because removed files stay on disk
# until VACUUM cleans them up (default 7-day retention)

# View all versions available for time travel:
dt = DeltaTable.forPath(spark, "/tmp/people_delta")
dt.history().select("version", "timestamp", "operation").show()

from delta.tables import DeltaTable
from pyspark.sql.functions import *

# ── Setup: Create the target Delta table (customers) ──
customers = [(1, "Alice", "alice@old.com"),
             (2, "Bob",   "bob@old.com"),
             (3, "Carol", "carol@old.com")]
df_target = spark.createDataFrame(customers, ["id", "name", "email"])
df_target.write.format("delta").save("/tmp/customers")

# ── Source: incoming updates/new records ──
updates = [(1, "Alice", "alice@new.com"),   # MATCHED → UPDATE email
           (4, "Dave",  "dave@example.com")]  # NOT MATCHED → INSERT
df_source = spark.createDataFrame(updates, ["id", "name", "email"])

# ── MERGE using DeltaTable API ──
target = DeltaTable.forPath(spark, "/tmp/customers")

target.alias("t") \
  .merge(
    df_source.alias("s"),
    "t.id = s.id"          # join condition
  ) \
  .whenMatchedUpdate(set={
    "email": "s.email"       # update email when ID matches
  }) \
  .whenNotMatchedInsert(values={
    "id":    "s.id",
    "name":  "s.name",
    "email": "s.email"
  }) \
  .execute()

DeltaTable.forPath(spark, "/tmp/customers").toDF().show()
# id=1 email updated; id=4 inserted; id=2,3 unchanged

# Full upsert + delete (only keep rows present in source)
target.alias("t") \
  .merge(df_source.alias("s"), "t.id = s.id") \
  .whenMatchedUpdateAll() \            # update ALL columns when matched
  .whenNotMatchedInsertAll() \         # insert ALL columns when new
  .whenNotMatchedBySourceDelete() \    # delete rows not in source
  .execute()

# SQL equivalent (easier to read):
spark.sql("""
  MERGE INTO customers t
  USING updates s
  ON t.id = s.id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *
  WHEN NOT MATCHED BY SOURCE THEN DELETE
""")

# Conditional MERGE — only update if something changed
target.alias("t") \
  .merge(df_source.alias("s"), "t.id = s.id") \
  .whenMatchedUpdate(
    condition="t.email != s.email",   # only update if email changed
    set={"email": "s.email"}
  ) \
  .whenNotMatchedInsertAll() \
  .execute()

from delta.tables import DeltaTable

dt = DeltaTable.forPath(spark, "/tmp/customers")

# Simple UPDATE: change email for id=1
dt.update(
    condition="id = 1",
    set={"email": "'newemail@example.com'"}
)

# UPDATE with column expression
dt.update(
    condition="age < 18",
    set={"category": "'minor'", "discount": "0.5"}
)

# UPDATE using Spark functions (col, lit, etc.)
from pyspark.sql.functions import col, upper
dt.update(
    condition=col("country") == "US",
    set={"country_code": lit("USD")}
)

# SQL syntax:
spark.sql("""
  UPDATE delta.`/tmp/customers`
  SET email = 'updated@example.com'
  WHERE id = 1
""")

from delta.tables import DeltaTable

dt = DeltaTable.forPath(spark, "/tmp/customers")

# Delete specific rows
dt.delete("id = 3")

# Delete with complex condition
dt.delete(col("is_deleted") == True)

# Delete all rows (not recommended — use TRUNCATE instead)
dt.delete()

# GDPR Right-to-Erasure: delete a user's data across all tables
dt.delete(col("user_id") == 12345)
# After deletion, the data is gone from queries
# but physically stays on disk until VACUUM

# SQL syntax:
spark.sql("""
  DELETE FROM delta.`/tmp/customers`
  WHERE id = 3
""")

# Step 1: Enable CDF on an existing table
spark.sql("""
  ALTER TABLE delta.`/tmp/customers`
  SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

# Enable at creation time:
df.write.format("delta") \
    .option("delta.enableChangeDataFeed", "true") \
    .save("/tmp/customers_cdf")

# Step 2: Make some changes
dt = DeltaTable.forPath(spark, "/tmp/customers_cdf")
dt.update("id = 1", {"email": "'changed@example.com'"})
dt.delete("id = 3")

# Step 3: Read the change feed — by version range
changes = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 1) \
    .option("endingVersion", 5) \
    .load("/tmp/customers_cdf")

changes.show()
# Output columns: id, name, email, _change_type, _commit_version, _commit_timestamp
# _change_type values:
#   "insert"             → new row added
#   "update_preimage"    → row BEFORE update
#   "update_postimage"   → row AFTER update
#   "delete"             → row that was deleted

# Read by timestamp instead of version
changes_ts = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingTimestamp", "2024-01-15 10:00:00") \
    .load("/tmp/customers_cdf")

# Streaming CDF: process changes as a stream!
stream_changes = spark.readStream.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 0) \
    .load("/tmp/customers_cdf")

# Process inserts only in streaming:
inserts_only = stream_changes.filter(col("_change_type") == "insert")

from delta.tables import DeltaTable

dt = DeltaTable.forPath(spark, "/tmp/customers")

# DRY RUN first — see what VACUUM WOULD delete (no actual deletion)
dt.vacuum(retentionHours=168)  # default 168 hours = 7 days

# Actually run VACUUM with 7-day retention
dt.vacuum(retentionHours=168)

# Shorter retention (e.g., GDPR erasure) — requires disabling safety check
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", False)
dt.vacuum(retentionHours=0)  # immediately delete all unreferenced files
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", True)

# SQL syntax:
spark.sql("VACUUM delta.`/tmp/customers` RETAIN 168 HOURS")
spark.sql("VACUUM delta.`/tmp/customers` DRY RUN")  # list files to be deleted

# Set default retention per table:
spark.sql("""
  ALTER TABLE delta.`/tmp/customers`
  SET TBLPROPERTIES ('delta.deletedFileRetentionDuration' = 'interval 30 days')
""")

# Basic OPTIMIZE — compacts small files within each partition
spark.sql("OPTIMIZE delta.`/tmp/events`")

# OPTIMIZE with ZORDER BY — co-locate data by column values
# Queries filtering on these columns scan far fewer files
spark.sql("OPTIMIZE delta.`/tmp/events` ZORDER BY (user_id, event_date)")

# OPTIMIZE a specific partition only (faster for large tables)
spark.sql("""
  OPTIMIZE delta.`/tmp/events`
  WHERE event_date = '2024-01-15'
  ZORDER BY (user_id)
""")

# DeltaTable API:
from delta.tables import DeltaTable
dt = DeltaTable.forPath(spark, "/tmp/events")
dt.optimize().executeCompaction()
dt.optimize().executeZOrderBy("user_id", "event_date")

# Configure target file size (default 1GB = 1073741824 bytes)
spark.conf.set("spark.databricks.delta.optimize.maxFileSize", 134217728)
# Above sets target to 128MB (better for smaller tables)

# Enable Bloom filter on a column (great for UUID/email/string ID lookups)
spark.sql("""
  CREATE TABLE IF NOT EXISTS events (
    event_id STRING,
    user_id  STRING,
    event_ts TIMESTAMP,
    payload  STRING
  ) USING DELTA
  TBLPROPERTIES (
    'delta.dataSkippingNumIndexedCols' = '4',
    'delta.bloomFilter.columns' = 'event_id, user_id',
    'delta.bloomFilter.fpp'     = '0.1',     -- false positive probability 10%
    'delta.bloomFilter.maxExpectedFpp' = '0.1'
  )
""")

# Alternatively, add bloom filter to existing table:
spark.sql("""
  ALTER TABLE delta.`/tmp/events`
  SET TBLPROPERTIES (
    'delta.bloomFilter.columns' = 'event_id',
    'delta.bloomFilter.fpp' = '0.01'    -- 1% false positive rate
  )
""")

# After OPTIMIZE, bloom filter indexes are built and stored per file
# Query: Spark checks bloom filter → skips files that definitely don't have the value
spark.sql("SELECT * FROM events WHERE event_id = '550e8400-e29b-41d4-a716-446655440000'").show()
# Without bloom: scans all files
# With bloom (fpp=0.01): skips ~99% of files for point lookups

# Create table with Liquid Clustering (no PARTITIONED BY needed)
spark.sql("""
  CREATE TABLE events_clustered (
    event_id STRING,
    user_id  STRING,
    event_ts TIMESTAMP,
    country  STRING,
    amount   DOUBLE
  ) USING DELTA
  CLUSTER BY (user_id, country)    -- liquid clustering keys
""")

# Write data normally — no special options needed
df.write.format("delta").mode("append").saveAsTable("events_clustered")

# Run OPTIMIZE to apply clustering incrementally
spark.sql("OPTIMIZE events_clustered")
# Only newly added files are clustered — much faster than full ZORDER!

# Change clustering keys without rewriting the whole table:
spark.sql("ALTER TABLE events_clustered CLUSTER BY (country, event_ts)")
# Next OPTIMIZE will gradually re-cluster with new keys

# Create table with generated columns
spark.sql("""
  CREATE TABLE sales (
    sale_id    BIGINT,
    sale_ts    TIMESTAMP,
    amount     DOUBLE,
    -- Generated column: auto-extracted from sale_ts
    sale_date  DATE    GENERATED ALWAYS AS (CAST(sale_ts AS DATE)),
    sale_year  INT     GENERATED ALWAYS AS (YEAR(sale_ts)),
    sale_month INT     GENERATED ALWAYS AS (MONTH(sale_ts)),
    total_tax  DOUBLE  GENERATED ALWAYS AS (amount * 0.18)
  ) USING DELTA
  PARTITIONED BY (sale_year, sale_month)   -- partition on generated cols!
""")

# Insert WITHOUT the generated columns — Delta fills them automatically
df_sales = spark.createDataFrame([
    (1001, "2024-03-15 10:30:00", 500.0),
    (1002, "2024-04-20 14:45:00", 750.0)
], ["sale_id", "sale_ts", "amount"])

df_sales = df_sales.withColumn("sale_ts", col("sale_ts").cast("timestamp"))
df_sales.write.format("delta").mode("append").saveAsTable("sales")

# Read back — generated columns are there!
# sale_date, sale_year, sale_month, total_tax auto-filled
spark.sql("SELECT * FROM sales").show()

# Add CHECK constraints to an existing table
spark.sql("""
  ALTER TABLE sales
  ADD CONSTRAINT amount_positive CHECK (amount > 0)
""")

spark.sql("""
  ALTER TABLE sales
  ADD CONSTRAINT valid_status CHECK (status IN ('pending', 'completed', 'cancelled'))
""")

spark.sql("""
  ALTER TABLE sales
  ADD CONSTRAINT valid_date CHECK (sale_date >= '2020-01-01')
""")

# List all constraints on a table:
spark.sql("DESCRIBE DETAIL sales").select("properties").show(truncate=False)

# Try inserting a bad row — Delta REJECTS it
bad_df = spark.createDataFrame([(1003, "2024-01-01", -100.0)], ["sale_id", "sale_ts", "amount"])
try:
    bad_df.write.format("delta").mode("append").saveAsTable("sales")
except Exception as e:
    print(f"Write rejected: {e}")
# Error: DeltaInvariantViolationException: CHECK constraint violated: amount_positive

# Remove a constraint:
spark.sql("ALTER TABLE sales DROP CONSTRAINT amount_positive")

# Install delta-sharing client:
# pip install delta-sharing

import delta_sharing

# The recipient gets a profile file (JSON with server URL + token)
profile_file = "/path/to/recipient_profile.share"

# List all shares, schemas, and tables you have access to
client = delta_sharing.SharingClient(profile_file)
shares = client.list_shares()
tables = client.list_all_tables()
print(tables)

# Read a shared table into Pandas
df_pandas = delta_sharing.load_as_pandas(
    f"{profile_file}#share_name.schema_name.table_name"
)

# Read a shared table into Spark
df_spark = delta_sharing.load_as_spark(
    f"{profile_file}#share_name.schema_name.table_name"
)
df_spark.show()

# Provider side (Databricks Unity Catalog):
spark.sql("""
  CREATE SHARE customer_share;
  ALTER SHARE customer_share ADD TABLE main.sales.customers;
  GRANT SELECT ON SHARE customer_share TO RECIPIENT partner_org;
""")

# Existing Delta table has: id (INT), name (STRING), email (STRING)

# Case 1: Extra column in source → REJECTED by default
df_extra = spark.createDataFrame(
    [(1, "Alice", "alice@ex.com", "NYC")],
    ["id", "name", "email", "city"]   # 'city' not in table schema
)
try:
    df_extra.write.format("delta").mode("append").save("/tmp/customers")
except Exception as e:
    print("REJECTED:", e)
# AnalysisException: A schema mismatch detected when writing to Delta table.
# To enable schema migration, set spark.databricks.delta.schema.autoMerge.enabled=true

# Case 2: Wrong column type → REJECTED
df_wrong_type = spark.createDataFrame(
    [("one", "Alice", "alice@ex.com")],  # id should be INT, not STRING
    ["id", "name", "email"]
)
# This will also fail schema validation

# Case 3: Missing column → ACCEPTED (filled with null)
df_missing = spark.createDataFrame(
    [(5, "Eve")],
    ["id", "name"]   # 'email' missing → stored as NULL
)
df_missing.write.format("delta").mode("append").save("/tmp/customers")
# OK — missing columns are nulled out

# mergeSchema: add new columns to the table schema on write
df_with_new_col = spark.createDataFrame(
    [(6, "Frank", "frank@ex.com", "Engineer")],
    ["id", "name", "email", "job_title"]   # new: job_title
)

# Option 1: Write option
df_with_new_col.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save("/tmp/customers")
# Result: table now has id, name, email, job_title
# Old rows have job_title = NULL

# Option 2: Global session config (applies to all Delta writes in session)
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")

# ─────────────────────────────────────────────────────────────────────────────
# overwriteSchema: COMPLETELY REPLACE the table schema (destructive!)
# Use this when you want to change column types or drop columns
# ─────────────────────────────────────────────────────────────────────────────

df_new_schema = spark.createDataFrame(
    [(1, "Alice", 90000.0)],
    ["id", "name", "salary"]   # completely different schema
)

df_new_schema.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("/tmp/customers")
# WARNING: Old schema (email, job_title) is gone. Old data is removed.

from delta.tables import DeltaTable

# Access by path
dt = DeltaTable.forPath(spark, "/tmp/customers")

# Access by catalog name (Hive/Unity Catalog)
dt = DeltaTable.forName(spark, "main.sales.customers")

# ── history() — Full audit trail ──
history = dt.history()
history.select("version", "timestamp", "operation", 
               "operationParameters", "userMetadata").show(20, truncate=False)
# Shows: CREATE TABLE, WRITE, UPDATE, DELETE, MERGE, OPTIMIZE, etc.

# ── Time Travel (already covered in 20.4, but via DeltaTable) ──
df_v0  = spark.read.format("delta").option("versionAsOf", 0).load("/tmp/customers")
df_ts  = spark.read.format("delta").option("timestampAsOf", "2024-01-15").load("/tmp/customers")

# ── restoreToVersion() — Roll back to a previous state ──
# This is DESTRUCTIVE: creates a new commit that restores old state
dt.restoreToVersion(2)    # roll back to version 2

# ── restoreToTimestamp() ──
dt.restoreToTimestamp("2024-01-10 08:00:00")

# SQL equivalents:
spark.sql("RESTORE TABLE customers TO VERSION AS OF 2")
spark.sql("RESTORE TABLE customers TO TIMESTAMP AS OF '2024-01-10'")

# ── toDF() — Get the current data as a DataFrame ──
current_df = dt.toDF()
current_df.show()

# ── isDeltaTable() — Check if a path is a Delta table ──
print(DeltaTable.isDeltaTable(spark, "/tmp/customers"))   # True
print(DeltaTable.isDeltaTable(spark, "/tmp/plain_parquet"))  # False

# ── detail() — Table metadata ──
spark.sql("DESCRIBE DETAIL delta.`/tmp/customers`").show(vertical=True)
# Shows: numFiles, sizeInBytes, partitionColumns, createdAt, etc.