# Databricks notebook source
# MAGIC %md
# MAGIC # Lakehouse Digest — Production Health-Check Notebook
# MAGIC
# MAGIC A runnable diagnostic toolkit drawn from Lakehouse Digest issues 001–009.
# MAGIC Run each section against the tables and jobs you own; every check tells you
# MAGIC what "healthy" looks like and links the issue that explains the fix.
# MAGIC
# MAGIC **Assumptions:** Databricks Runtime 14.x+ (DBR 16.x verified), Delta Lake tables,
# MAGIC Unity Catalog or Hive metastore. Read-only checks unless marked otherwise —
# MAGIC nothing here deletes data.
# MAGIC
# MAGIC Last verified: September 27, 2026 · https://lakehousedigest.com

# COMMAND ----------

# MAGIC %md
# MAGIC ## 0. Setup — point these at your own objects
# MAGIC
# MAGIC Edit the widgets, then **Run All**. Every check below reads from these.

# COMMAND ----------

dbutils.widgets.text("catalog", "main", "Catalog")
dbutils.widgets.text("schema", "default", "Schema")
dbutils.widgets.text("table", "", "Table to diagnose (catalog.schema.table)")
dbutils.widgets.text("job_id", "", "Lakeflow Job ID (optional)")

CATALOG = dbutils.widgets.get("catalog")
SCHEMA = dbutils.widgets.get("schema")
TABLE = dbutils.widgets.get("table") or f"{CATALOG}.{SCHEMA}.events"
print(f"Diagnosing table: {TABLE}")

# COMMAND ----------

# MAGIC %md
# MAGIC ## 1. Small-files check — are you drowning in tiny files?
# MAGIC
# MAGIC **Healthy:** average file size ≥ 64–128 MB, file count roughly proportional to data volume.
# MAGIC **Sick:** thousands of files under 10 MB → slow listings, slow reads, merge pain.
# MAGIC
# MAGIC **Fix:** `OPTIMIZE` (+ `ZORDER`/liquid clustering where it fits).
# MAGIC Read: Issue 003 — https://lakehousedigest.com/issue-003.html

# COMMAND ----------

files_df = spark.sql(f"""
  SELECT
    COUNT(*) AS file_count,
    ROUND(SUM(size) / 1024 / 1024, 1) AS total_mb,
    ROUND(AVG(size) / 1024 / 1024, 2) AS avg_file_mb,
    ROUND(MIN(size) / 1024 / 1024, 2) AS min_file_mb,
    ROUND(MAX(size) / 1024 / 1024, 2) AS max_file_mb,
    SUM(CASE WHEN size < 10 * 1024 * 1024 THEN 1 ELSE 0 END) AS files_under_10mb
  FROM (
    SELECT size FROM delta.`{TABLE}`.files  -- replace with: DESCRIBE DETAIL + file listing for your path
  )
""") if False else None

# Portable version using DESCRIBE DETAIL + Delta log file listing:
detail = spark.sql(f"DESCRIBE DETAIL {TABLE}").collect()[0]
print(f"Table: {TABLE}")
print(f"  numFiles        : {detail['numFiles']}")
print(f"  sizeInBytes     : {round(detail['sizeInBytes']/1024/1024, 1)} MB")
if detail['numFiles']:
    avg_mb = detail['sizeInBytes'] / detail['numFiles'] / 1024 / 1024
    print(f"  avg file size   : {round(avg_mb, 2)} MB")
    if avg_mb < 16:
        print("  ⚠️  VERDICT: small-file problem likely — run OPTIMIZE (see Issue 003).")
    elif avg_mb < 64:
        print("  ⚠️  VERDICT: borderline — OPTIMIZE during the next maintenance window.")
    else:
        print("  ✅ VERDICT: file sizing looks healthy.")

# COMMAND ----------

# MAGIC %md
# MAGIC ## 2. VACUUM safety — will your cleanup break time travel?
# MAGIC
# MAGIC **Healthy:** `delta.deletedFileRetentionDuration` ≥ your longest time-travel / rollback need.
# MAGIC **Sick:** retention shorter than the window anyone depends on → `VACUUM` can orphan readers.
# MAGIC
# MAGIC **Rules that never change:**
# MAGIC - The retention safety check exists in **both** Databricks and open-source Delta Lake
# MAGIC   (`checkRetentionPeriodSafety`; disable flag `spark.databricks.delta.retentionDurationCheck.enabled`).
# MAGIC - Predictive-optimization auto-VACUUM honors **your configured** retention (7-day default,
# MAGIC   7-day minimum on VACUUM FULL) — it is not unconditionally 7 days.
# MAGIC
# MAGIC Read: Issue 005 — https://lakehousedigest.com/issue-005.html

# COMMAND ----------

props = {r['key']: r['value'] for r in spark.sql(f"SHOW TBLPROPERTIES {TABLE}").collect()}
retention = props.get('delta.deletedFileRetentionDuration', 'interval 7 days (default)')
print(f"delta.deletedFileRetentionDuration = {retention}")
print(f"delta.enableChangeDataFeed          = {props.get('delta.enableChangeDataFeed', 'false')}")
print()
print("Checklist:")
print("  [ ] No downstream job reads versions older than the retention window")
print("  [ ] No ML training snapshot pins an old version beyond retention")
print("  [ ] If CDF is on, consumers have caught up past the retention horizon")
print("If any box is unchecked, raise the retention BEFORE running VACUUM.")

# COMMAND ----------

# MAGIC %md
# MAGIC ## 3. Skew detector — is one key starving your join?
# MAGIC
# MAGIC **Healthy:** task durations roughly even across the stage.
# MAGIC **Sick:** a few tasks run 10x longer — classic skew. Remember: AQE **can** split a
# MAGIC hot-key partition into roughly even tasks (replicating the matching side) when the
# MAGIC join/shuffle layout is eligible — salting is the fallback, not the first move.
# MAGIC
# MAGIC **Quick key-distribution probe** (run on the big side of your join):
# MAGIC
# MAGIC Read: Issue 004 — https://lakehousedigest.com/issue-004.html

# COMMAND ----------

# MAGIC %md
# MAGIC ```sql
# MAGIC -- Replace join_key / your_table. Top keys by row count:
# MAGIC SELECT join_key, COUNT(*) AS n
# MAGIC FROM <your_big_table>
# MAGIC GROUP BY join_key
# MAGIC ORDER BY n DESC
# MAGIC LIMIT 20;
# MAGIC ```
# MAGIC
# MAGIC **Read the output:** if the top key holds >5–10% of rows, treat the join as skewed.
# MAGIC Path A: check AQE eligibility (broadcast or shuffle-hash with supported join type) and let AQE split it.
# MAGIC Path B: salt the key as the fallback. Validate per-path — and only trust
# MAGIC `input.count() == result.count()` when the dimension key is **unique**
# MAGIC (duplicate dimension keys multiply rows in a left join).

# COMMAND ----------

# MAGIC %md
# MAGIC ## 4. MERGE source dedupe — will your upsert throw?
# MAGIC
# MAGIC `DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE` means your source has
# MAGIC **more than one row per merge key**. The fix is deterministic: reduce to one row
# MAGIC per key with `ROW_NUMBER()` + a tiebreaker **before** the MERGE.
# MAGIC
# MAGIC Read: Issue 006 — https://lakehousedigest.com/issue-006.html

# COMMAND ----------

# MAGIC %md
# MAGIC ```sql
# MAGIC -- 4a. Detect duplicates in your CDC/source for merge key (id):
# MAGIC SELECT id, COUNT(*) AS n
# MAGIC FROM <cdc_source>
# MAGIC GROUP BY id
# MAGIC HAVING COUNT(*) > 1
# MAGIC LIMIT 20;
# MAGIC
# MAGIC -- 4b. Deterministic dedupe: one row per key, newest event wins,
# MAGIC --     event_id breaks ties so the result is stable run-to-run:
# MAGIC WITH ranked AS (
# MAGIC   SELECT *,
# MAGIC     ROW_NUMBER() OVER (
# MAGIC       PARTITION BY id
# MAGIC       ORDER BY event_time DESC, event_id DESC
# MAGIC     ) AS rn
# MAGIC   FROM <cdc_source>
# MAGIC )
# MAGIC MERGE INTO <target> t
# MAGIC USING (SELECT * FROM ranked WHERE rn = 1) s
# MAGIC   ON t.id = s.id
# MAGIC WHEN MATCHED THEN UPDATE SET *
# MAGIC WHEN NOT MATCHED THEN INSERT *;
# MAGIC ```
# MAGIC
# MAGIC **Streaming note:** in `foreachBatch`, make the batch upsert idempotent
# MAGIC (same batch replayed = same result) so checkpoint replays can't double-apply.

# COMMAND ----------

# MAGIC %md
# MAGIC ## 5. Streaming checkpoint — can this stream actually recover?
# MAGIC
# MAGIC **Healthy:** checkpoint location is stable, exclusive to one stream, on durable storage.
# MAGIC **Sick:** checkpoint deleted/moved, shared between streams, or on ephemeral disk →
# MAGIC the "offset that vanished" — recovery replays everything or silently skips.
# MAGIC
# MAGIC Read: Issue 007 — https://lakehousedigest.com/issue-007.html

# COMMAND ----------

# MAGIC %md
# MAGIC ```python
# MAGIC # 5a. List active streams and their checkpoint locations:
# MAGIC for s in spark.streams.active:
# MAGIC     print(s.name, "->", s.lastProgress.get("checkpointLocation") if hasattr(s, "lastProgress") else s.id)
# MAGIC
# MAGIC # 5b. Checklist (answer from your job config, not guesswork):
# MAGIC #  [ ] checkpointLocation is on durable cloud storage (not /tmp, not DBFS ephemeral)
# MAGIC #  [ ] exactly ONE stream writes to each checkpoint dir
# MAGIC #  [ ] the checkpoint dir is never deleted between restarts
# MAGIC #  [ ] source retention (e.g. Kafka) covers your max expected downtime
# MAGIC ```

# COMMAND ----------

# MAGIC %md
# MAGIC ## 6. Driver-safety probe — collect() vs. broadcast runaway
# MAGIC
# MAGIC Two different failures, two different fixes — diagnose before changing the limit:
# MAGIC
# MAGIC | Symptom | Likely cause | Fix order |
# MAGIC |---|---|---|
# MAGIC | `Total size of serialized results ... bigger than spark.driver.maxResultSize` on `collect()`/`toPandas()` | driver collecting too much | 1. collect less 2. raise limit **last** |
# MAGIC | Same error during a **join** (no collect in your code) | runaway broadcast | 1. fix the broadcast (`spark.sql.autoBroadcastJoinThreshold`, explicit hints) 2. raise limit last |
# MAGIC
# MAGIC Read: Issue 002 — https://lakehousedigest.com/issue-002.html

# COMMAND ----------

# MAGIC %md
# MAGIC ```sql
# MAGIC -- 6a. Find the runaway: which side of the join is exploding?
# MAGIC -- Check the Spark UI SQL tab: look for BroadcastExchange with huge size,
# MAGIC -- or EXPLAIN the query and inspect the estimated sizes per side.
# MAGIC EXPLAIN <your_join_query>;
# MAGIC ```
# MAGIC
# MAGIC **Rule:** raising `spark.driver.maxResultSize` without diagnosing first just moves
# MAGIC the OutOfMemoryError somewhere more expensive.

# COMMAND ----------

# MAGIC %md
# MAGIC ## 7. Job repair readiness — when the 3am page comes
# MAGIC
# MAGIC Lakeflow Jobs: **repair only the failed tasks + their dependents** — don't rerun
# MAGIC the whole DAG. But first: fix the cause, then repair; and make every task's
# MAGIC writes idempotent so a retry can't double-apply.
# MAGIC
# MAGIC Read: Issue 001 — https://lakehousedigest.com/issue-001.html
# MAGIC
# MAGIC ```
# MAGIC Repair checklist:
# MAGIC   [ ] Root cause identified in the failed task's logs (not just "it failed")
# MAGIC   [ ] Cause fixed (code, config, or upstream data) — BEFORE hitting repair
# MAGIC   [ ] Task writes are idempotent (overwrite / merge-by-key, not blind append)
# MAGIC   [ ] Repair scope = failed tasks + downstream dependents only
# MAGIC ```

# COMMAND ----------

# MAGIC %md
# MAGIC ## Done — what to do with the findings
# MAGIC
# MAGIC 1. Anything flagged ⚠️ gets a ticket with the linked issue attached.
# MAGIC 2. Re-run this notebook monthly or after any incident — it's a regression net.
# MAGIC 3. New here? Start at https://lakehousedigest.com — one practical Databricks
# MAGIC    tactic every day: tip → implementation → Q&A.