🔀 V-Order Tuning Deep Dive¶

# V-Order is ON by default in Fabric
# To verify or explicitly set:
spark.conf.get("spark.sql.parquet.vorder.enabled")  # "true"

# Explicitly enable (usually not needed)
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")

# Disable for specific write-heavy workloads
spark.conf.set("spark.sql.parquet.vorder.enabled", "false")

# Gold tables powering Power BI dashboards — V-Order is critical
df_gold = spark.sql("""
    SELECT
        gaming_date,
        floor_zone,
        machine_type,
        SUM(coin_in) as total_coin_in,
        SUM(coin_out) as total_coin_out,
        COUNT(DISTINCT machine_id) as active_machines,
        AVG(theoretical_hold_pct) as avg_hold_pct
    FROM silver.slot_telemetry_cleansed
    GROUP BY gaming_date, floor_zone, machine_type
""")

# V-Order applied automatically on write
df_gold.write.format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.slot_performance_daily")
# Direct Lake reads this table 3-5x faster than non-V-Ordered

# Dimension tables with low cardinality columns benefit most
df_dim_machine = spark.sql("""
    SELECT machine_id, manufacturer, model, denomination,
           floor_zone, section, install_date, status
    FROM silver.machine_master
""")
df_dim_machine.write.format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.dim_machine")
# V-Order groups similar values → excellent compression

# Streaming micro-batch: Disable V-Order during write
spark.conf.set("spark.sql.parquet.vorder.enabled", "false")

# Write streaming micro-batches without V-Order overhead
df_stream.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "Files/checkpoints/slot_stream/") \
    .toTable("bronze.slot_telemetry_stream")

# Re-enable V-Order for OPTIMIZE (applied periodically)
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")

# Ingest pipeline running every 5 minutes
# Disable V-Order during rapid ingestion
spark.conf.set("spark.sql.parquet.vorder.enabled", "false")

for batch in micro_batches:
    batch.write.format("delta").mode("append").saveAsTable("bronze.cage_transactions")

# Schedule OPTIMIZE with V-Order during off-peak hours
# This applies V-Order retroactively to all files
spark.sql("OPTIMIZE bronze.cage_transactions")
# V-Order is applied during OPTIMIZE by default

# Intermediate staging tables that exist for minutes
spark.conf.set("spark.sql.parquet.vorder.enabled", "false")

df_stage = df_raw.filter(col("is_valid") == True)
df_stage.write.format("delta").mode("overwrite").saveAsTable("staging.temp_validated")

# Process and drop — no need for V-Order optimization
df_final = spark.table("staging.temp_validated").join(...)
spark.sql("DROP TABLE staging.temp_validated")

# Optimize only recent partitions (avoid full-table rewrite)
spark.sql("""
    OPTIMIZE bronze.slot_telemetry
    WHERE gaming_date >= '2026-04-01'
""")

# Targeted optimization for specific high-query partitions
spark.sql("""
    OPTIMIZE gold.compliance_ctr_daily
    WHERE filing_date >= current_date() - INTERVAL 90 DAYS
""")

# Run OPTIMIZE as a scheduled notebook job (off-peak hours)
tables_to_optimize = [
    ("gold.slot_performance_daily", "gaming_date >= current_date() - INTERVAL 7 DAYS"),
    ("gold.player_loyalty_summary", None),  # Full optimize (small table)
    ("silver.player_transactions", "transaction_date >= current_date() - INTERVAL 30 DAYS"),
    ("bronze.slot_telemetry", "gaming_date >= current_date() - INTERVAL 3 DAYS"),
]

for table, where_clause in tables_to_optimize:
    sql = f"OPTIMIZE {table}"
    if where_clause:
        sql += f" WHERE {where_clause}"
    print(f"Optimizing: {sql}")
    result = spark.sql(sql)
    metrics = result.collect()[0]
    print(f"  Files added: {metrics.numFilesAdded}, removed: {metrics.numFilesRemoved}")

# Default target file size in Fabric: 128 MB
# Adjust for larger tables:
spark.conf.set("spark.microsoft.delta.optimizeWrite.fileSize", "268435456")  # 256 MB
spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", "true")

# For OPTIMIZE command:
spark.sql("""
    OPTIMIZE gold.slot_performance_daily
""")
# Target file size is controlled by:
# spark.databricks.delta.optimize.maxFileSize (default: 1GB for OPTIMIZE)

from delta.tables import DeltaTable

def analyze_file_sizes(table_name: str):
    """Analyze Delta table file sizes and fragmentation."""
    dt = DeltaTable.forName(spark, table_name)

    # Get file-level details
    files_df = spark.sql(f"""
        DESCRIBE DETAIL {table_name}
    """).collect()[0]

    print(f"Table: {table_name}")
    print(f"  Total size: {files_df.sizeInBytes / (1024**3):.2f} GB")
    print(f"  Number of files: {files_df.numFiles}")
    if files_df.numFiles > 0:
        avg_file_size = files_df.sizeInBytes / files_df.numFiles
        print(f"  Avg file size: {avg_file_size / (1024**2):.1f} MB")

        if avg_file_size < 32 * 1024 * 1024:  # < 32 MB
            print("  ⚠️ SMALL FILE PROBLEM — run OPTIMIZE")
        elif avg_file_size > 2 * 1024 * 1024 * 1024:  # > 2 GB
            print("  ⚠️ FILES TOO LARGE — consider repartitioning")
        else:
            print("  ✅ File sizes are healthy")

# Check all gold tables
gold_tables = spark.sql("SHOW TABLES IN gold").collect()
for t in gold_tables:
    analyze_file_sizes(f"gold.{t.tableName}")

# Run nightly or weekly depending on ingestion frequency
# Schedule: Every day at 2:00 AM (off-peak)

optimize_config = {
    "hot_tables": {  # Queried frequently, updated daily
        "tables": ["gold.slot_performance_daily", "gold.player_loyalty_summary"],
        "frequency": "daily",
        "where": "gaming_date >= current_date() - INTERVAL 7 DAYS",
        "zorder_cols": {"gold.slot_performance_daily": ["floor_zone", "machine_type"]},
    },
    "warm_tables": {  # Queried weekly, updated daily
        "tables": ["silver.player_transactions", "silver.slot_telemetry_cleansed"],
        "frequency": "weekly",
        "where": "transaction_date >= current_date() - INTERVAL 30 DAYS",
        "zorder_cols": {"silver.player_transactions": ["player_id"]},
    },
    "cold_tables": {  # Queried rarely, append-only
        "tables": ["bronze.slot_telemetry", "bronze.cage_transactions"],
        "frequency": "monthly",
        "where": None,
        "zorder_cols": {},
    },
}

# After a large batch load, compact immediately
def post_load_optimize(table_name: str, loaded_partition: str = None):
    """Run OPTIMIZE after a batch load completes."""
    detail_before = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]

    sql = f"OPTIMIZE {table_name}"
    if loaded_partition:
        sql += f" WHERE {loaded_partition}"

    result = spark.sql(sql)
    metrics = result.collect()[0]

    detail_after = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]

    print(f"Compaction results for {table_name}:")
    print(f"  Files: {detail_before.numFiles} → {detail_after.numFiles}")
    print(f"  Size: {detail_before.sizeInBytes/(1024**2):.0f} MB → {detail_after.sizeInBytes/(1024**2):.0f} MB")
    print(f"  Files added: {metrics.numFilesAdded}, removed: {metrics.numFilesRemoved}")

# Usage after loading today's slot telemetry
post_load_optimize("bronze.slot_telemetry", "gaming_date = '2026-04-27'")

# OPTIMIZE creates new files; old files remain for time travel
# VACUUM removes old files past retention period

# Default retention: 7 days (168 hours)
spark.sql("VACUUM gold.slot_performance_daily RETAIN 168 HOURS")

# For tables where time travel is less critical:
# WARNING: Reduces time travel capability
spark.sql("VACUUM bronze.slot_telemetry RETAIN 72 HOURS")

# Check space reclaimed
detail = spark.sql("DESCRIBE DETAIL gold.slot_performance_daily").collect()[0]
print(f"Current size after VACUUM: {detail.sizeInBytes / (1024**3):.2f} GB")

import time

def measure_query_performance(query: str, iterations: int = 5) -> dict:
    """Measure query performance with timing statistics."""
    times = []
    for i in range(iterations):
        # Clear cache for fair comparison
        spark.catalog.clearCache()

        start = time.time()
        df = spark.sql(query)
        df.collect()  # Force materialization
        elapsed = time.time() - start
        times.append(elapsed)

    return {
        "avg_seconds": sum(times) / len(times),
        "min_seconds": min(times),
        "max_seconds": max(times),
        "iterations": iterations
    }

# Test query
test_query = """
    SELECT floor_zone, SUM(coin_in) as revenue
    FROM gold.slot_performance_daily
    WHERE gaming_date BETWEEN '2026-04-01' AND '2026-04-27'
    GROUP BY floor_zone
"""

# Measure BEFORE optimize
before = measure_query_performance(test_query)
print(f"Before OPTIMIZE: avg={before['avg_seconds']:.2f}s")

# Run OPTIMIZE
spark.sql("OPTIMIZE gold.slot_performance_daily WHERE gaming_date >= '2026-04-01'")

# Measure AFTER optimize
after = measure_query_performance(test_query)
print(f"After OPTIMIZE: avg={after['avg_seconds']:.2f}s")
print(f"Improvement: {((before['avg_seconds'] - after['avg_seconds']) / before['avg_seconds'] * 100):.0f}%")

# Check if data skipping is effective
spark.sql("""
    EXPLAIN FORMATTED
    SELECT * FROM gold.slot_performance_daily
    WHERE gaming_date = '2026-04-27' AND floor_zone = 'HIGH_LIMIT'
""").show(truncate=False)

# Look for in the output:
# - "number of files read" (lower = better skipping)
# - "number of files pruned" (higher = better skipping)
# - "predicate pushdown" (should show your WHERE clause)

# ❌ WRONG: OPTIMIZE after every micro-batch
for batch in batches:
    batch.write.format("delta").mode("append").saveAsTable("bronze.events")
    spark.sql("OPTIMIZE bronze.events")  # Wastes massive compute

# ✅ CORRECT: Batch OPTIMIZE on schedule
for batch in batches:
    batch.write.format("delta").mode("append").saveAsTable("bronze.events")
# Later (scheduled job):
spark.sql("OPTIMIZE bronze.events WHERE event_date >= current_date() - INTERVAL 1 DAY")

# ❌ WRONG: ZORDER on UUID or timestamp (too many unique values)
spark.sql("OPTIMIZE silver.events ZORDER BY (event_id)")  # event_id is UUID — no clustering benefit

# ✅ CORRECT: ZORDER on query filter columns with reasonable cardinality
spark.sql("OPTIMIZE silver.events ZORDER BY (event_type, event_date)")

# ❌ WRONG: Turning off V-Order for the entire workspace
spark.conf.set("spark.sql.parquet.vorder.enabled", "false")  # All tables lose optimization

# ✅ CORRECT: Disable only for specific write-heavy operations
spark.conf.set("spark.sql.parquet.vorder.enabled", "false")
# ... streaming writes ...
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")  # Re-enable immediately