Analytical Engines: Where the Speed Actually Comes From

An analyst runs a query that sums revenue for one product category over the last month. On the operational database it takes minutes and makes the on-call engineer nervous. On the analytical platform it returns in a fraction of that. The SQL is the same and the data is the same, so the difference is entirely in how each system goes about answering. The example is invented, but the experience is ordinary.

Explaining that gap requires being precise about what is being compared, because four different quantities get called “performance” and they do not move together.

  • Bytes scanned. How much data left storage. Column layout, encoding, and skipping act here, and on some platforms this is also the billing unit.
  • CPU work per byte. How many cycles it takes to process what was read. The execution model acts here, and it does nothing for the first quantity.
  • Wall-clock time. What the user experiences, which depends on both of the above, on how much work runs in parallel, and on what else is running.
  • Money. Compute time multiplied by its rate, or bytes scanned multiplied by a price, depending on the platform’s model.

The layers below attack different ones of these, which is why improvements do not simply multiply: reading ten times fewer bytes does not help a query that was bound by CPU on the rows that survived the filter, and a faster execution model does nothing for a query whose time is spent waiting on storage. What compounding does happen comes from removing successive bottlenecks, and the only way to know which one you have is to look. This article explains how each layer works, where each stops helping, and what it costs to operate.

Two engines, two different bets

The operational database is not badly designed. It is designed for a different access pattern, and the design decisions follow from that.

Operational (OLTP)Analytical (OLAP)
Typical requestRead or change a few rows identified by keyScan many rows, read few columns, aggregate
What it optimizesLatency per transaction, concurrency of small writesBytes read per query, throughput over large scans
Natural layoutRows together, so one row is one contiguous readColumns together, so one column is one contiguous read
IndexesMany, to find individual rows fastFew or none; the scan is the plan, made cheap by skipping
Write patternMany small writes, updated in placeBulk appends and rewrites; updates are expensive
Cost of the other workloadAn analytical scan competes with transactions for buffer pool and CPUA single-row lookup is possible but wasteful

The last row is the practical one. Running the month-end report against the order database is not slow because the database is bad at aggregation. In the invented example it is slow because no access path suits the query: with no index covering the filter and the two columns being summed, the planner falls back to reading the table. Given a suitable index the same database can do much better — PostgreSQL, for instance, documents index-only scans, where a query answered entirely from an index avoids touching the table at all. What remains true regardless is the second problem: a long scan competes for the buffer pool and CPU that the checkout path needs, so the report slows down transactions whether or not it is efficient in itself.

Reading less: the column layout

The first reduction comes from not reading columns nobody asked for. In a row-oriented file, the values of one row sit together, so reading two columns out of fifty still means touching every row’s worth of bytes. Storing each column contiguously turns that into reading two columns’ worth. For a wide table and a narrow query, this alone is most of the difference.

The complication is that a single unbroken column spanning a whole table is an awkward unit of work: whether part of it can be read on its own depends on how it is blocked, indexed, and compressed, and dividing it among many readers requires agreeing where the boundaries are. The formats used in practice make those boundaries explicit by splitting first by rows and then by column. Parquet’s documented structure is that a file holds N columns divided into M row groups, where each row group contains all the columns for one subset of rows, stored as one column chunk per column. The file ends with its metadata, followed by the metadata length and a magic number, which the documentation explains is what allows the file to be written in a single pass. The details of this nesting are covered in row group and column chunk, and the format comparison in Parquet, ORC, Avro, and Arrow.

That two-level structure is what makes the rest straightforward. A row group is a self-contained unit of work that one reader can take, and it is small enough that per-chunk statistics describe something narrower than the whole table. Reading only the columns a query names is column pruning; it is the cheapest optimization in the stack because it needs no statistics and no tuning, only a query that does not say SELECT *.

Reading less again: encoding and compression

Putting one column’s values next to each other has a second effect that is easy to underrate. Values in a column are of one type and often highly repetitive, which makes them far more compressible than a row’s worth of mixed fields. Columnar formats exploit this with encodings chosen per column.

  • Dictionary encoding replaces repeated values with small integer codes into a dictionary of distinct values. A country column with 200 distinct values across a billion rows becomes a billion small integers plus a tiny dictionary.
  • Run-length encoding stores a value once with a count of how many times it repeats. It is transformative on a sorted or clustered column and does nothing on a shuffled one.
  • Bit-packing uses only as many bits as the range needs, so dictionary codes for 200 values take 8 bits rather than 32.
  • Delta encoding stores differences rather than values, which suits monotonically increasing columns such as timestamps and identifiers.

General-purpose compression is then applied on top. The important point for a platform team is that the effectiveness of all of this depends on properties of the data that a pipeline controls: cardinality and ordering. A column with few distinct values encodes well. A column sorted or clustered on disk encodes far better than the same column shuffled. Sorting on ingest is therefore not only a query optimization but a storage cost decision, and the two usually point the same way.

There is a limit worth knowing, and it is narrower than it is usually stated. What high cardinality defeats is dictionary encoding: a dictionary the size of the column saves nothing. It does not automatically defeat compression, and the distinction matters.

  • A sequential identifier is entirely distinct and compresses extremely well, because the differences between consecutive values are tiny and delta encoding stores those instead.
  • Free text is distinct value by value but still contains shared prefixes, repeated words, and skewed character frequencies, all of which general-purpose compression exploits.
  • Random or hashed values are the genuinely hard case. They are distinct and carry no structure, so neither dictionaries, deltas, nor entropy coding find anything to remove.

So when a table is unexpectedly large, the useful check is not just how many distinct values a column has but whether those values have structure. A UUID column and an auto-incrementing key column have the same cardinality and very different footprints.

Reading less a third time: skipping what cannot match

The largest reduction available is often not reading the data at all. If the engine knows the minimum and maximum of a column within a chunk of data, and the query filters that column to a range outside it, the chunk can be skipped without being opened. Nothing about this requires an index in the operational sense; it requires small statistics stored next to the data, and formats keep them at more than one granularity, so skipping can happen per file, per row group, and in some cases per page within a column chunk.

Three mechanisms do this at different granularities, and they are often confused.

MechanismHow the engine knowsWhat it needs from you
Partition pruningPartition metadata records which values each partition holds, so partitions that cannot match are excluded before any data is opened. Where that metadata lives varies: a Hive-style layout encodes it in the directory path, a catalog or table format keeps it as metadataA partition column chosen at design time, and queries that filter on it
File or chunk skippingMin and max statistics recorded per file, per row group, and in some formats per pageNothing explicit, but how much it excludes depends on how narrow each chunk’s range is, which is what clustering controls
Predicate pushdownThe filter is evaluated as low as possible, at the storage layer or file reader, rather than after loadingA query shape the engine can push down, and a source that supports it

Snowflake’s documentation, checked in September 2026, describes a version of this that removes the design-time choice. Its tables are divided automatically into micro-partitions of 50 to 500 MB of uncompressed data, formed in the order data arrives, with columns stored and compressed independently inside each. It keeps per-partition metadata including the range of values for each column and the number of distinct values, and uses that to prune at query time. Its documentation argues that this beats traditional static partitioning on three counts: nothing has to be defined or maintained up front, the units are small enough for efficient DML, and because ranges may overlap, the scheme does not produce the skew that a badly chosen partition key does.

Automatic or not, how much skipping is available depends on how the data is physically arranged. Statistics work on any layout — a chunk whose maximum is below the filter value is excluded regardless of how the table was written — but the narrower and less overlapping the ranges, the more chunks a given predicate can exclude. Clustering is the lever for that, and Snowflake measures the result as clustering depth, the average depth of overlapping micro-partitions, where lower is better, while noting that query performance is a better indicator of good clustering than the depth number by itself. The limiting case is the one to recognize: if every chunk spans the full range of values, no predicate on that column excludes anything, and the statistics cost space without saving reads.

Two failure modes follow. Partitioning on a high-cardinality column produces a partition per value and the small file problem below. Partitioning on something queries never filter by produces no pruning at all, and teams discover this only when they read a query profile and find the scan reading everything. A partition scheme is a guess about how the table will be queried, and the way to check the guess is to look at the query log and the engine’s plan output, not at the scheme itself.

A heavily partitioned table has a second cost that is easy to mistake for a scanning problem: retrieving the partition list itself. Athena’s documentation describes this directly and offers partition projection as the remedy — rather than calling the catalog to enumerate partitions, Athena computes partition values and locations from properties configured on the table, which avoids the metadata lookup that becomes a bottleneck when a table has very many partitions. It is a fix for planning time, not for scan volume, and it applies where partitions follow a predictable pattern such as a continuous range of dates or integers. Its documentation also warns that if more than half the projected partitions are empty, traditional partitions perform better.

Doing more per cycle: the execution model

Once the engine has decided what to read, the remaining question is how fast it can process it. Here the decisive change happened in query engine design rather than in storage.

The classical design is the Volcano iterator model, in which each operator exposes a next() that returns one tuple, and a query is a tree of such operators pulling from each other. It is elegant and composes well. Its cost is that every value passes through a function call, and the 2005 MonetDB/X100 paper measured what that means. In MySQL, the routine implementing an addition cost 38 instructions per addition and roughly 49 cycles on the machine tested, of which about 20 cycles were the call itself, amortized over a single operation. More seriously, the paper’s argument is that tuple-at-a-time execution causes high interpretation overhead and hides opportunities for CPU parallelism from the compiler, because a loop that performs one addition cannot be pipelined.

The opposite extreme has its own problem. The paper analyzes MonetDB’s earlier column-at-a-time model, which avoids interpretation overhead entirely but materializes whole columns between operators, so on decision support workloads it became bound by memory bandwidth instead.

Vectorized execution is the middle position the paper proposes and that most modern analytical engines now use: operators still pull from each other as in Volcano, but each call returns a vector of values rather than one tuple. The per-call overhead is amortized over hundreds of values, the inner loop is a simple loop over an array that a compiler can pipeline and vectorize, and the intermediate results stay small enough to live in CPU cache. X100 used a default vector size of 1024, with the reasoning that all vectors together should comfortably fit the CPU cache, while too small a vector loses the CPU parallelism and brings back the interpretation overhead. On a 100 GB TPC-H workload the authors reported raw execution power one to two orders of magnitude above the systems they compared against, on 2005 hardware.

Two things follow for someone choosing or operating an engine. An engine that reads columnar files but executes tuple-at-a-time gets the storage benefit and not the CPU benefit, which is one reason two systems reading identical Parquet can differ several-fold. And the vector-in-cache design is why analytical engines care so much about data types: a fixed-width numeric column processes in a tight loop, while a variable-length string column does not.

Doing it in parallel: how the work is split

Vectorization is about efficiency inside one execution unit. A massively parallel engine adds a second, independent dimension: many units working on different parts of the data at once. The two are complementary, and confusing them makes performance discussions circular, because a query can be slow from poor CPU efficiency, from poor distribution, or from both.

Take the opening query — revenue by category for one month — on a table split across many files and many workers. A typical plan has four stages.

  • Parallel scan. Each worker reads a disjoint set of files or row groups and applies the date filter. If the table is partitioned by date so that partition metadata settles the filter, it reads two columns, category and revenue; otherwise it reads three, because the date column has to be examined to evaluate the predicate. No intermediate results are exchanged between workers at this stage, and it scales with their number as long as storage keeps up.
  • Local aggregation. Each worker sums revenue per category over only the rows it read. Its output is at most one row per category rather than one per input row, which is why this stage exists: it shrinks what has to cross the network next.
  • Redistribution. Partial results are exchanged so that every partial sum for a given category lands on the same worker. This is the shuffle: the stage in this plan that redistributes intermediate results among workers. Data crosses the network elsewhere too — reading from remote storage, and returning the result — but those are not stages the plan can rearrange.
  • Final aggregation. Each worker adds up the partials it received for its categories, and the results are returned.

Two consequences of this shape explain most distributed query behavior. First, the stages that do not exchange data scale easily and the one that does is the constraint, so a great deal of engine design is about moving less across that boundary — pushing filters earlier, aggregating locally first, and choosing join strategies that avoid a shuffle. Broadcasting a small table to every worker so a join can happen locally is exactly that trade: it spends memory and network on copies in order to remove an exchange.

Second, a stage finishes when its slowest worker finishes, so uneven division costs the whole query. Where that unevenness comes from depends on the operation, and the aggregation above is a poor example of it: because each worker reduces its rows to one per category before the exchange, a category with a hundred times the rows produces the same single partial row as any other, and the local aggregation absorbs most of the imbalance. Simple sums and counts are forgiving that way.

The cases that are not forgiving are the ones where the local step reduces little, so something close to raw rows has to be redistributed. A join on a key whose values are heavily concentrated sends all the rows for that key to one worker, which then does far more work than its peers. Exact distinct counts are a milder version of the same thing: a worker can deduplicate locally, but when the distinct values are numerous that deduplication removes little, so what crosses the network stays large. Imbalance can also arrive before any exchange, when the input splits handed to workers differ greatly in size. That is not the same as one large file: a Parquet file is read in row groups, so a large one can be divided among several workers. The problem case is an input that cannot be split — a compressed format without block boundaries, or a single row group holding most of the data. These are data skew, a different failure from reading too much or computing inefficiently, with different remedies, covered in partitioning, shuffles, and data skew.

Choosing the plan: statistics and the optimizer

Between the SQL and the execution sits the part most often blamed and least often understood. The question the optimizer answers is which of many equivalent plans to run: which order to join tables in, which join algorithm to use, when to sort, what to push down.

The approach in use today was laid out in 1979 by Selinger and colleagues in the System R access path selection paper. Its structure has three parts that still describe modern optimizers. Costs are estimated and compared, with the lowest-cost path chosen, where cost is a weighted sum of disk I/O and per-record calls into the storage interface. Result sizes are estimated from selectivity factors, the fraction of rows a predicate is expected to keep, computed from catalog statistics such as the number of rows in a relation, the number of pages it occupies, and the number of distinct keys in an index. And join orders are enumerated rather than fixed, with the refinement of interesting orders: a plan whose output happens to be sorted in a way a later operation wants is kept even if it is not the cheapest so far, because that sort order may save more later.

The practical consequence is that plan quality depends on statistics being both present and current, and the symptoms of stale statistics are recognizable.

  • A join keeps a strategy that no longer fits. A plan switching from broadcast to shuffle after a table grows is the optimizer working correctly. The symptom of stale statistics is the opposite: the engine still believes one side is small, broadcasts a table that has since grown past the threshold, and the query fails or thrashes memory. Spark’s broadcast threshold is a size in bytes compared against the statistics the catalog holds, which is why refreshing those statistics changes the plan.
  • A query that ran for months regresses sharply after a load, with no change to the SQL.
  • The plan shows an estimated row count orders of magnitude away from the actual, which is the single most useful number to check in any query profile.

Estimation is hardest exactly where analytical queries live. Correlated predicates, skewed value distributions, and filters on expressions rather than columns all push estimates away from reality, and errors compound through a multi-join plan. This is why engines increasingly adapt at run time: Spark’s adaptive execution converts a sort-merge join into a broadcast hash join when the runtime statistics of one side turn out to be below the threshold, which it describes as less efficient than having planned the broadcast in the first place but better than continuing with the sort-merge. Adaptation narrows the damage from bad estimates; it does not remove the value of good ones. When estimated and actual row counts are far apart, look at the data and the statistics before rewriting the SQL.

Where the theory meets the file listing

Everything above assumes chunks of a reasonable size. A recurring way a well-designed analytical setup performs badly is that it is not reading a few large files but thousands of tiny ones.

Small files arise naturally: a streaming job that commits every minute writes a file per minute per partition, and a partition scheme with high cardinality multiplies that. The cost is not mainly the bytes. It is that opening a file has a price unrelated to how much is inside it, and engines model this explicitly. Spark exposes a setting for the estimated cost of opening a file, expressed as the number of bytes that could have been scanned in the same time, and uses it when deciding how many files to pack into one read task; its default is 4 MB. That number is a planning assumption rather than a measured property of any storage system, and the documentation notes that over-estimating is deliberate. Treat it as the shape of the problem — below some size, a file is mostly overhead — and establish the actual size for your storage and runtime by measuring.

  • Per-file request cost. On object storage each file means at least one metadata lookup and one or more range requests, each with its own latency. Enough files turn a scan into a latency problem rather than a bandwidth one.
  • Planning cost. Every file considered is work before any data is read: enumerating candidates and evaluating their statistics. How expensive that is depends on where the file list lives — a directory listing on object storage is slower than a catalog or manifest that the table format maintains, and results may be cached — but it scales with the number of files either way.
  • Weaker encoding. Dictionaries, run lengths, and per-chunk metadata are built per chunk, and their fixed cost takes a larger share of a small one. Whether that matters depends on the data rather than on the row count alone — a hundred rows repeating a few long strings still dictionary-encode well, while a hundred varied rows mostly pay overhead — but the same data spread across many small chunks generally occupies more space than in a few large ones.
  • Scheduling overhead. Each read task has setup cost, and thousands of tiny tasks spend a growing share of the query coordinating rather than reading.

Note what is not on that list. Fine-grained statistics are not the problem: smaller units describe narrower ranges and can in principle be excluded more precisely. The problem is that the fixed cost per file eventually exceeds what that precision is worth.

That framing is also what makes compaction a trade rather than a cleanup. Larger files amortize the per-file cost and encode better; smaller files allow finer skipping and cheaper rewrites. The target size is the point where those balance for the workload at hand, which is why engines ship both a file-open cost and a maximum bytes-per-read-task as tunable numbers. And compaction is a scheduled job somebody owns, competing for the same compute as the queries.

Caches, concurrency, and who slows down whom

The last layer is what happens when more than one query runs. Analytical engines cache aggressively — remote files on local disk, decoded data in memory, sometimes whole result sets — and caching is what makes the second run of a dashboard query fast. It also makes benchmarks lie, because the number everyone remembers is the warm one.

Concurrency turns caching into a contention problem. Three resources are shared, and they fail differently.

  • Memory. A large join or sort that does not fit spills to disk, which is a cliff rather than a slope: the query does not get slightly slower, it gets much slower. Several such queries at once turn a cluster into a disk-bound one.
  • CPU and slots. A long scan holds workers for its duration. Short interactive queries queued behind it inherit its latency, which is how one backfill makes an entire BI tool feel broken.
  • Cache. A large scan evicts the working set of the dashboards, so their next run is cold. The damage outlasts the query that caused it.

This is the argument for workload isolation: separate compute for interactive queries and for heavy batch work, so that the second cannot evict or starve the first. The cost is lower utilization and a cold cache on the smaller cluster, which is a real trade rather than a free win. A useful rule when sizing is that the number to protect is the latency of the short queries, because that is the one users notice, while the throughput of the long ones is usually a scheduling question.

It is also worth remembering the case where none of this applies. A dataset of a few gigabytes may be handled better on one machine than by a distributed engine, because at that size the coordination that makes distribution work can cost more than the parallelism returns, which is the argument in single-node analytics and, for the in-memory case, working with tabular data at scale.

Diagnosing a slow query in order

The layers above suggest an order for looking, cheapest to check first. The order is not a claim about which causes are most frequent; it puts the checks that take one glance at a query profile ahead of the ones that need an experiment.

CheckWhat a bad answer looks likeUsual remedy
How many bytes did it read, against the table size?Close to the whole table for a filtered queryNo pruning is happening: check the filter against the partition and clustering columns
How many files did it open?Thousands, for a modest amount of dataCompaction, and a partition scheme with lower cardinality
Which columns were read?All of themRemove SELECT *, including inside views
Estimated rows versus actual, at each stepOrders of magnitude apartRefresh statistics; look for correlated or expression filters
Did it spill?Disk written during a join or sortMore memory per worker, a smaller working set, or a different join strategy
Was the work evenly divided?One task running long after the others finishedSkew handling on the join or grouping key
What else was running?The query is fast in isolation and slow in the morningIsolation or scheduling, not query tuning

Questions to explore further

  • For your largest table, what fraction of its bytes does a typical query actually read, and does anyone track that number over time?
  • Which of your partition columns appear in the filters your users actually write, and which were chosen because they were convenient to the loader?
  • What is the average file size in your busiest table, and who owns compaction for it?
  • When a dashboard is slow in the morning and fast in the afternoon, do you know whether that is caching, contention, or a scheduled job?

References

All sources were checked on September 15, 2026. Product behavior described from vendor documentation reflects that date and changes over time; measurements from the 2005 paper reflect the hardware of that time.


Discover more from Insightful Data Lab

Subscribe to get the latest posts sent to your email.

Similar Posts

Questions, corrections, or additional insights?

This site uses Akismet to reduce spam. Learn how your comment data is processed.