Hands-on practice: Tuning High-Latency Queries

Optimization typically begins when a specific Service Level Agreement (SLA) is breached. A practical exercise is presented to dissect a query exhibiting high latency on a distributed SQL engine. The objective is to move from a purely logical understanding of the query to a physical analysis of its execution, systematically reducing the variables $V_{scan}$ (scanned volume) and $V_{shuffle}$ (network movement).

We will analyze a common scenario: an aggregation query running against a 50 TB fact table named fact_web_events and a 20 GB dimension table named dim_users. The current execution time is approximately 18 minutes. Our target is to reduce this to under 60 seconds.

Analyzing the Baseline Performance

The problematic SQL statement attempts to calculate weekly active users by region.

SELECT 
    u.region,
    COUNT(DISTINCT e.user_id) as unique_users
FROM fact_web_events e
JOIN dim_users u 
    ON e.user_id = u.user_id
WHERE e.event_timestamp >= DATEADD('day', -7, CURRENT_DATE())
GROUP BY 1;

Upon executing this query, the query profile indicates two distinct performance bottlenecks. First, the TableScan operator on fact_web_events consumes 70% of the total time. Second, the Join operator reports significant "Bytes Spilled to Local Storage," indicating that memory buffers were insufficient for the join operation.

The following diagram illustrates the execution DAG (Directed Acyclic Graph) with the identified resource contention points highlighted in red.

Execution plan highlighting resource contention. The red nodes indicate where the physical execution diverges from the optimal path due to ineffective pruning and memory spillage.

Optimizing Storage Layout for Pruning

The profile reveals that the TableScan read 50,000 out of 50,000 micro-partitions. Despite the WHERE clause filtering for the last 7 days, the engine scanned the entire history. This occurs because data was ingested based on arrival time or randomly, meaning records for "last week" are scattered across every micro-partition.

To address this, we must align the physical storage with the query pattern. We apply a clustering key (or sort key depending on the platform) on event_timestamp.

-- Syntax varies by platform (Snowflake, Redshift, BigQuery)
-- Applying clustering to align physical storage
ALTER TABLE fact_web_events 
CLUSTER BY (event_timestamp);

After re-clustering the table, the metadata service records the min/max values of event_timestamp for each micro-partition. When the query runs again, the engine compares the predicate to these metadata headers. It can now ignore partitions containing data older than 7 days.

If the dataset covers 5 years, and we query 1 week, we theoretically reduce $V_{scan}$ by a factor of roughly 260 ($5 \times 52$). In practice, due to overlap, we often see a 95% to 98% reduction in scanned bytes.

Correcting Join Strategies

The second bottleneck is the join. The profile shows that the engine performed a Shuffle Join. Because fact_web_events is massive, the engine attempted to redistribute both tables across the network based on the hash of user_id. This resulted in massive network congestion and disk spillage when the data exceeded node memory.

Given that dim_users is 20 GB, it is large but potentially small enough to fit into the collective memory of a medium-sized cluster. However, a Broadcast Join is generally preferred here if the dimension table is significantly smaller than the fact table. In a broadcast join, the smaller table is replicated to all nodes, while the large fact table remains local.

If the optimizer statistics are outdated, the engine might overestimate the size of dim_users and default to a shuffle. We force a statistics update or explicitly hint the join strategy.

-- Analyzing tables to update optimizer statistics
ANALYZE TABLE dim_users;
ANALYZE TABLE fact_web_events;

-- Validating the join strategy shift
EXPLAIN
SELECT /*+ BROADCAST(u) */ 
    u.region,
    COUNT(DISTINCT e.user_id)
FROM fact_web_events e
JOIN dim_users u 
    ON e.user_id = u.user_id
WHERE e.event_timestamp >= DATEADD('day', -7, CURRENT_DATE())
GROUP BY 1;

By broadcasting dim_users, we eliminate the network cost of moving the filtered fact_web_events data. The join now occurs locally on each node where the fact data resides.

Result Verification

After applying clustering to enable partition pruning and optimizing the join strategy, we re-execute the query. The improvements are measurable across three primary metrics: execution time, data scanned, and bytes spilled to remote disk.

The reduction in scanned data directly correlates to I/O savings. The elimination of spillage indicates that the operation now fits entirely within memory, drastically reducing latency.

Comparison of query performance metrics. Note the logarithmic scale on the Y-axis; the reduction in scanned data and spilled bytes drives the exponential decrease in execution time.

The tuning process demonstrated here follows a predictable pattern for MPP systems. First, we minimize I/O by ensuring the storage layout supports the query's filter predicates ($V_{scan}$). Second, we minimize data movement by selecting the appropriate join strategy ($V_{shuffle}$). By methodically addressing these components, we transform a resource-intensive query into a highly efficient operation.