Query Engines
Engines that answer SQL over storage they do not own — what that separation buys, and every guarantee it quietly hands back.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
What changes when the system planning your query is not the system that stores your data?
Analysts and BI tools that want one SQL dialect over a lake, a warehouse and a handful of operational databases; transformation jobs that read raw files and write modelled tables; and notebooks where somebody is about to run SELECT * against a year of history. All of them want the interface of a database. None of them get its guarantees.
A query engine works in splits: a bounded, independently readable piece of one source — a row group inside a Parquet file, a byte range of a text file, a page of rows from a remote table. A query is *planned* in operators and *executed* in splits, and almost every performance question in this module reduces to how many splits the planner created and how much each one had to read (Parquet Internals).
Point the engine at the object store, register the tables in a catalog, and treat it as a database that happens to be very large. The SQL is standard, the tables have names, EXPLAIN works, and for a while every query that a warehouse could answer this thing answers too — at a fraction of the setup effort, because nothing had to be loaded first.
The engine has no control over how the bytes are laid out. A table written as forty thousand tiny files per day is read as forty thousand splits, and the query spends its life opening objects rather than reading rows (File Size and the Small-Files Problem).
- The engine has no control over how the bytes are laid out. A table written as forty thousand tiny files per day is read as forty thousand splits, and the query spends its life opening objects rather than reading rows (File Size and the Small-Files Problem).
- There are no indexes to fall back on. A database can rescue a badly written query with a selective index; a scan engine cannot, so a predicate that does not line up with the partitioning or the file statistics means reading everything (Partition Pruning).
- The planner has no statistics unless something computed them. Join order, build side selection and broadcast decisions are all made from estimates, and when the estimate for a remote table is a guess the plan is a guess (Cost-Based Optimization).
- Nothing enforces constraints. There is no primary key, no foreign key, no
NOT NULLthat the engine can rely on — the table is whatever files the catalog currently points at, including the half-written ones if the writer did not publish atomically (Atomic Publish). - Two engines reading the same files disagree about types. A column written as an
INT96timestamp, or a decimal with a scale one reader widens and another truncates, produces two different answers to the same SQL over the same bytes (Nullability & Defaults). - The query that worked in the notebook joins a 200-million-row fact to itself and the coordinator, which was sized for planning, is asked to hold the result of a final aggregation it never anticipated (Distributed Query Execution).
What is actually happening
- The defining property is separation of the engine from the storage it reads. A database owns its files, its page layout, its indexes, its statistics and its transaction log, and can therefore promise things about all of them. A query engine owns a plan, a scheduler and a pool of workers, and borrows everything else (Separating Storage from Compute).
- That means the engine's knowledge of a table comes entirely from two places: a catalog that says where the files are and what columns they claim to have, and whatever metadata the files themselves carry — Parquet footers, row-group statistics, table-format manifests (Open Table Formats).
- Planning is therefore the same shape as in any database — parse, bind names against the catalog, rewrite, cost, choose a physical plan — but with far weaker inputs. The literature on this is the same literature (The Planner: Enumerating Ways to Answer); what differs is how much of it the engine can actually rely on.
- Execution is a tree of operators — scan, filter, project, join, aggregate, exchange — pushed out to workers as tasks over splits. The only operators that can reduce work before it starts are the ones the *reader* can evaluate, which is why the pushdowns in this module matter more here than in a database with indexes.
- The same architecture appears at very different scales. DuckDB runs the whole thing in one process on one machine over the same Parquet files; a distributed engine runs it across a cluster; a warehouse runs it over storage it does own. The plan shapes are recognisably the same, and the difference is who controls the layout (DuckDB Concepts).
What the engine owns and what it borrows
The whole module follows from one picture. In a database, one system owns the parser, the planner, the statistics, the buffer pool, the file layout, the indexes and the write-ahead log; every guarantee it makes is possible because it controls all of them together (What a Database Actually Is). A query engine owns the left half of that list and borrows the right half from whoever wrote the files.
The catalog is the seam. It maps a table name to a location, a format and a schema — and that is the entire basis on which the planner reasons. If the catalog is wrong about the schema, the engine is wrong about the data and cannot know it. If the catalog does not know about a partition, that partition does not exist as far as every query is concerned.
Notice which arrows in the diagram point *into* the engine. The engine reads from the catalog and from storage; nothing gives it the ability to change how storage is organised. That asymmetry is why "the query is slow" so often has an answer that lives outside the engine entirely.
Where a query engine is weaker than a database, and why
It is tempting to read the table below as a list of shortcomings. It is better read as a list of *consequences*: each row is something a database can promise precisely because it owns storage, and something an engine cannot promise precisely because it does not. Nothing here is a bug to be fixed in a later version.
The rows that surprise people are the statistics row and the constraints row. Analysts arrive with database instincts — "the planner will figure it out", "the key is unique" — and both instincts are load-bearing in a database and unfounded here. A join whose cardinality estimate is a guess produces a plan that is a guess, and a GROUP BY on a column you believe to be unique will happily aggregate duplicates (Duplicate Rows).
The final row is the one worth internalising. A database can rescue you with an index; a scan engine can only skip what the layout and the file statistics let it skip. That is why the three pushdown lessons in this module are not micro-optimisations — for this architecture they are the whole optimisation surface.
| Concern | Database | Query engine over storage it does not own | What this means for you |
|---|---|---|---|
| Physical layout | Owned. The storage engine decides page layout, clustering and rewrite timing. | Borrowed. Whatever the writers produced is what the reader gets. | Layout becomes a pipeline design decision, made months before the query exists (Physical Data Layout). |
| Indexes | Available, selective, maintained transactionally. | None. Only coarse skipping via partitions and file/row-group statistics. | A non-selective predicate has no fallback. It reads everything (An Index Scan Is Not Automatically Faster). |
| Statistics | Maintained by the system, refreshed on a policy it controls. | Only what the files or the table format carry; often absent for remote sources. | Join order and broadcast decisions are made from estimates that may be defaults (Cost-Based Optimization). |
| Constraints | Enforced: primary keys, foreign keys, nullability. | Declared at best, enforced never. | Uniqueness and referential integrity become data tests, not guarantees (Data Tests). |
| Transactions | Full ACID across statements and tables. | Snapshot reads per table where a table format provides them; nothing across tables or sources. | A query joining two tables can see two different points in time (Federated Query). |
| Concurrency control | Readers and writers coordinated by the storage engine (MVCC: Multi-Version Concurrency Control). | Coordination is whatever the table format's commit protocol provides. | A writer without atomic publish makes readers see half a load (Atomic Publish). |
| Caching | Buffer pool tuned to its own access patterns (The Buffer Pool). | Optional caches the engine keeps, over storage that may change beneath them. | Cache invalidation depends on table version, not on the engine noticing. |
When this architecture is the right one
The honest comparison is not "engine versus warehouse" as products but as answers to a specific constraint. If the data already sits in an open format in object storage, an engine reading it in place removes an entire copy, an entire load pipeline and an entire class of divergence between the copy and the original. If the data does not, then you are choosing where to put it, which is a different question (Lake vs Warehouse vs Lakehouse).
The option people skip is the single-node one. A dataset that fits in memory on one large machine, queried by a handful of people, is not a distributed systems problem, and running it as one imports a coordinator, an exchange, a scheduler and a cluster to operate — all to solve a problem you do not have (DuckDB Concepts).
And the option people reach for too eagerly is federation. Reaching an operational database directly from an analytical engine is a genuine capability with a genuine cost, and the cost lands on a system whose availability is somebody's on-call rotation (Federated Query, Workload Isolation).
Where does the data live, how big is the working set, and who else needs to read it?
when The working set fits comfortably on one machine; the users are a few analysts, a notebook, or a transformation step inside a pipeline.
cost One machine's ceiling, and no shared concurrency control. Buys the entire distributed layer being absent — no cluster, no coordinator, no scheduler to operate.
when Data is already in open formats in object storage, several teams need to read the same copy, and the working set exceeds one machine.
cost You own the layout problem completely: file size, partitioning, compaction and statistics are yours (File Compaction). Buys openness and one copy.
when The workload is stable, concurrency is high, and you want the system to manage clustering, statistics and maintenance for you.
cost A load step, a copy that can diverge from the source, and query access that is mediated by one vendor's engine (The Data Warehouse).
when The question genuinely spans systems and the alternative — building ingestion for a source you may query twice — is not yet justified.
cost No cross-source consistency, unpredictable latency, and analytical load on operational systems. Best treated as exploratory rather than as a serving path (Federated Query).
when One database holds the data and the queries return acceptably against a replica.
cost None, and this is more often the right answer than the architecture diagrams suggest. Revisit when a second source appears or when scans start affecting production (Workload Isolation).
Trino, Presto, Spark SQL, DuckDB, ClickHouse and the cloud warehouses all move rapidly on which connectors exist, which pushdowns each one supports, and how caching and result reuse behave. The architectural split described here has been stable for a decade; any specific capability claim should be checked against your engine's current documentation and, better, against the plan it prints.
How to build it
Most important first.
- Decide who owns the layout before you choose the engine. An engine over an object store is only as good as the files it is pointed at, so if nobody owns file size, partitioning and sort order, you have bought a query interface onto a problem you did not fix (Physical Data Layout).
- Use a table format rather than a directory of files wherever the data is written more than once. Manifests give the planner file-level statistics, atomic commits and schema evolution — the three things a raw directory listing cannot provide (Open Table Formats).
- Write queries the reader can help with: filter on the partition column, in the partition column's own type, without wrapping it in a function; project the columns you need (Predicate Pushdown, Projection Pushdown).
- Keep operational sources behind a boundary. Federation is a real capability and it is also a way to point an analytical scan at a production database; decide deliberately which tables an engine may reach and what it is allowed to pull (Federated Query, Workload Isolation).
- Give the engine memory limits and query limits that fail fast. An unbounded analytical query on shared infrastructure does not fail alone — it takes the workers with it, and the second-order incident is every other query queued behind it.
- Do not let the engine become the model. A query engine makes it easy to join raw files ad hoc; that convenience is exactly how a platform ends up with fourteen definitions of revenue and no table to point at (The Metrics Layer).
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- The engine guarantees that it will execute the plan it produced against the files the catalog listed at planning time. It does not guarantee those are the same files by the time the last split is read, unless the table format gives it a snapshot.
- Where a table format is used, a query normally reads one committed snapshot of that table — read isolation for one table, from that table's own commit log. That guarantee does not extend across two tables and never extends across two systems (Federated Query).
- There is no completeness guarantee. The engine reports the rows in the files it was told about; a partition that was never written is not an error, it is a quiet zero (Missing Rows).
- Type fidelity is a guarantee of the format and the reader, not of the engine. Two readers of the same file can disagree, and neither will raise (The Parquet Read Path).
- Nothing is promised about repeatability across time. Re-running the same SQL tomorrow reads whatever the table contains tomorrow — which is why reproducible analysis needs a snapshot or a timestamp, not just a saved query.
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- The cheapest check is a file and partition census: for each expected period, does the table have the partitions and the file count you expect, and is any partition suspiciously small? It catches missing writes, half-published loads and a writer that silently stopped.
- It misses everything about the content of the files. A partition with the right number of files and the right row count can be entirely nulls in the column that matters, because a schema change upstream made the reader unable to resolve it (Semantic Changes).
- Pair it with a per-column null-rate check on the columns queries actually filter on. A column that becomes null does not fail a query — it silently removes rows from the result of every predicate over it (Data Tests).
- Querying files in place removes the load step entirely: data is queryable as soon as it is committed to the table, so the freshness of a query equals the freshness of the writer. That is the strongest argument for this architecture and it is genuinely strong.
- It also means the engine inherits the writer's partial states. A job that appends files for twenty minutes before it finishes makes a table that is legitimately, observably incomplete for twenty minutes unless the publish is atomic (Atomic Publish).
- Federating to an operational source gives the freshest possible answer and the least predictable latency: the query is now waiting on a system whose response time is governed by its own transactional workload.
- Schema evolution is resolved at read time, per file. Older files genuinely lack the column added last month, and the engine fills it with null — which is correct behaviour and also indistinguishable from data loss to anybody reading the result (Schema Evolution).
- How a column is matched between the file and the table schema — by name or by position — is the single most consequential detail here. Position-based matching plus a dropped column silently shifts every column after it into the wrong slot.
- Adding a column to the table without rewriting history is nearly free and usually safe. Renaming or retyping one is a rewrite in disguise unless the table format tracks column identity separately from column name (Breaking Schema Changes).
- Recovery here is mostly rewriting files rather than re-running queries: a query produces no durable state, so a failed query is retried and nothing needs repair.
- Where the engine wrote output —
CREATE TABLE AS,INSERT INTO— the failure mode is a partially written table, and the fix is the same one everywhere in this domain: write to a location nobody reads, validate, then publish by swapping a pointer (Atomic Publish). - A table whose files are wrong is repaired by rewriting the affected partitions, not by rewriting the query. Keep the raw arrival so that is possible (Keeping Raw History: The Recovery Position and the Liability).
What can go wrong
- A query that plans in seconds and then reads the entire table, because the predicate could not be pushed and nobody looked at the plan (Predicate Pushdown).
- Coordinator memory exhausted by a final aggregation or a broadcast join whose build side was much larger than the estimate suggested.
- Metadata dominating: hundreds of thousands of small files, where listing and opening objects costs more than reading them (File Compaction).
- A federated source overwhelmed by a scan the engine considered ordinary and the operational database considered an outage (Source Pushdown).
- Two engines over the same table producing different results because they resolved a type or a null differently, and both being believed.
- Your own limit as the failure: a memory cap that kills the one nightly query that legitimately needs it, at 03:00, silently, into a retry loop.
- "A query engine replaces the warehouse." It replaces the *loading* step. Warehouses own their layout, statistics and clustering, and that ownership is exactly what makes their planning better on the workloads they were tuned for (Lake vs Warehouse vs Lakehouse).
- "It is slow, so we need more workers." Adding workers helps when work is parallel and evenly split. It does nothing for a query that reads too much, for a coordinator-side bottleneck, or for one straggler split holding the whole stage (Data Skew).
- "The engine will optimise my query." It will reorder joins and push what it can push. It will not fix a predicate it cannot see through, a table with no statistics, or a layout that forces a full scan (Query Optimizers).
- "Same SQL means same answer." Two engines over the same files can differ on decimal scale, timestamp zone handling, null ordering and string collation. These differences are quiet and they change aggregates.
- A query engine is an access path that often bypasses the source system's own access control. The database's row-level policies do not travel with a Parquet export of that database, and the engine will happily serve columns the source would have masked (Row and Column Security).
- Catalog-level permissions become the real control plane. Whoever can register a table can grant reach to the data behind it, which makes catalog write access a privileged operation, not an administrative convenience (Data Access Control).
Operating it
- Per query: bytes scanned, splits scheduled, and rows output at each plan stage. Bytes scanned against rows returned is the single most useful ratio a data platform can plot (Scan Cost).
- The plan itself. Whether a filter appears as a partition predicate, a reader-level filter or an engine-level filter is visible in
EXPLAINand invisible anywhere else (Reading EXPLAIN ANALYZE). - Queue time versus execution time per query. When the two diverge the problem is concurrency and admission, not the query, and adding workers is a real fix rather than a hopeful one.
- File count and average file size per table, tracked over time. It only ever gets worse on its own (File Size and the Small-Files Problem).
- At 10x data, the engine is usually fine and the layout is not. The queries that degrade first are the ones whose predicates never pruned, because their cost was always proportional to the whole table.
- At 100x, planning itself becomes a cost: enumerating millions of files, computing splits and distributing them is work the coordinator does alone, and it is the first thing to become the bottleneck (Straggler Tasks).
- Consumer count scales differently from data volume: it is a concurrency and admission problem. A hundred analysts running modest queries can starve the cluster more effectively than one enormous job (Queueing: Why Systems Get Slow Before They Get Broken).
- Bytes scanned dominates, and is decided almost entirely by layout and by which pushdowns applied. This is the driver you can move by an order of magnitude with no new hardware (What Actually Drives Data Platform Cost).
- Metadata operations are their own cost line and behave differently: they are charged per request, so a small-files problem produces a bill that does not correlate with data volume at all.
- Bytes exchanged between workers is the second driver, and is a property of the join and aggregation strategy rather than of the data size (The Shuffle).
- Idle cluster capacity is a real cost of the always-on engine model, and the reason single-node engines are increasingly the right answer for datasets that fit on one machine (DuckDB Concepts).
- Separating engine from storage buys openness — many engines over one copy, no proprietary load step, storage priced and scaled independently. It costs every guarantee that came from owning the files: no indexes, no constraints, weaker statistics, and layout as somebody else's problem.
- One SQL interface over many systems buys reach and costs predictability. The same query text can be a fast scan of a local Parquet table or a slow drag through a remote database, and nothing in the SQL says which (Federated Query).
- Ad-hoc query power over raw files buys speed of exploration and costs governance: it is much easier to produce a number than to produce a number anyone else can reproduce.
Engine plan explorer
Change an input and watch which number moves — and which one does not. Everything here comes from a model in this repository, not from a measurement.
| Filter, then join | |
|---|---|
| Right when | The join sees only the rows that survive the predicate, so its build side is small and its output is smaller. |
| Wrong when | Nothing, when the predicate is on the base table. This is the rewrite every planner tries first. |
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- GENERALThe separation of a planning-and-execution engine from the storage it reads is architectural and stable. What varies is how much metadata the storage side offers — a raw directory of CSV gives the planner almost nothing, a table format gives it manifests and statistics.
- ENGINE-SPECIFICSplit sizing, memory management, spill behaviour and which pushdowns each connector implements differ substantially between engines, and between versions of the same engine. Treat every specific claim here as something to confirm against the plan your engine prints.
- SCALE-SPECIFICBelow roughly a single machine's worth of data, an in-process engine over the same files removes the coordinator, the exchange and the cluster entirely. The distributed architecture is a response to data that does not fit, not a maturity level.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns what it means for a coordinator and a set of workers to agree on a plan, survive a worker loss, and decide when a query is complete. That domain is being built separately.