DEV Community

Cover image for Teach Your Query Engine to Ignore Most of Your Data: A Field Guide to Partitioning, Clustering, Sorting, and Bucketing
Nariman Baubekov
Nariman Baubekov

Posted on Edited on

Teach Your Query Engine to Ignore Most of Your Data: A Field Guide to Partitioning, Clustering, Sorting, and Bucketing

Partitioning, clustering, sorting, and bucketing all answer some version of the same question: how do we organize data so the engine can ignore as much of it as possible? The differences between them come down to one thing — at what granularity each one operates — and mixing that up is the most common way people apply the right technique to the wrong problem.

That mismatch shows up most painfully in joins. A query engine's default move, when it doesn't know in advance where two tables' matching rows live, is to redistribute every row from both tables across the cluster, hashed by the join key, so anything that could possibly match ends up on the same node. That redistribution — a shuffle — is often the single most expensive thing happening in a join, and it has nothing to do with query logic or cluster size. It's a consequence of how the data is laid out on disk, which means it's fixable without touching the query at all.

This article is a field guide to that fix: partitioning, clustering, sorting, and bucketing — and the sometimes-surprising rules for how they combine, or don't, depending on the table format underneath them.

Contents

A quick note on scope: these four techniques matter most in architectures where storage and compute are decoupled — the warehouse/lakehouse (OLAP) model we're focused on here. Single-node table partitioning and multi-node sharding are both real, mature techniques used elsewhere — just different axes from what we're covering. Code examples default to Spark SQL, using Delta Lake syntax for partitioning, clustering, and sorting — but, as you'll see in the bucketing section, Delta doesn't actually support bucketing at all, so that one section uses plain Spark/Hive managed-table syntax instead. Snowflake, BigQuery, Redshift, and Apache Iceberg show up as clearly marked asides where their approach genuinely diverges — which, for at least two of these techniques, is more than the shared vocabulary suggests.

Why data layout matters more than you think

Most query engines — Spark, Snowflake, BigQuery, Trino, Redshift — share a common bottleneck: I/O. Reading data from disk or object storage is almost always slower than the computation performed on it. So the fastest query is the one that reads the least data.

Engines lean on two mechanisms to make that true:

  • Pruning — skipping entire files or blocks that can't possibly contain relevant rows, without opening them.
  • Predicate pushdown — filtering rows as early as possible, ideally while still reading from storage, rather than loading everything into memory first.

Partitioning, clustering, sorting, and bucketing are all techniques for making pruning and predicate pushdown more effective — they just operate at different levels of granularity, which is the thing that's surprisingly easy to lose track of once you start using them together.

Four techniques, four levels of granularity: directory, file, row order, hash bucket

Partitioning decides which directory a row lives in. Clustering decides which file, within or across those directories. Sorting decides row order inside a file. Bucketing is the odd one out — not really "finer" than sorting, just solving a different problem (joins, not filtering) using the same underlying idea of grouping related rows together.

Meet the tables

Let's set up one running example for the rest of this article — the two tables behind almost any e-commerce warehouse, and the same two tables behind the reconciliation-job scenario the next section walks through in detail:

CREATE TABLE web_orders (
    order_id      BIGINT,
    customer_id   BIGINT,
    order_date    DATE,
    country       STRING,
    product_sku   STRING,
    amount        DECIMAL(10,2)
)
USING DELTA;

CREATE TABLE customers (
    customer_id    BIGINT,
    signup_date    DATE,
    country        STRING,
    lifetime_value DECIMAL(12,2)
)
USING DELTA;
Enter fullscreen mode Exit fullscreen mode

web_orders is the large, fast-growing fact table; customers is smaller, but still too large to comfortably broadcast to every node once it's grown past a few million rows. Every technique in this article gets applied to one or both of these.

At small scale, none of this matters — a few million rows fit comfortably in memory and a join is fast regardless of layout. Both tables here are assumed to have grown into the hundreds of millions of rows, which is the point where layout decisions stop being theoretical.

A framework for the decision

Rather than treating "how should I lay this table out" as something to intuit, it helps to run it through a repeatable process: Symptom → Diagnose → Candidates → Decide → Verify. Here's that process applied to a shape of problem every data engineer eventually runs into: a nightly reconciliation job joining web_orders against customers that's finished in under 20 minutes every night for a year — until one night it doesn't. It blows past three hours and threatens the 6am deadline finance needs the numbers by, with nothing about the query or the cluster having changed. The only thing that changed is size: both tables grew past the point where the engine's default join strategy is still cheap.

  1. Symptom — the nightly join between web_orders and customers blows past its usual 20-minute runtime and threatens an SLA.
  2. Diagnose — the query profile shows one shuffle stage consuming the overwhelming majority of the job's time, moving far more data across the network than either table's actual size would suggest is necessary.
  3. Candidates — broadcast customers to every node instead of shuffling it (works only while it's small enough to fit in memory on each node); repartition both tables by customer_id right before every run (still pays a shuffle nightly, just a smaller one); or eliminate the shuffle entirely with a genuine bucket-to-bucket join — which, as the bucketing section below explains, isn't something Delta tables can do at all.
  4. Decidecustomers has already grown too large to broadcast reliably, and this join runs every single night indefinitely, not once — so a one-time layout change is worth it precisely because the shuffle alternative gets paid again, in full, every night forever. Since Delta doesn't support bucketing, that means migrating and maintaining web_orders and customers as Iceberg tables using the bucket(256, customer_id) transform specifically for this join, rather than trying to force a shuffle-free join out of Delta's clustering tools, which — as of today, mid-2026 — don't offer one.
  5. Verify — re-run the job and confirm the shuffle stage is gone from the query profile, replaced by a direct bucket-to-bucket join.

That's the lens for the rest of this article: each technique is a way of narrowing "how much data does this specific operation actually have to touch, or move."

Partitioning: splitting data into separate, skippable chunks

Partitioning divides a table into distinct physical segments based on the value of one or more columns — most commonly a date. In a partitioned web_orders, all rows for 2026-08-01 live in one directory, and all rows for 2026-08-02 in another:

/web_orders/order_date=2026-08-01/
/web_orders/order_date=2026-08-02/
/web_orders/order_date=2026-08-03/
Enter fullscreen mode Exit fullscreen mode
CREATE TABLE web_orders (
    order_id BIGINT, customer_id BIGINT, order_date DATE,
    country STRING, product_sku STRING, amount DECIMAL(10,2)
)
USING DELTA
PARTITIONED BY (order_date);
Enter fullscreen mode Exit fullscreen mode

Partition pruning: most folders are never opened for a date-filtered query

When a query filters on WHERE order_date = '2026-08-02', the engine can skip every partition except that one — it just doesn't list them. This is partition pruning, and it's the cheapest, highest-leverage optimization out there, because the engine never even opens the irrelevant files.

A caveat on "just doesn't list them": that's literally true for classic Hive-style tables, where the engine really does resolve a query down to a directory listing. It's not quite how modern table formats work, though. Delta Lake and Iceberg both prune against a metadata layer instead — Delta's transaction log and Iceberg's manifest files each record every live file's path and per-column statistics explicitly, so the engine already knows which files matter before touching the filesystem at all; it never issues a directory listing in the first place. The directory-per-value layout in the example above (order_date=2026-08-02/) is still what you'll typically see on disk with these formats, so the mental model of "one folder per value" holds up visually — it's just resolved through metadata now, not a filesystem LIST call.

Iceberg pushes this further with hidden partitioning: a table can be partitioned by a column transform like days(order_timestamp) while the underlying files are laid out arbitrarily, with no directory structure implying the partition value at all — because the engine relies entirely on the per-file partition values recorded in the manifest, never on where a file physically sits. Iceberg also has a distinct migration path for adopting an existing Hive-partitioned dataset's on-disk layout in place, but that's about the physical layout of a specific transform (or legacy data being brought in), not a pruning mechanism — actual pruning is manifest-based either way, regardless of which Iceberg catalog (Hive Metastore, Glue, REST, or otherwise) is tracking the table.

When it works well:

  • The column you filter on most often (date, region, tenant ID) is also a good partitioning key.
  • Partitions are reasonably large — hundreds of megabytes to a few gigabytes each, not thousands of tiny files.
  • Query patterns are predictable and filter-heavy on the partition column.

Where it goes wrong:

  • Over-partitioning. Partitioning by a high-cardinality column (customer_id, a full timestamp) creates millions of tiny partitions — a "small files problem" where metadata overhead swamps any I/O savings.
  • Partition skew. If most of the data lands in one partition (a single busy day, a dominant customer), that partition becomes a hot spot and defeats the purpose.
  • Mismatched query patterns. Partitioning web_orders by country doesn't help if most queries actually filter by order_date instead.

A good rule of thumb: partition by the low-to-moderate-cardinality column that shows up in the WHERE clause of most of your queries — usually a time dimension.

Clustering: co-locating related data

Clustering organizes data so that rows with similar values in a chosen column are physically stored near each other — even when that column isn't the partition key. Think of partitioning as sorting a bookshelf into sections by genre, and clustering as also grouping books by author within each section: coarse-grained skipping from partitioning, finer-grained skipping from clustering.

customer_id is a great example of a column that would make a terrible partition key (too high-cardinality, would explode into millions of tiny folders) but is exactly the kind of column clustering is built for:

OPTIMIZE web_orders
ZORDER BY (customer_id);
Enter fullscreen mode Exit fullscreen mode

Clustering narrows each file's value range, enabling file-level skipping

This is where the terminology gets genuinely confusing across vendors, so let's be precise about it:

Z-ordering vs. Liquid Clustering — not the same thing. ZORDER BY uses a specific space-filling-curve algorithm to interleave a column's bits so that rows close together on multiple dimensions end up physically close together on disk. Databricks' newer Liquid Clustering is a different mechanism entirely — it doesn't use a Z-curve, and it was built specifically to replace Z-ordering because Z-order's benefit degrades as more data is appended and needs full-table re-optimization to restore. If you're on a recent Databricks runtime, CLUSTER BY (Liquid Clustering) is generally the recommended default over ZORDER BY now, not an equivalent alternative to it.

What clustering does not do, as of this writing: both Z-ordering and Liquid Clustering only ever narrow what gets read — they don't guarantee that matching rows from two different tables end up on the same physical node, so joining two clustered tables on their clustering column has historically still required a full network shuffle, the same as joining two completely unorganized tables. Databricks has "Co-clustered joins" in Private Preview as of mid-2026, which removes that shuffle for Liquid-clustered tables specifically — early benchmarks show a meaningful win (roughly 51% faster, 87% less data shuffled on one internal benchmark) — but it isn't generally available yet, so treat clustering today as a data-skipping technique only, not a join-shuffle fix, until that feature ships more broadly.

Clustering is especially useful for:

  • High-cardinality filter columns that would make bad partition keys but are frequently filtered or joined on.
  • Multiple query patterns. Traditional partitioning locks you into one primary access pattern; clustering on several columns lets queries filtering on any of them benefit.
  • Avoiding the small-files problem that high-cardinality partitioning would otherwise cause.

Vendor aside — Snowflake: clustering keys work on a related but distinct mechanism. You designate columns, and a background service reorganizes micro-partitions so rows with similar values end up together, improving the effectiveness of Snowflake's automatic pruning metadata (min/max per micro-partition). It's conceptually close to Z-ordering's goal, achieved with different internals.

Vendor aside — BigQuery: this is the one worth catching before it causes confusion. BigQuery's CLUSTER BY is, mechanically, closer to what this article calls sorting — it physically orders rows within already-partitioned blocks, rather than doing Snowflake-style background micro-partition reorganization. Same word, meaningfully different mechanism from Delta or Snowflake's version of "clustering."

The tradeoff across all of these is maintenance cost. Clustering usually isn't one-time — new data can violate the clustering order, and periodic re-optimization is needed to keep the benefit. That consumes compute, so it's worth applying selectively to the largest, most-queried tables rather than everything.

Sorting: ordering rows within a file

Sorting is the most granular of the four: it controls row order within a file or block, not which file a row lands in.

Why does row order matter if the engine still has to open the file? Two reasons:

Zone-map pruning: sorted files have narrow, non-overlapping ranges; unsorted ones don't

  1. Zone maps / min-max statistics. Columnar formats (Parquet, ORC) store min/max values per column, per block. If a file's rows are sorted by order_id, each block covers a narrow, known range. A query filtering WHERE order_id = 4823901 checks those statistics and skips blocks whose range doesn't include it — without decompressing or scanning them. Unsorted, every block's range tends to span nearly the whole table, and nothing can be ruled out.
  2. Compression. Sorted data compresses better. Group country = 'US' rows together instead of scattering them among fifty others, and run-length/dictionary encoding become far more effective — shrinking file size, and in turn, I/O.
-- One-off sort during a write, or as part of periodic compaction
INSERT INTO web_orders
SELECT * FROM staging_orders
SORT WITHIN PARTITIONS BY order_id;
Enter fullscreen mode Exit fullscreen mode

Sorting and clustering don't stack the way partitioning and clustering do, and it's worth being precise about why: on a Delta table, both dictate the same thing — the physical row order within a file. Z-ordering web_orders by customer_id and then separately running a linear sort by order_id isn't two compounding optimizations; a file only has one physical row order at a time, so the second operation just overwrites the first. If both dimensions genuinely matter, the Delta-native move is a single multi-column Z-order — ZORDER BY (customer_id, order_id) — which blends both columns into one interleaved curve rather than giving either column a clean, independent ordering.

Sorting stacks cleanly with a co-location technique only where the two are mechanically separate concerns. A plain Spark/Hive bucketed table's CLUSTERED BY (customer_id) SORTED BY (order_id) combines a hash-based bucket assignment (which file a row lands in) with a genuine linear sort within that bucket (row order inside the file) — different mechanisms operating on different questions, not one overwriting the other.

Fragility gotcha: sort order isn't an invariant the way a uniqueness constraint is — nothing rejects a write that violates it, so it degrades silently rather than failing loudly. Three common ways it erodes:

  • Appends. A new batch lands as a brand-new file with its own min/max range, uncoordinated with the existing sorted layout. One new file barely matters; hundreds of appends without a re-sort start recreating the "wide, overlapping ranges" problem from scratch.
  • Merges/upserts. MERGE INTO rewrites whatever files the matched rows live in — in whatever order the merge operation happens to produce, not necessarily the original sort order. A single merge touching rows scattered across many keys can re-shuffle the sort order of every file it rewrites.
  • Bare compaction. Ironically, the maintenance job meant to help can hurt sort order if it's not told to preserve it — plain OPTIMIZE cares about file count and size, not row order, unless you explicitly pair it with a sort spec (OPTIMIZE ... ZORDER BY, not bare OPTIMIZE).

Because there's no error or warning when this happens, the usual symptom is "this query got slower over the last few months" with no obvious trigger — until someone re-runs the optimization job, or checks file-skipping stats to catch it earlier.

Bucketing: fixed-count hash partitioning for joins

Bucketing splits data into a fixed number of buckets based on the hash of a column's value — typically a join key. Unlike partitioning, the bucket count is set upfront, and every row is deterministically assigned via hash(column) % num_buckets.

One correction worth making explicitly: this isn't a Delta Lake feature. Delta Lake doesn't support Hive-style bucketing at all — CLUSTERED BY ... INTO N BUCKETS is plain Spark/Hive managed-table syntax, a different table type entirely:

CREATE TABLE web_orders (
    order_id BIGINT, customer_id BIGINT, order_date DATE,
    country STRING, product_sku STRING, amount DECIMAL(10,2)
)
USING PARQUET
PARTITIONED BY (order_date)
CLUSTERED BY (customer_id) SORTED BY (order_id) INTO 256 BUCKETS;
Enter fullscreen mode Exit fullscreen mode

Note that a plain Spark managed table can combine partitioning, bucketing, and an in-bucket sort in one statement — genuinely all three techniques compounding on the same table, which Delta specifically can't offer for the bucketing part.

Skipping vs. co-location — the distinction the previous section's clustering tools don't cross. Z-ordering and Liquid Clustering (Delta's tools) only narrow what gets read on one table at a time — they never guarantee that matching rows from two different tables end up in the same physical location, which is exactly what a join needs to avoid a shuffle. Bucketing's actual job is co-location: it guarantees every row with customer_id = 42 lands in a specific, deterministic bucket, on both tables, so the engine can align bucket 1 with bucket 1 without ever checking the data itself. That's a stronger, more specific guarantee than clustering makes, which is why bucketing solves the shuffle problem and clustering (for now) doesn't.

The main benefit isn't pruning — it's avoiding expensive shuffles during joins:

Bucket-to-bucket join stays node-local; an unbucketed join requires a network shuffle

If web_orders and a customers dimension table are both bucketed on customer_id with the same bucket count, the engine can perform a bucket-to-bucket join — matching bucket 1 to bucket 1, bucket 2 to bucket 2, and so on, without redistributing data across the cluster first. In distributed engines like Spark, that shuffle is very often the single most expensive step in a join, so avoiding it entirely is a large win, not an incremental one.

Bucketing also helps with:

  • Sampling — pulling a representative subset by reading a fixed number of buckets.
  • Predictable file counts — bucket count doesn't grow with key cardinality the way partition count does.

The downside is rigidity. Bucket count is usually fixed at table-creation time, and changing it means rewriting the entire table. Bucketing also assumes you know your dominant join key in advance — a more specialized tool than partitioning.

Vendor aside — Iceberg: this is the closest thing to a modern, lakehouse-native successor to Hive bucketing, with one structural difference worth noting: in Hive, partitioning and bucketing are two separate DDL clauses describing two different mechanisms. In Iceberg, bucket(N, col) is a partition transform — bucketing isn't a separate concept from partitioning at all, it's one specific way of defining a partition. Different plumbing, same practical outcome: paired with Spark's Storage Partitioned Joins feature, it achieves genuinely shuffle-free joins the same way Hive bucketing does, without the small-files brittleness of Hive-style directory-per-value partitioning.

Putting it together: a mental model

Technique Grain Primary benefit Best for
Partitioning Directory/file-group level Skip entire chunks without listing them Low-cardinality columns in most WHERE clauses (dates, regions)
Clustering File level, across or within partitions Co-locate related rows for better skipping High-cardinality filter columns, multiple query patterns
Sorting Row order within a file Tighter min/max stats, better compression Any column with frequent range/equality filters
Bucketing Fixed hash groups Avoid shuffles in joins/aggregations Known, stable join keys in distributed engines

A well-designed large table often uses more than one of these together — but which ones combine depends on the table format and, on Delta specifically, which clustering tool is in play. Partitioning by order_date (coarse pruning) and Z-ordering by customer_id (fine pruning) compound cleanly on the same Delta table. That combination assumes Z-order, though — Databricks' own documentation states Liquid Clustering is explicitly incompatible with both Hive-style partitioning and Z-order on the same table; it's designed to replace them outright, not layer on top of them, so a table using CLUSTER BY doesn't get to keep its PARTITIONED BY clause too.

Sorting doesn't stack with either clustering approach on Delta, for the reason covered above — they compete for the same physical row order. And getting a shuffle-free join against customers means stepping outside Delta's clustering tools entirely — either a plain Spark/Hive managed table bucketed and sorted by customer_id, or an Iceberg table using the bucket() transform, which is the path the reconciliation-job scenario above actually needed. None of these techniques compete with each other conceptually — they operate at different levels of granularity — but not every pair of them is available on every table format at once, which is exactly the kind of detail that's easy to miss until a query plan doesn't behave the way the theory predicted.

Practical guidance

Design around actual query patterns, not intuition. Look at query logs before choosing keys. The "obvious" partition column isn't always the one your queries actually filter on most.

Watch file sizes. Both too many tiny files and a few enormous files hurt performance. Most engines have a sweet spot — often 128MB–1GB per file — that balances parallelism against overhead.

Budget for maintenance. Clustering and sorting degrade as new data arrives. Whether you script it manually or rely on platform-driven automation (like Databricks' Predictive Optimization), factor in periodic operations like OPTIMIZE, VACUUM, and ANALYZE as an ongoing compute cost—not a one-time setup task.

Measure before and after. Check the query profile or EXPLAIN plan to confirm pruning is actually happening. A partition or clustering key that isn't showing up in the plan's pruning stats isn't earning its cost.

Don't over-engineer early. These techniques matter most at scale — tens of gigabytes and up. On smaller tables, the overhead of maintaining a complex layout can outweigh the benefit; a full scan of a small table is already fast.

Closing thought

That reconciliation job didn't need a bigger cluster or a faster network. It needed web_orders and customers moved onto a bucketed layout for that one join — a change that had nothing to do with compute and everything to do with the engine no longer needing to guess where a matching row might be sitting. Query optimization gets framed as a compute problem — bigger clusters, more memory, faster CPUs — but for most real-world workloads at this scale, the biggest wins come from an unglamorous source: making sure the engine never has to look at, or move, data it doesn't need. Partitioning, clustering, sorting, and bucketing are, at their core, different answers to the same question. Get it right — on a table format that actually supports the technique you need — and everything downstream gets faster and cheaper almost for free.


Have a layout decision that saved (or cost) you a surprising amount on your warehouse bill? Drop it in the comments.

Top comments (0)