Trino Explain Plan Review

Trino EXPLAIN Plan Review

When to Use

• A query is unexpectedly slow and you need to find the bottleneck
• Verifying that predicate/partition pushdown is actually happening
• Detecting data skew across tasks
• Comparing plan cost estimates against actual runtime metrics
• Reviewing query plan before deploying to production

EXPLAIN Syntax Reference

-- Default: DISTRIBUTED plan (most useful for production diagnosis)
EXPLAIN SELECT ...;

-- Distributed plan (explicit)
EXPLAIN (TYPE DISTRIBUTED) SELECT ...;

-- Logical plan (high-level operators)
EXPLAIN (TYPE LOGICAL) SELECT ...;

-- IO plan (what tables/columns are read, with constraints)
EXPLAIN (TYPE IO) SELECT ...;

-- Validate syntax only — returns TRUE/FALSE
EXPLAIN (TYPE VALIDATE) SELECT ...;

-- Output as JSON (useful for programmatic analysis)
EXPLAIN (FORMAT JSON) SELECT ...;

-- Runtime metrics — execute the query and report actual operators
EXPLAIN ANALYZE SELECT ...;

-- Full verbose metrics: percentile distributions, driver counts
EXPLAIN ANALYZE VERBOSE SELECT ...;

DISTRIBUTED Plan: Fragment Types

A distributed plan splits into numbered Fragments separated by exchange operators. Each fragment runs on specific nodes.

Fragment 0 [SINGLE]         ← coordinator collects final output
  Output[order_id, total]
    └── RemoteStreamingExchange[GATHER]
          └── Fragment 1 [HASH]   ← join stage, repartitioned by join key
                HashJoin[order_id]
                  ├── RemoteExchange[REPARTITION] ← probe side from Fragment 2
                  └── LocalExchange[BROADCAST]    ← build side from Fragment 3
Fragment 1 ...
Fragment 2 [SOURCE]         ← scans fact_orders splits
  ScanFilterProject[table=fact_orders, ...]
Fragment 3 [SOURCE]         ← scans dim_customer splits
  ScanFilterProject[table=dim_customer, ...]

Key insight: Every RemoteExchange = network shuffle. Minimize them.

Reading an EXPLAIN DISTRIBUTED Plan

EXPLAIN
SELECT o.customer_id, SUM(o.amount) AS total
FROM iceberg.silver.orders o
JOIN iceberg.gold.dim_customer c ON o.customer_id = c.customer_id
WHERE o.order_date >= DATE '2024-01-01'
GROUP BY o.customer_id;

Look for:

EXPLAIN ANALYZE: Runtime Metrics

EXPLAIN ANALYZE runs the query and annotates the plan with actual metrics.

EXPLAIN ANALYZE VERBOSE
SELECT customer_id, SUM(amount)
FROM iceberg.silver.orders
WHERE order_date = DATE '2024-06-01'
GROUP BY customer_id;

Sample output interpretation:

Fragment 1 [HASH]
  CPU: 45.23s, Scheduled: 62.10s, Blocked: 3.21s (I/O: 1.5s, Overall: 1.7s)
  Output: 1234567 rows (98.5MB), Filtered: 78.3%
  Input avg.: 15.2M rows, Input std.dev.: 234.5% ← HIGH SKEW
  
  - ScanFilterProject[table=silver.orders] {
      Input: 15.2M avg (range: 1K–45M)          ← SKEWED splits
      CPU fraction: 35%
    }

Metrics to Flag

Common Plan Patterns and Fixes

Pattern 1: Missing Partition Pruning

-- BAD PLAN: full scan despite date filter
ScanFilterProject[table=silver.orders]
  Estimates: {rows: 3650000000, cpu: 3.65T, ...}
  Filter: (YEAR(order_date) = 2024)   ← function wrapping prevents pushdown

-- FIX: use range predicate
WHERE order_date >= DATE '2024-01-01' AND order_date < DATE '2025-01-01'

-- GOOD PLAN: only matching partitions read
ScanFilterProject[table=silver.orders]
  Estimates: {rows: 365000000, cpu: 365B, ...}
  Constraint: order_date in [2024-01-01, 2024-12-31]

Pattern 2: Accidental Cross Join

-- BAD: missing join condition creates cross join
Fragment 1 [BROADCAST]
  CrossJoin
    ├── Source: fact (1B rows)
    └── Broadcast: dim (100 rows) → 100B rows output

-- FIX: add ON clause with correct column

Pattern 3: Data Skew in Hash Join

-- SYMPTOM: one task processes 90% of rows
Fragment 2 [HASH]
  Input avg.: 5M rows, Input std.dev.: 890%   ← extreme skew
  HashJoin[customer_id]

-- CAUSE: customer_id distribution is highly skewed (many NULLs or one hot key)
-- FIX 1: filter NULLs before join
-- FIX 2: salt the hot key
SELECT CONCAT(CAST(customer_id AS VARCHAR), '_', CAST(MOD(ABS(order_id), 8) AS VARCHAR))

Pattern 4: Unnecessary Remote Exchange

-- BAD: intermediate aggregation forces repartition
Fragment 1 [HASH]
  RemoteExchange[REPARTITION]   ← extra shuffle
    Aggregate[partial]
      ScanFilterProject

-- GOOD when: pre-aggregation on SOURCE fragment reduces data 100x before shuffle
-- BAD when: cardinality is unchanged (COUNT DISTINCT of unique key)

Pattern 5: Small Files → Too Many Splits

-- SYMPTOM: thousands of splits, each tiny
ScanFilterProject[table=bronze.events]
  Splits: 45000 × 1MB    ← 45K tiny files

-- FIX: run optimize to compact
ALTER TABLE iceberg.bronze.events EXECUTE optimize(file_size_threshold => '128MB');

IO Plan: Verify Table Access

EXPLAIN (TYPE IO, FORMAT JSON)
SELECT order_id FROM iceberg.silver.orders
WHERE order_date = DATE '2024-06-01';

{
  "inputTableColumnInfos": [
    {
      "table": {"catalog": "iceberg", "schemaTable": {"schema": "silver", "table": "orders"}},
      "columns": ["order_id", "order_date"],   -- columns actually read
      "constraint": {
        "columnConstraints": [
          {
            "columnName": "order_date",
            "nullability": "NOT_NULL",
            "domain": {"ranges": [{"low": "2024-06-01", "high": "2024-06-01"}]}
          }
        ]
      },
      "estimate": {"outputRowCount": 1234567.0, "outputSizeInBytes": 9876543.0}
    }
  ]
}

Use IO plan to verify:

• Only necessary columns are read (columns list)
• Partition constraint is applied (domain.ranges)
• Estimated row count is reasonable (not full table)

EXPLAIN ANALYZE Verbose: Skew Detection

EXPLAIN ANALYZE VERBOSE
SELECT customer_id, COUNT(*)
FROM iceberg.silver.orders
GROUP BY customer_id;

In verbose output, look for:

Input rows distribution:
  p25: 1.2M, p50: 3.5M, p75: 8.1M, p99: 45.2M   ← p99 >> p50 = SKEW
  
Active drivers: 48/64 (25% idle) ← some tasks starved

High p99/p50 ratio (> 5x) indicates hot partitions requiring data redistribution.

Quick EXPLAIN Checklist

-- Run EXPLAIN first:
EXPLAIN (TYPE DISTRIBUTED, FORMAT TEXT) <your query>;

-- Check:
-- [ ] Number of fragments ≤ 5 for simple queries
-- [ ] ScanFilterProject has partition constraints (not full scan)
-- [ ] No Filter node above TableScan (predicate pushed down)
-- [ ] JOIN uses BROADCAST for small build side (< 100MB)
-- [ ] No unexpected CrossJoin nodes
-- [ ] Estimated rows are reasonable (not billions for small result)

-- Then run EXPLAIN ANALYZE to get actuals:
EXPLAIN ANALYZE VERBOSE <your query>;

-- Check actuals:
-- [ ] Input std.dev. < 50% (no skew)
-- [ ] Blocked time < 5% of scheduled time
-- [ ] Estimated rows ≈ actual rows (CBO working)
-- [ ] CPU fraction > 30% of scheduled time

Anti-Patterns

1. Only running EXPLAIN without ANALYZE — estimated costs are based on statistics that may be stale; always run EXPLAIN ANALYZE to see actual row counts.
2. Ignoring Input std.dev. — high deviation (> 200%) is the primary skew signal; it means a few tasks do most of the work while others sit idle.
3. Missing the difference between SINGLE and HASH fragments — SINGLE means one node does all the work; a SINGLE fragment handling billions of rows is a bottleneck.
4. Not checking Filter node placement — a Filter above TableScan means the predicate wasn't pushed into the connector; rewrite the filter to enable pushdown.
5. Reading TEXT format for automated analysis — use FORMAT JSON for programmatic processing; TEXT is for human reading only.

References

• EXPLAIN syntax: trino.io/docs/current/sql/explain.html
• EXPLAIN ANALYZE: trino.io/docs/current/sql/explain-analyze.html
• Pushdown verification: trino.io/docs/current/optimizer/pushdown.html
• Related skills: [[trino-query-optimization]], [[trino-memory-and-spill-tuning]], [[trino-iceberg-best-practices]]