🌲 Partition Strategy Decision Tree¶

# This query benefits from partition pruning
df = spark.sql("""
    SELECT machine_id, SUM(coin_in) as total_coin_in
    FROM bronze.slot_telemetry
    WHERE gaming_date = '2026-04-27'
    GROUP BY machine_id
""")
# Only files in gaming_date=2026-04-27/ are read

# This query does NOT benefit from partition pruning
df = spark.sql("""
    SELECT gaming_date, SUM(coin_in) as total_coin_in
    FROM bronze.slot_telemetry
    WHERE machine_id = 'SLOT-001'
    GROUP BY gaming_date
""")
# All partitions must be scanned (filter is not on partition column)
# For this pattern, use ZORDER BY (machine_id) instead

def evaluate_partition_candidates(table_name: str, candidate_columns: list) -> None:
    """Analyze columns for partition suitability."""
    df = spark.table(table_name)
    total_rows = df.count()
    detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]
    total_size_mb = detail.sizeInBytes / (1024 ** 2)

    print(f"Table: {table_name}")
    print(f"Total rows: {total_rows:,}")
    print(f"Total size: {total_size_mb:,.0f} MB")
    print(f"{'─' * 70}")

    for col_name in candidate_columns:
        stats = df.groupBy(col_name).count().agg(
            F.count("*").alias("distinct_values"),
            F.min("count").alias("min_rows"),
            F.max("count").alias("max_rows"),
            F.avg("count").alias("avg_rows"),
            F.stddev("count").alias("stddev_rows"),
        ).collect()[0]

        distinct = stats.distinct_values
        avg_partition_size_mb = total_size_mb / distinct if distinct > 0 else 0
        skew_ratio = stats.max_rows / stats.avg_rows if stats.avg_rows > 0 else 0

        print(f"\nColumn: {col_name}")
        print(f"  Distinct values: {distinct:,}")
        print(f"  Avg partition size: {avg_partition_size_mb:,.0f} MB")
        print(f"  Row distribution: min={stats.min_rows:,}, max={stats.max_rows:,}, avg={stats.avg_rows:,.0f}")
        print(f"  Skew ratio: {skew_ratio:.1f}x")

        # Assessment
        if distinct > 100_000:
            print(f"  ❌ AVOID: Too many distinct values — use ZORDER or liquid clustering")
        elif avg_partition_size_mb < 10:
            print(f"  ⚠️ WARNING: Partitions too small ({avg_partition_size_mb:.0f} MB) — over-partitioning risk")
        elif avg_partition_size_mb > 5_000:
            print(f"  ⚠️ WARNING: Partitions too large ({avg_partition_size_mb:,.0f} MB) — consider finer granularity")
        elif skew_ratio > 10:
            print(f"  ⚠️ WARNING: High skew ({skew_ratio:.0f}x) — some partitions much larger than others")
        else:
            print(f"  ✅ GOOD candidate for partitioning")

# Usage
from pyspark.sql import functions as F
evaluate_partition_candidates(
    "bronze.slot_telemetry",
    ["gaming_date", "floor_zone", "machine_type", "denomination", "machine_id"]
)

def detect_over_partitioning(table_name: str) -> None:
    """Detect if a table is over-partitioned."""
    detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]
    num_files = detail.numFiles
    total_size_mb = detail.sizeInBytes / (1024 ** 2)

    if num_files == 0:
        print(f"{table_name}: Empty table")
        return

    avg_file_mb = total_size_mb / num_files
    print(f"Table: {table_name}")
    print(f"  Files: {num_files:,}")
    print(f"  Total size: {total_size_mb:,.0f} MB")
    print(f"  Avg file size: {avg_file_mb:.1f} MB")

    if avg_file_mb < 10:
        print(f"  🔴 OVER-PARTITIONED: Average file size {avg_file_mb:.1f} MB is far below 128 MB target")
        print(f"  ⮕ Consider removing partitioning or reducing partition granularity")
    elif avg_file_mb < 32:
        print(f"  🟡 BORDERLINE: Run OPTIMIZE to compact files")
    else:
        print(f"  ✅ File sizes are acceptable")

# Option 1: Remove partitioning and rebuild
df = spark.table("bronze.over_partitioned_table")
df.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("bronze.properly_sized_table")
# No .partitionBy() — let OPTIMIZE and V-Order handle it

# Option 2: Reduce partition granularity
# Before: partitionBy("gaming_date", "machine_id")  ← too fine
# After: partitionBy("gaming_date")  ← appropriate granularity
df.write.format("delta") \
    .mode("overwrite") \
    .partitionBy("gaming_date") \
    .saveAsTable("bronze.slot_telemetry_v2")

# Option 3: Switch to liquid clustering
spark.sql("""
    CREATE TABLE bronze.slot_telemetry_v3
    CLUSTER BY (gaming_date, floor_zone)
    AS SELECT * FROM bronze.over_partitioned_table
""")

def detect_under_partitioning(table_name: str, expected_filter_col: str) -> None:
    """Detect if a table would benefit from partitioning."""
    detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]
    total_size_gb = detail.sizeInBytes / (1024 ** 3)
    partitioning = detail.partitionColumns

    print(f"Table: {table_name}")
    print(f"  Total size: {total_size_gb:.1f} GB")
    print(f"  Current partitioning: {partitioning if partitioning else 'NONE'}")

    if total_size_gb > 1 and not partitioning:
        distinct_values = spark.table(table_name) \
            .select(expected_filter_col).distinct().count()
        per_partition_gb = total_size_gb / distinct_values if distinct_values else total_size_gb

        print(f"  🔴 UNDER-PARTITIONED: {total_size_gb:.0f} GB with no partitioning")
        print(f"  ⮕ Partitioning by '{expected_filter_col}' ({distinct_values} values) would give ~{per_partition_gb:.1f} GB per partition")

        if 0.1 < per_partition_gb < 2:
            print(f"  ✅ '{expected_filter_col}' is an excellent partition candidate")
        elif per_partition_gb < 0.1:
            print(f"  ⚠️ Consider coarser granularity (e.g., month instead of day)")
        else:
            print(f"  ⚠️ Consider finer granularity or multi-column partitioning")

# Daily partitioning (most common for casino/federal data)
df.write.format("delta") \
    .mode("append") \
    .partitionBy("gaming_date") \
    .saveAsTable("bronze.slot_telemetry")

# Monthly partitioning (for smaller datasets)
from pyspark.sql.functions import date_format

df_with_month = df.withColumn("report_month", date_format("report_date", "yyyy-MM"))
df_with_month.write.format("delta") \
    .mode("append") \
    .partitionBy("report_month") \
    .saveAsTable("bronze.usda_crop_production")

# Year partitioning (for very small daily volumes)
df_with_year = df.withColumn("report_year", year("report_date"))
df_with_year.write.format("delta") \
    .mode("append") \
    .partitionBy("report_year") \
    .saveAsTable("bronze.doi_land_survey")

# Example: Date + floor_zone (5 zones × 365 days = 1,825 partitions/year)
# Each partition: ~300 MB (good for a 550 GB/year table)
df.write.format("delta") \
    .mode("append") \
    .partitionBy("gaming_date", "floor_zone") \
    .saveAsTable("silver.slot_telemetry_cleansed")

# Anti-example: Date + machine_id (500 machines × 365 days = 182,500 partitions)
# Each partition: ~3 MB (OVER-PARTITIONED — do NOT do this)

# The first partition column creates the top-level directory structure
# Order from lowest to highest cardinality for best file organization

# ✅ GOOD: floor_zone (5 values) first, then gaming_date (365 values)
.partitionBy("floor_zone", "gaming_date")
# Result: 5 top-level directories, each with 365 subdirectories

# 🟡 ACCEPTABLE: gaming_date first (365 values), then floor_zone (5 values)
.partitionBy("gaming_date", "floor_zone")
# Result: 365 top-level directories, each with 5 subdirectories

# Method 1: EXPLAIN shows partition pruning
spark.sql("""
    EXPLAIN EXTENDED
    SELECT * FROM bronze.slot_telemetry
    WHERE gaming_date = '2026-04-27'
""").show(truncate=False)
# Look for: PartitionFilters: [gaming_date = 2026-04-27]

# Method 2: Check actual files read via Spark UI metrics
df = spark.sql("""
    SELECT machine_id, SUM(coin_in) as total
    FROM bronze.slot_telemetry
    WHERE gaming_date = '2026-04-27'
    GROUP BY machine_id
""")
df.explain(True)  # Look for "PushedFilters" and "PartitionFilters"

def verify_partition_pruning(table_name: str, filter_condition: str) -> dict:
    """Verify that partition pruning is effective for a given query."""
    # Get total files
    total_detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]
    total_files = total_detail.numFiles

    # Get the query plan
    query = f"SELECT * FROM {table_name} WHERE {filter_condition}"
    plan = spark.sql(query)._jdf.queryExecution().executedPlan().toString()

    # Check for partition filter in plan
    has_partition_filter = "PartitionFilters" in plan and filter_condition.split("=")[0].strip() in plan

    # Count files actually read (approximate via plan)
    result = {
        "table": table_name,
        "filter": filter_condition,
        "total_files": total_files,
        "partition_pruning_detected": has_partition_filter,
        "plan_snippet": plan[:500]
    }

    if has_partition_filter:
        print(f"✅ Partition pruning IS active for: {filter_condition}")
    else:
        print(f"⚠️ Partition pruning NOT detected for: {filter_condition}")
        print(f"   Check if '{filter_condition.split('=')[0].strip()}' is a partition column")

    return result

# Test
verify_partition_pruning("bronze.slot_telemetry", "gaming_date = '2026-04-27'")

# New table with liquid clustering
spark.sql("""
    CREATE TABLE silver.player_transactions_clustered
    CLUSTER BY (player_id, transaction_date)
    AS SELECT * FROM silver.player_transactions
""")

# Change clustering columns (not possible with partitioning!)
spark.sql("""
    ALTER TABLE silver.player_transactions_clustered
    CLUSTER BY (transaction_type, transaction_date)
""")

# Trigger clustering
spark.sql("OPTIMIZE silver.player_transactions_clustered")

# Optimize specific partitions only (avoid full-table rewrite)
spark.sql("""
    OPTIMIZE bronze.slot_telemetry
    WHERE gaming_date >= '2026-04-20' AND gaming_date <= '2026-04-27'
""")

# Optimize with ZORDER within partitions
spark.sql("""
    OPTIMIZE silver.slot_telemetry_cleansed
    WHERE gaming_date >= current_date() - INTERVAL 7 DAYS
    ZORDER BY (machine_id)
""")

# Partition-aware VACUUM
spark.sql("VACUUM bronze.slot_telemetry RETAIN 168 HOURS")
# VACUUM respects partitions — only removes old files within each partition

# ❌ WRONG: Partitioning a 500 MB table
df.write.format("delta") \
    .partitionBy("gaming_date", "floor_zone") \
    .saveAsTable("gold.daily_summary")
# Creates hundreds of tiny files for a small table

# ✅ CORRECT: No partitioning for small tables
df.write.format("delta") \
    .saveAsTable("gold.daily_summary")
# OPTIMIZE + V-Order is sufficient

# ❌ WRONG: UUID as partition column
df.write.format("delta") \
    .partitionBy("transaction_id") \
    .saveAsTable("bronze.transactions")
# 1 file per transaction = millions of tiny files

# ✅ CORRECT: Use ZORDER or liquid clustering for high-cardinality
df.write.format("delta") \
    .saveAsTable("bronze.transactions")
spark.sql("OPTIMIZE bronze.transactions ZORDER BY (transaction_id)")

# ❌ WRONG: Partition and forget
df.write.format("delta") \
    .mode("append") \
    .partitionBy("gaming_date") \
    .saveAsTable("bronze.slot_telemetry")
# After 100 appends, each partition has 100 tiny files

# ✅ CORRECT: Schedule regular OPTIMIZE
spark.sql("""
    OPTIMIZE bronze.slot_telemetry
    WHERE gaming_date >= current_date() - INTERVAL 7 DAYS
""")

# ❌ WRONG: Trying to change partition columns on existing table
spark.sql("ALTER TABLE bronze.slot_telemetry SET PARTITIONED BY (floor_zone)")
# This is NOT supported in Delta Lake

# ✅ CORRECT: Create new table with desired partitioning and migrate
df = spark.table("bronze.slot_telemetry")
df.write.format("delta") \
    .mode("overwrite") \
    .partitionBy("floor_zone") \
    .saveAsTable("bronze.slot_telemetry_v2")

# Or switch to liquid clustering (can be changed later)
spark.sql("""
    CREATE TABLE bronze.slot_telemetry_v3
    CLUSTER BY (gaming_date, floor_zone)
    AS SELECT * FROM bronze.slot_telemetry
""")