Distributed SQL and Query Scaling Questions

Principles and practices for running and optimizing SQL queries in distributed query engines and cloud data warehouses. Candidates should understand how distributed execution affects query performance including partitioning strategies, shuffle operations, data skew, partition pruning, and cost based optimization in engines such as Spark SQL, Presto, and BigQuery. This topic includes designing queries to minimize data movement, choosing appropriate partition keys, leveraging cluster resources efficiently, and interpreting execution plans and job stages to diagnose bottlenecks in large scale queries.

MediumTechnical

80 practiced

A Trino (Presto) query shows 'Network' stages dominating runtime and EXPLAIN ANALYZE reports large bytes transferred between workers. Describe step-by-step how you'd reduce network shuffle and optimize the plan, including SQL rewrites, use of broadcast, pre-aggregation, and relevant session/engine settings to inspect or change.

Sample Answer

Situation: EXPLAIN ANALYZE shows “Network” stages dominating and very large bytes transferred between workers — classic shuffle/large-exchange problem.

Step-by-step approach to reduce network shuffle and optimize the plan

1. Inspect the plan and metrics- Run: EXPLAIN (ANALYZE, VERBOSE) your_query; note stages with largest "Input" and "Output" bytes and which operators cause exchanges (e.g., repartitioning joins, group-by).- Check row counts, avg row size, and per-stage spill/compression stats to understand where most data moves.

2. SQL rewrites to reduce data volume- Push predicates early: move WHERE filters into subqueries/CTEs so pruning happens before joins.- Project only needed columns: avoid SELECT * so fewer bytes go across network.- Filter dimension tables before joins (apply date, status, etc.).Example:WITH dim_filtered AS ( SELECT id, small_col FROM dim WHERE active = TRUE)SELECT f.key, d.small_colFROM ( SELECT key, SUM(amount) AS amt FROM fact WHERE dt >= DATE '2025-01-01' GROUP BY key) fJOIN dim_filtered d ON f.key = d.id;

3. Pre-aggregation (reduce rows before shuffle)- Aggregate fact data prior to joining to shrink cardinality.- If many facts per key, GROUP BY in an inner subquery or CREATE TABLE AS pre_agg.Example:CREATE TABLE tmp.fact_preagg ASSELECT key, SUM(amount) AS total_amountFROM factWHERE dt BETWEEN '2025-01-01' AND '2025-03-31'GROUP BY key;

Then join tmp.fact_preagg to dimensions (much smaller shuffle).

4. Use broadcast join when appropriate- If one side (dimension) is small (fits in worker memory), broadcast it to avoid expensive repartitioning.- Session change:SET SESSION join_distribution_type = 'BROADCAST';Then run the join query.- Or force broadcast for a single query:SELECT /*+ BROADCAST(d) */ ... — if your Trino distribution supports hints (check version).- Beware: broadcasting very large tables causes OOM; estimate sizes first (SHOW STATS or use approximate counts + avg row size).

5. Reduce row size / serialization cost- Cast wide types to smaller types (bigint -> integer) if safe.- Avoid large repeated JSON blobs; extract needed fields before join.

6. Co-locate / bucketed joins (if using Hive/connector that supports bucketing)- If two large tables are frequently joined on the same key, consider creating bucketed tables with the same number of buckets to enable local bucketed joins and reduce shuffle (check connector docs).- Example: create table with bucketed_by (key) and bucket_count = N.

7. Materialize intermediate results- Use CREATE TABLE AS or temporary tables to persist expensive precomputed intermediate results. This avoids repeating heavy shuffles across repeated dashboard queries.

8. Engine/session settings to tune & inspect- Inspect: - SET SESSION exchange_compression_enabled = true; (reduces network bytes if CPU vs network trade-off acceptable) - SET SESSION task_concurrency = <n>; (controls parallel tasks per worker) - SET SESSION hash_partition_count = <higher/lower>; (adjust partitioning granularity to balance per-task size) - Query memory settings (query_max_memory_per_node) if spills/failed tasks occur.- Force plan type (as noted): SET SESSION join_distribution_type = 'BROADCAST' or 'PARTITIONED'.- Monitor cluster UI / coordinator: per-stage bytes, peak memory, spill to disk.

9. Validate and iterate- After each change re-run EXPLAIN ANALYZE and compare bytes transferred and wall time.- If broadcasting reduced network but increased memory pressure, revert and try pre-agg or bucketed approach.

Example sequence for a BI dashboard:1) Estimate sizes: SELECT count(*) and avg_row_size from tables.2) If dim << fact: SET SESSION join_distribution_type='BROADCAST'; run query.3) If dims not small enough: create fact_preagg (GROUP BY key) and join to dims.4) Re-run EXPLAIN ANALYZE; check bytes and response time improvements.

Reasoning: reduce rows and row size before joins, avoid global repartitioning by broadcasting small tables, and where neither helps, co-locate/bucket or materialize precomputed aggregates. Use session properties to control planner behavior and compression to trade CPU for network. Iterate with EXPLAIN ANALYZE to confirm improvements.

MediumSystem Design

64 practiced

You're responsible for a shared analytics cluster used by interactive dashboards and nightly ETL. Describe how you would size and configure the cluster, allocate resources, and implement queuing or isolation to meet low-latency dashboard SLOs while keeping ETL reliable and cost-effective.

Sample Answer

Requirements & constraints- Dashboards: low-latency interactive queries (SLO e.g., 95th percentile < 300ms).- Nightly ETL: high-throughput, reliable, can be long-running but should not break dashboards.- Cost-sensitive; single shared cluster must serve both.

High-level approach- Classify workloads into “interactive” and “batch”.- Enforce strong isolation (soft and hard) with capacity planning, resource pools, and scheduling/queuing.- Use caching/materialized outputs to reduce load from interactive queries.

Cluster sizing & config (practical steps)- Baseline by measuring typical peak interactive QPS and ETL throughput. For example if average interactive concurrent queries = 50 with median CPU 0.2 vCPU/query and memory 1 GB/query, reserve ~10 vCPUs & 50 GB for interactive pool plus overhead.- Start with two node types (or node pools): small low-latency nodes for interactive (fast CPU, SSD, low contention) and larger cheaper nodes for batch. On cloud: two instance groups.- Configure cluster-wide settings: I/O-optimized disks, query parallelism limits, network tuning.

Resource allocation & isolation- Resource pools / queues: create an interactive pool with guaranteed resources (X% of cluster), small concurrency limit, short timeout; a batch pool with remaining capacity and higher concurrency.- Use query-level resource limits: per-query CPU/memory caps; per-user or per-service quotas.- Priority scheduling: interactive queries get high priority preemptible against batch; batch can be suspended or limited during interactive peaks.- Admission control: allow only N interactive queries concurrently; extra interactive queries queue with tight SLA.

Queuing and scheduling- Short FIFO + priority queue for interactive; long-running batch goes to a separate queue.- Implement fair share or weighted queues so critical ETLs (e.g., data ingestion) get scheduled during off-peak or have guaranteed slots.- For critical nightly ETLs, schedule time windows (e.g., 1–5am) and use backfill retries; mark non-critical jobs as “best-effort.”

Performance optimization to meet SLOs- Cache hot aggregates & use materialized views refreshed incrementally to serve dashboards.- Precompute slow joins/aggregations during ETL to reduce interactive work.- Use result-set caching at BI tool level and DB result cache TTL tuned to business needs.

Autoscaling and cost control- Autoscale batch node pool horizontally based on queue depth and CPU usage; keep interactive pool right-sized (scale min > guaranteed capacity to avoid SLO misses).- Use spot/preemptible instances for batch nodes to reduce cost; ensure checkpointing/retry for preemptions.- Nightly scale-up (autoscale or scheduled) to allow big ETL windows and scale-down during daytime.

Observability & SLO enforcement- Instrument: query latency, queue lengths, CPU/memory per pool, cache hit rate.- Alert on SLO violation and queue backlogs.- Implement automated throttling when interactive latency degrades: reduce batch concurrency or pause low-priority jobs.

Operational policies & governance- Classification policy for jobs (interactive/batch/critical).- Developer guidelines: use materialized views, avoid heavy ad-hoc scans in dashboards; provide sandbox clusters for exploration.- Regular capacity reviews based on growth.

Trade-offs- Stronger isolation increases cost but ensures SLOs.- Heavy precomputation increases storage and ETL complexity but reduces interactive latency.

This approach preserves low-latency for dashboards via guarantees + caching, keeps ETL reliable via scheduled windows and scalable batch capacity, and controls cost using autoscaling and spot instances.

EasyTechnical

76 practiced

Given the table schema below (BigQuery standard SQL), write a query that computes daily active users for 2024-04-01 and explain why the query will use partition pruning. Also list two pitfalls that would prevent pruning.

CREATE TABLE dataset.events (
  event_id STRING,
  user_id INT64,
  event_time TIMESTAMP,
  event_date DATE,
  event_type STRING
) PARTITION BY event_date;

Write the SQL and rationale.

Sample Answer

Query (compute DAU for 2024-04-01):

SELECT
  DATE '2024-04-01' AS day,
  COUNT(DISTINCT user_id) AS dau
FROM
  dataset.events
WHERE
  event_date = DATE '2024-04-01'
  -- OR: event_date = @target_date  -- parameterized query also prunes
  ;

Why this uses partition pruning:- The table is PARTITION BY event_date. The WHERE clause directly filters the partitioning column with a constant date literal (or a bound parameter). BigQuery can therefore scan only the single partition for 2024-04-01 instead of the whole table, greatly reducing IO and cost.- Partition pruning happens when the partition filter is a deterministic, directly comparable expression on the partition column (e.g., =, IN (list), BETWEEN with constants).

Two common pitfalls that prevent pruning:1. Applying a function to the partition column: e.g., WHERE DATE(event_time) = '2024-04-01' or WHERE CAST(event_date AS STRING) = '2024-04-01' — transforming the partition column can stop BigQuery from recognizing a direct partition filter.2. Filtering on a different column (or via a subquery) instead of event_date: e.g., WHERE event_time BETWEEN '2024-04-01' AND '2024-04-02' (uses timestamp, may not prune) or WHERE (some_user_defined_expr) = (SELECT event_date FROM ...). Also OR conditions that mix partitioned and non-partitioned predicates can disable pruning.

Best practice: always include a direct, deterministic filter on the partitioning column (event_date) with literals or parameters.

EasyTechnical

67 practiced

Describe the difference between materialized views, result caching (BI tool cache), and report snapshots. For a high-traffic executive dashboard that shows hourly aggregates across many filters, recommend an approach balancing freshness, cost, and latency; include pros/cons for each option.

Sample Answer

Materialized views, result caching, and report snapshots all speed queries but differ in where/when results are stored and how fresh they are.

Definitions- Materialized view: a persisted, query-result table maintained in the database (either refreshed on-demand, on schedule, or incrementally). Querying reads precomputed rows.- Result caching (BI tool or DB cache): short-lived in-memory or disk cache of query results keyed by SQL/filters; served by the BI layer or DB to avoid recompute.- Report snapshot: pre-generated report/dashboard export (CSV/image or stored dataset) produced on a schedule and presented as fixed historical view.

Recommendation for an hourly-aggregate, high-traffic executive dashboard- Use a hybrid: materialized views that pre-aggregate data into hourly buckets (partitioned by time and key dimensions), refreshed incrementally or every 10–15 minutes for near-real-time freshness; combine with BI tool result caching for identical queries and dashboard tiles (cache TTL ~5–15 minutes). For very expensive, low-change tiles, produce a nightly snapshot for auditing/historical extracts.

Pros / Cons- Materialized views - Pros: low query latency, centralized logic, can be indexed/partitioned, supports many filters if pre-aggregated appropriately. - Cons: storage cost, refresh complexity; if too many dimension combinations, explosion of MV variants.- Result caching (BI cache) - Pros: minimal cost, immediate latency reduction for repeated views, easy to configure per tile. - Cons: cache misses still expensive; cache invalidation tricky with frequent updates; per-user filters may limit hit rate.- Report snapshots - Pros: simplest, predictable cost and performance, good for archived or compliance views. - Cons: stale by definition; not suitable for near-real-time needs.

Practical considerations- Design aggregates to match common filter paths (top nav filters) to maximize MV and cache hit rates.- Use query logging to identify hot tiles and tune cache TTLs and MV refresh frequencies.- Monitor cost: prefer incremental refreshes and partition pruning; avoid precomputing every combination—use on-the-fly filtering against aggregated base tables.- SLA guidance: for executives, target <=15 min freshness with sub-second tile loads; the hybrid approach balances freshness, cost, and latency.

MediumTechnical

77 practiced

For a product analytics workload that needs sub-second aggregations for many ad-hoc filters, explain when to use a star-schema (fact + dimensions) versus denormalized wide tables. Discuss consequences for join complexity, query latency, storage costs, ETL complexity, and how materialized aggregates fit into either approach.

Sample Answer

Situation: You need sub-second aggregations across many ad‑hoc filters for product/ads analytics dashboards. Choose modeling by trade-offs.

Recommendation summary:- Use star-schema (fact table + dimension tables) when you need flexible, normalized attributes, smaller storage growth for dimension attributes, and easier governance. Use denormalized wide tables when query latency and simplicity are paramount and schema changes are infrequent.

Join complexity:- Star-schema: Queries require joins between large fact and multiple dimension tables; BI tools can push joins to the warehouse, but many joins increase planning and can complicate filters (need correct surrogate keys, late-binding dims).- Denormalized: Minimal or no joins — filters apply directly. Simpler SQL for analysts.

Query latency:- Star-schema: Joins add CPU and IO; without indexes/materialized views, ad‑hoc sub-second performance is hard at scale.- Denormalized: Much faster for point/aggregate queries because data locality reduces lookups — better for sub-second SLAs.

Storage costs:- Star-schema: Dimensions stored once → lower storage when attributes repeat. Fact is narrower.- Denormalized: Wider rows duplicate attributes → more storage and larger scan sizes, increasing compute cost for queries.

ETL complexity:- Star-schema: ETL must maintain surrogate keys, slowly changing dimensions, and join keys — more pipeline logic but clearer lineage.- Denormalized: Simpler ETL to produce flattened rows, but updates/attribute changes require full rehydration or complex diffing. Schema evolution (adding attributes) can be heavier.

Materialized aggregates:- Fit both models. With star-schema, materialized aggregates (precomputed rollups) reduce join cost and speed filters — build aggregates keyed by common dims. In denormalized setups, aggregates over flattened tables are faster to compute and serve. Use: - Refresh cadence aligned with business needs (near‑real time vs batch). - Multi-level aggregates (hour/day, by top dims) and bloom filters/bitmap indexes for selective filters. - Columnar storage and clustering/sort keys to accelerate filtered scans.

Practical guidance for a BI analyst:- If dashboards require many dynamic ad‑hoc filters and sub-second response, favor denormalized tables or wide materialized views for the hot paths (served in a reporting mart). Keep a normalized raw star-schema upstream for governance, and build ETL that produces denormalized/materialized slices for fast queries. Monitor storage vs compute cost and maintain automated refreshes for aggregates.

Unlock Full Question Bank

Get access to hundreds of Distributed SQL and Query Scaling interview questions and detailed answers.

Join thousands of developers preparing for their dream job.