Why architecture matters here
Analytical latency budgets are architectural forcing functions. A dashboard tolerates one or two seconds; that budget forbids JVM job startup, forbids writing intermediate results to disk between stages, and forbids re-reading data that a smarter plan would never touch. Impala's architecture is the systematic elimination of those costs: daemons are always running (no startup), operators stream through memory (no materialization barriers), and planning invests heavily in not reading data — partition pruning, Parquet min/max and dictionary skipping, and runtime bloom filters that shrink a fact-table scan using keys actually present in the dimension side.
The same choices define the trade-offs you operate around. Pipelining means a query's whole tree runs at once, so its memory footprint is the sum of its concurrent operators; a hash join that misestimates cardinality by 100× doesn't slow down gracefully — it hits its memory limit and either spills (slower) or fails (if spilling is disabled or exhausted). That is why statistics freshness is not hygiene but survival, and why admission control exists: without a gate, concurrent queries compose their peak memory demands and the unlucky one dies at 3pm on quarter-end. Failure semantics follow too: lose an executor mid-query and (transparent query retry aside) the query fails and re-runs, because streams can't be replayed from a checkpoint that was never written. Impala chose fast-and-restartable over slow-and-resumable — correct for interactive work, wrong for eight-hour ETL, which is exactly the boundary a healthy platform draws between Impala and Spark/Hive.
Understanding the machine also changes how analysts write SQL: joins ordered so filters flow to the biggest scans, explicit stats on join keys, and partition schemes matched to dashboard predicates routinely turn 30-second queries into sub-second ones with zero hardware.
The architecture: every piece explained
Top row: the control plane. Any impalad can act as coordinator for a query (dedicated coordinators are best practice at scale): it parses SQL, resolves metadata from its local catalog cache, and runs the planner — first a single-node plan, then a distributed plan cut into fragments at exchange boundaries, with join strategies chosen by statistics (broadcast the small side vs partition both sides). The catalog daemon loads schema, partition, and file-level metadata from the Hive Metastore once and distributes it; DDL and refreshes flow through it. The statestore is deliberately dumb: a pub-sub heartbeat service that tells every daemon who is alive and ships catalog deltas — if it dies, queries keep running on last-known membership; it is soft-state, rebuilt on restart. Admission control is the gate: resource pools with per-pool memory and concurrency budgets; each query's planned per-node memory estimate is checked against pool headroom and either admitted, queued (with timeout), or rejected with a reason.
Middle row: the data plane. Executor daemons receive fragment instances — scan, join build, join probe, aggregate — scheduled for locality with the data (HDFS blocks, Kudu tablets). Exchange operators connect fragments across the network: streaming, batched row transfers (hash-partitioned, broadcast, or merged) with no disk in the path. LLVM codegen compiles the hot inner loops per query — expression evaluation, tuple comparison, hash computation — into machine code specialized to the concrete types, eliminating virtual dispatch; for very short queries where compile time would dominate, codegen is skipped adaptively. Runtime filters are produced by join builds (bloom filter of observed keys, min/max bounds) and pushed to probe-side scans — a scan can skip entire Parquet row groups or Kudu partitions for keys that cannot match, often the single biggest win on star-schema queries.
Bottom rows: storage is columnar files on HDFS/S3 (Parquet dominant, Iceberg tables increasingly the management layer) or Kudu for mutable, low-latency rows; short-circuit local reads and the remote data cache (for object storage) keep scans fed. Spill to disk gives hash joins, aggregations, and sorts an escape valve under memory pressure — buffered blocks encrypted and written to scratch dirs — trading speed for survival. The ops strip lists what actually keeps clusters healthy: COMPUTE STATS freshness, pool sizing to real concurrency, reading query profiles, and metadata hygiene (partition counts, file sizes, catalog memory).
End-to-end flow
Walk a star-schema query: revenue by region for last week, sales fact joined to store dimension, submitted by a BI tool to a dedicated coordinator. Parse and analyze take milliseconds against the local catalog cache. The planner prunes 358 of 365 partitions by the date predicate, estimates the dimension at 40k rows post-filter — small enough to broadcast — and emits a three-fragment plan: F2 scans the dimension and builds hash tables on every executor; F1 scans the pruned fact partitions, probes, pre-aggregates, and exchanges partial aggregates by region; F0 on the coordinator merges and returns.
Admission control checks the plan's estimate — 900MB per node against the BI pool's headroom — and admits instantly. Fragment instances fan out to 20 executors. F2 finishes its build in 80ms and — the elegant part — each build publishes a bloom filter of the store keys that actually survived the dimension's WHERE clause. Those runtime filters arrive at F1's scan nodes within the configured wait window; scans consult Parquet row-group min/max stats and the bloom filter, skipping two-thirds of the remaining row groups. Codegen'd probe loops chew the surviving batches; partial aggregates stream to the merging fragment as they are produced — no stage barrier anywhere. First rows reach the dashboard at ~400ms; completion at 700ms.
Now the stress case: an analyst joins two fact tables with stats computed eight months ago. The planner estimates the build side at 2M rows; reality is 400M. The hash join blows through its memory reservation, spills partitions to scratch disk, and the query completes in 90 seconds instead of failing — visible in the profile as spilled bytes and a red flag on estimate vs actual rows. Had the pool been configured strictly (no spill, hard mem_limit), the query would have been killed with a clear memory-limit-exceeded error naming the operator. Meanwhile a flood of 60 dashboard queries hits the same pool: admission control admits 24 (the concurrency budget), queues the rest briefly, and the queue drains in seconds — the cluster degrades by queuing, not by OOM-killing random victims. An executor node dies mid-flight: affected queries fail fast; transparent retry re-runs them on the survivors; the statestore broadcasts the new membership and the next plans simply exclude the corpse.