Presto: SQL on Everything — Papers Distilled

In one breath — the whole paper, compressed

Presto is an open-source distributed SQL query engine built at Facebook to replace the collection of incompatible SQL-like systems that organizations had to deploy for different classes of analytics. A single coordinator parses, plans and optimizes each query, then distributes stages as tasks onto workers that share one long-lived JVM and process data through a pipelined, in-memory driver loop over columnar pages. Everything below SQL is behind a Connector API with four parts - Metadata, Data Location, Data Source and Data Sink - so the same engine can read the Hive warehouse, sharded MySQL, Raptor flash storage, key-value stores and Kafka, even inside one query. Rather than exposing knobs, Presto is deliberately adaptive: lazy split assignment, HTTP long-polling shuffles with end-to-end backpressure, a multi-level feedback queue for CPU, overcommitted memory pools, dynamic writer concurrency and speculative processing of dictionary-encoded blocks. The paper reports that this one engine covers everything from 50ms advertiser dashboards to five-hour ETL jobs while processing hundreds of petabytes and quadrillions of rows per day.

Before this paper — the world it landed in

By 2013 Facebook's analytics ran on Hive, which compiled SQL-like queries into MapReduce or Tez jobs and persisted intermediate results to the filesystem between stages. That design bought fault tolerance but made minutes the floor for even trivial queries, so interactive exploration, BI tools and user-facing dashboards were effectively impossible on the warehouse. The alternative was an MPP warehouse such as Vertica, Teradata, Redshift or Exadata, but those reach peak performance only on data ingested into their own internal store, which means copying petabytes and maintaining a second source of truth. Newer engines like Spark SQL and Impala narrowed the gap but still lacked end-to-end pipelining or stayed inside the Hadoop ecosystem. The practical result was that a single organization ran several mutually incompatible SQL-like systems, one per latency class, and users had to know which one to aim at.

The problem — what was actually breaking

Ease of use collapses when an organization is forced to deploy multiple incompatible SQL-like systems just to cover different classes of analytics problems.
A single engine has to span latency requirements that differ by five orders of magnitude, from 50ms advertiser dashboards to five-hour ETL jobs over 100+TB of input.
Systems that persist shuffle data to a filesystem between stages, such as Hive and Spark SQL, gain fault tolerance but add latency that makes them a poor fit for interactive use.
Traditional warehouse products can read external data only to a limited degree and are fastest on data first loaded into their own internal store, forcing expensive ingestion.
Running hundreds of concurrent queries inside one long-lived shared JVM per worker demands integrated CPU scheduling, memory isolation and admission control that a per-query process model gets for free.
Setting per-node and global memory limits conservatively enough to survive skew leaves far too few queries able to run concurrently on a 500-node cluster, and the Hive connector's millions of splits per query would exhaust coordinator memory before execution even started.

Core ideas — the contributions, and why they work

One engine, four workloads

Presto is presented through four production use cases at Facebook - Interactive Analytics, Batch ETL, A/B Testing, and Developer/Advertiser Analytics - that differ in nearly every dimension. Interactive Analytics runs 50-100 concurrent exploratory queries over roughly 50GB-3TB of compressed data with users watching a wall clock; Batch ETL trades latency for resource efficiency and cluster throughput; A/B Testing needs complete, accurate results computed on the fly at 5-30s so users can slice arbitrarily; Developer/Advertiser Analytics needs 50ms-5s responses at 99.999% availability over highly selective queries. The claim is not that one configuration serves all four, but that one codebase, one SQL dialect and one Connector API can be configured into all four. That is what makes the difference between a query engine and a product family.

The Connector API as a federation contract

Rather than treating external systems as a fallback path, Presto puts every data source behind a plugin interface split into a Metadata API, a Data Location API, a Data Source API and a Data Sink API. The split matters because it separates what the optimizer needs at plan time (schemas, statistics, physical layouts, partitioning, sorting, indices) from what workers need at run time (splits to read, pages to return, sinks to write). Because connectors report layouts rather than just rows, the optimizer can push predicates into sharded MySQL, choose an index nested-loop join against a production key-value store, or eliminate a shuffle entirely when both join inputs are already partitioned on the join key. Over a dozen connectors were contributed to the main repository, and a generic Thrift RPC connector reduces adding SQL over a proprietary service to implementing about half a dozen endpoints.

Adaptiveness over configurability

The paper states this as an explicit engineering philosophy: a multi-tenant engine running arbitrary user-defined computation must adapt not just to different query characteristics but to combinations of them, so tuning per workload does not scale. Concretely, splits are assigned lazily to whichever task has the shortest queue, so slow splits and slow workers self-correct; output buffer utilization throttles effective concurrency; input buffer monitoring sets HTTP request concurrency; writer concurrency grows when the upstream stage backs up; and the page processor speculates on whether to evaluate a whole dictionary or just the referenced indices. Before end-to-end backpressure existed, a few jobs with slow clients could hold tens of gigabytes of buffer memory and starve latency-sensitive queries. Adaptivity is what lets one cluster host query shapes nobody enumerated in advance.

Pipelined in-memory shuffle over HTTP

Presto exchanges intermediate results through in-memory buffers served over HTTP long-polling: a producing task holds output in a buffer, consumers poll, and the token in each response implicitly acknowledges the previous segment so no separate ack path is needed. Nothing is written to disk between stages, and data streams from stage to stage as soon as it is available, which is why some query shapes return results before all input is read. This is the single decision that separates Presto from Hive and Spark SQL, and it is a deliberate trade: giving up the natural retry point that a materialized shuffle provides in exchange for the latency that interactive and user-facing use cases require. Long-polling in particular keeps response time low when the transferred payload is small, which is the common case for dashboard queries.

Fine-grained integrated resource management

Because many queries share one JVM per worker, Presto ships its own scheduler rather than relying on the OS or a cluster manager. CPU uses cooperative multi-tasking with a maximum quanta of one second per split, and tasks are classified into the five levels of a multi-level feedback queue by accumulated CPU time, so cheap queries stay near the top and finish fast while expensive ones drain lower levels. Memory is split into user memory, which a user can reason about from the query and data, and system memory, which is a byproduct of implementation such as shuffle buffers, with separate per-node and global limits on each. The engine deliberately overcommits cluster memory on the bet that not every query hits its per-node limit on the same node at the same time, and backstops that bet with spilling and the reserved pool.

Code generation aimed at the JIT

Presto generates JVM bytecode both for expression evaluation and for whole operators, and the interesting argument is why generic interpretation fails on this specific platform. Because the engine switches between splits from unrelated tasks every quanta, a shared tight processing loop would collect profiling information polluted by other queries, so the JIT would never specialize it. Generating a separate Java class per task gives each one its own profile, lets the JIT inline monomorphic call sites, unroll loops over typed columns, auto-vectorize, and re-optimize as the data a task sees changes over its lifetime. The engine also shapes itself around the G1 collector - avoiding allocations above the humongous threshold, using segmented and flat arrays instead of large linked object graphs - because GC behaviour otherwise dictates throughput.

How it works — the mechanism, concretely

Query lifecycle on the coordinator

A client posts SQL over a RESTful HTTP interface; the coordinator evaluates queue policies, then an ANTLR-based parser produces a syntax tree that the analyzer uses to resolve types, coercions, functions and scopes and to extract subqueries, aggregations and window functions. The logical planner emits a tree of purely logical plan nodes carrying no execution information, with each node's children as its inputs. The optimizer then applies transformation rules greedily until a fixed point: each rule has a pattern matching a plan sub-tree and rewrites it into a logically equivalent sub-plan. Presto ships predicate and limit pushdown, column pruning and decorrelation, plus two cost-based optimizations that consume table and column statistics - join strategy selection and join reordering - with fuller Cascades-style search still in progress.

Shuffle-minimizing distributed planning

The optimizer cuts the plan into stages, the units that can run in parallel across workers, and inserts buffered in-memory shuffles between them; since shuffles cost latency, buffer memory and CPU, minimizing their number is the central physical planning objective. Plan nodes advertise output properties - partitioning, sorting, bucketing, grouping - and can declare required and preferred properties, so redundant shuffles are elided and others are retuned. Presto greedily picks a partitioning that satisfies as many required properties as possible, which may mean partitioning on fewer columns and accepting more skew. When connectors report a layout, this collapses real work: in A/B Testing both join inputs are partitioned on the same column so a co-located join removes the shuffle, and the four-shuffle naive plan of Figure 3 collapses to a single data processing stage.

Stage, task and split scheduling

Two policies govern stage order. All-at-once schedules every stage concurrently to minimize wall clock time and serves the latency-sensitive use cases; phased execution finds the strongly connected components of the data flow graph that must start together to avoid deadlock and runs them in topological order, so a hash join does not stream the probe side until the build side is done, which is far more memory efficient for batch work. Task scheduling then splits stages into leaf and intermediate: leaf stages read connectors and, absent constraints and given enough splits, are scheduled on every worker in the cluster because decompression, decoding and filtering dominate CPU and parallelize well; the scheduler can also honour a plugin-supplied network topology to prefer rack-local reads. Intermediate stage tasks can go anywhere, and the engine can change their count during execution.

Lazy split enumeration and assignment

A split is an opaque handle to an addressable chunk of data whose contents are connector-specific - a file path and byte offsets for a distributed filesystem, or table info plus key and value format plus a host list for Redis. Presto asks connectors to enumerate splits in small batches and assigns them to tasks lazily, which buys four things: query response time is decoupled from enumeration time, which for the Hive connector can take minutes to list partitions and files; queries with a LIMIT or an early cancel often finish before enumeration completes; the coordinator assigns each new split to the task with the shortest queue, absorbing variance in per-split CPU cost and worker speed; and coordinator memory never has to hold metadata for millions of splits at once. Every leaf-stage task needs at least one split to become runnable, while intermediate tasks are always runnable and finish only when aborted or when all upstream tasks complete. The stated cost is that accurate progress reporting becomes difficult.

Local execution, buffers and backpressure

Within a task, a driver loop moves pages - a columnar encoding of a row sequence, made of per-column Blocks with flat in-memory representation - between operator pairs that can make progress, rather than using a Volcano-style pull of recursive iterators. The loop is chosen precisely because operators can be brought to a known state and yield rather than block indefinitely, which is what makes cooperative multi-tasking work; the scheduler also switches away early when output buffers are full, input buffers are empty, or memory is exhausted. On the network side, the engine watches output buffer utilization and lowers effective concurrency by reducing eligible splits when buffers stay full, which both shares network fairly and prevents a slow BI client downloading 10-50MB from pinning tens of gigabytes of buffers. The receiver tracks a moving average of bytes per request to compute target HTTP concurrency, and the resulting backpressure propagates upstream. Writes use the same idea in reverse: writer concurrency is increased by adding tasks on more nodes when the producing stage exceeds a buffer utilization threshold, balancing S3 small-file blowup against write throughput.

Memory pools, spilling and the reserved pool

Every non-trivial allocation is classified as user or system memory and reserves from the corresponding pool; queries exceeding the global or per-node limit are killed, and when a node is out of memory, reservations are blocked by halting task processing. The paper works the arithmetic: on a 500-node cluster with 100GB of query memory per node and a 5TB global per-query limit, ten queries fit, but allowing 2:1 skew forces the per-node limit down to 20GB and guarantees only five concurrent queries - unacceptably few, hence deliberate overcommit. Two mechanisms keep an overcommitted cluster healthy. Spilling revokes memory from eligible tasks in ascending order of execution time until the last request is satisfiable, writing hash join and aggregation state to disk. If spilling is not configured or nothing revocable remains, the reserved pool takes over: the largest query on the starved node is promoted to the reserved pool on every worker, and to avoid the deadlock of different workers stalling different queries, only one query cluster-wide may occupy it at a time.

What the paper showed — measurements and proofs

On a 100-node cluster (28-core Xeon E5-2680 v4 at 2.40GHz, 1.6TB flash, 256GB DDR4 per node) running a low-memory subset of TPC-DS at scale factor 30TB, the same queries were run three ways - Raptor, Hive/HDFS without statistics, and Hive/HDFS with table and column statistics - and Figure 6 shows runtime is greatly affected by connector characteristics with no change to the query or cluster configuration.
Lazy data loading measured on a sample of the production Batch ETL workload reduced data fetched by 78%, cells loaded by 22% and total CPU time by 14%.
Figure 7's runtime CDF for the four production use cases spans from roughly 16ms to about 19 hours on a log scale, demonstrating that one engine serves web use cases with 20-100ms requirements and programmatically scheduled ETL jobs running for hours.
A four-hour trace of an Interactive Analytics cluster (Figure 8) shows Presto holding roughly 90% average worker CPU utilization even as concurrency falls from a peak of 44 queries to a low of 8, while still allocating large fractions of cluster CPU to newly admitted queries within milliseconds.
Table I records the deployed operating points: Developer/Advertiser Analytics at 50ms-5s on tens of nodes with hundreds of concurrent queries over sharded MySQL; A/B Testing at 1-25s on hundreds of nodes over Raptor; Interactive Analytics at 10s-30min on hundreds of nodes with 50-100 concurrent queries over Hive/HDFS; Batch ETL at 20min-5hr on up to 1000 nodes over Hive/HDFS.
Operational scale reported for late 2018 (version 0.211): clusters up to about 1000 nodes, hundreds of petabytes and quadrillions of rows processed per day, and a median worker exporting around 10,000 real-time performance counters.

Limits and trade-offs — conceded and discovered

Conceded in the paper: as of late 2018 Presto has no meaningful built-in fault tolerance for coordinator or worker crashes - a coordinator failure makes the cluster unavailable, a worker crash fails every query on that node, and recovery is delegated to client retries, standby coordinators and multiple active clusters.
Conceded in the paper: spilling exists for hash joins and aggregations but no Facebook deployment is configured to use it, because clusters already have terabytes of distributed memory, users value predictable fully in-memory latency, and local disks add hardware cost in shared-storage deployments.
Conceded in the paper: the reserved pool is wasteful, since it must be sized on every node to fit a query running up against the local memory limit, only one query cluster-wide may occupy it, and other tasks on a node whose general pool is exhausted simply stall until that query finishes.
Conceded in the paper: lazy split enumeration makes it difficult to estimate and report query progress accurately, and the greedy shuffle-reduction heuristic may partition on fewer columns and thereby increase partition skew.
Exposed by later work: the cost-based optimizer was incomplete at publication - only join reordering and join strategy selection used statistics, with Cascades-style search still in progress - and both successor projects later added the dynamic filtering and materialized-exchange fault-tolerant execution that this design deliberately omitted.

What it became — the systems that inherited it

Presto became the template for the federated, storage-agnostic SQL engine: Amazon Athena is built on it, Uber, Netflix, Airbnb, Bloomberg and LinkedIn ran it in production, and Qubole, Treasure Data and Starburst built commercial offerings around it. In 2019 the project split, with the original authors' fork becoming PrestoSQL and then Trino, while Facebook's line continued as PrestoDB under the Linux Foundation's Presto Foundation. Trino went on to close the gaps this paper names - a full cost-based optimizer, dynamic filtering that prunes probe-side scans from build-side values, and fault-tolerant execution with exchanges materialized to object storage, which is exactly the optional checkpointing the paper says it was evaluating. Meta pushed the other direction and rewrote the worker's execution layer in C++ as Velox and Prestissimo, keeping the coordinator, connector and split model while replacing the JVM operators the paper spends Section V defending. The Connector API and the catalog model it implies are now the standard way query engines attach to Iceberg, Delta Lake and Hudi tables, so the modern lakehouse pattern of separating an open table format from a pluggable engine is largely Presto's architecture generalized. Its most durable idea may be the smallest: that shuffle materialization is a tunable trade against latency rather than a fixed property of a distributed query engine.

In the paper’s words — verbatim

“In aggregate, Presto processes hundreds of petabytes of data and quadrillions of rows per day at Facebook.”

§I

“Tests on a sample of production workload from the Batch ETL use case show that lazy loading reduces data fetched by 78%, cells loaded by 22% and total CPU time by 14%.”

§V-D

“However, as of late 2018, Presto does not have any meaningful built-in fault tolerance for coordinator or worker node crash failures.”

§IV-G

Vocabulary — as this paper uses it

Connector API: The plugin contract by which Presto reaches an external data store, composed of a Metadata API, Data Location API, Data Source API and Data Sink API. It is designed so that connector implementations can stay performant inside a physically distributed execution engine.
Split: An opaque handle to an addressable chunk of data in an external storage system, whose contents are defined by the connector - a file path and offsets for a filesystem, or a key, value format and host list for Redis. Splits are the unit assigned to leaf-stage tasks and the unit of scheduling on a worker thread.
Stage and task: A stage is a part of the plan that can be executed in parallel across workers; every stage is distributed as one or more tasks, each running the same computation over different input data. Leaf stages read from connectors, intermediate stages consume only results from other stages.
Pipeline and driver loop: A pipeline is a chain of operators inside a task, such as the build pipeline and probe pipeline of a hash join, joined to other pipelines by a local in-memory shuffle. The driver loop executes a split by repeatedly moving data between every pair of operators that can make progress, until the quanta expires or nothing can advance.
Page and Block: A page is the unit of data the driver loop moves between operators: a columnar encoding of a sequence of rows, made of one Block per column with a flat in-memory representation. Blocks may be plain, dictionary-encoded or run-length encoded, and flatness matters because pointer chasing, unboxing and virtual calls dominate tight loops.
Data layout: A physical description of a table that a connector exposes to the optimizer - locations plus partitioning, sorting, grouping and index properties. A connector may return several layouts for one table so the optimizer can pick the one that best serves the query, for example an index on the predicate columns.
User memory versus system memory: User memory is allocation a user can reason about from basic knowledge of the query and input, such as an aggregation proportional to its cardinality; system memory is a byproduct of implementation, such as shuffle buffers, and may be uncorrelated with query shape. Presto sets separate limits on user memory and on total user-plus-system memory.
Reserved pool: A per-node subdivision of query memory used to unblock a cluster that has run out of general pool memory without spilling. The largest query on the starved node is promoted into the reserved pool on all workers, and only one query cluster-wide may occupy it so that different workers cannot stall different queries into deadlock.
Raptor: A shared-nothing storage engine written specifically for Presto that keeps metadata in MySQL and data as ORC files on local flash disks, supporting sorting, bucketing and temporal columns. It backs the A/B Testing use case, where predictable high-throughput low-latency reads matter more than querying data in place.

On the timeline — where this sits in the story

Year2012–2013

MilestonePresto / Trino-style distributed SQL

Problem solvedEnterprise data lived in many systems; moving everything into one warehouse before querying was costly and slow.

How it solved itPresto enabled interactive SQL over multiple data sources and storage systems.

Why it matteredPopularized federated/distributed SQL engines for heterogeneous data platforms.

DDIACh. 7: Sharding Ch. 11: Batch Processing

View on the timeline