DeepConcepts

Databricks / delta lake / data layout / optimize

Databricks Liquid Clustering

The misconception

That CLUSTER BY is PARTITIONED BY without the directories, and that OPTIMIZE on a clustered table is OPTIMIZE ZORDER under a new name. The incremental part is a candidate-selection rule, not a better sort: OPTIMIZE ZORDER makes every file in a partition a candidate on every run, while clustering excludes files already sealed into a full-size ZCube. The same rule is why changing clustering keys silently stops improving your layout until you run OPTIMIZE FULL, and why very small OPTIMIZE batches rewrite the same bytes several times over.

15 min

Liquid clustering is sold as the replacement for partitioning and for Z-ordering, and both halves of that sentence are true. But the thing that actually changed is not how the data gets sorted. It is which files the next OPTIMIZE is allowed to touch. Everything surprising about clustering — including the two ways it silently stops working — falls out of that one rule.

A quick vocabulary check, because three similar-sounding things are about to be compared. Partition columns put rows into directories named event_date=2026-08-13/; the value is in the path, so the engine can skip a whole directory without opening anything. Z-ordering is a command, OPTIMIZE … ZORDER BY (c), that rewrites files so each one holds a narrow range of c; nothing is renamed, and the engine skips files using the per-file minimum and maximum values recorded in the log. Clustering keys, set with CLUSTER BY (c), are a property of the table that makes every future OPTIMIZE do the Z-ordering job automatically, and that can be changed later.

Below is a table that receives the same amount of new data every day for two months. You want queries filtering on customer_id — a high-cardinality column, millions of distinct values — to read few files. Pick a maintenance strategy and drag the day slider. Watch the top-left number: gigabytes rewritten per gigabyte ingested.

Nightly maintenance

A ZCube is the unit of clustered output: one group of files written together by one clustering run, tagged with a shared identifier. Once a ZCube reaches the minimum size it is sealed, and later runs stop considering its files. That single exclusion is the whole of "incremental".

GB rewritten per GB ingested
rewritten by tonight's run
rewritten since day 1
table size
files a customer_id lookup opens
Gigabytes rewritten by each night's OPTIMIZE, day 1 to today

rewrote no more than the day's ingest · rewrote more than the day's ingest · the most expensive night so far

Layout units on storage

sealed — no future OPTIMIZE will read it · still open, will be rewritten again · clustered by keys you no longer use

Push the day slider to 60 and compare the top two options against the third. Whole-table Z-ordering rewrites the entire 2.3 TB table every night — 71 TB of writes to maintain 2.3 TB of data — and a point lookup still opens 60 files. Adding WHERE event_date >= today − 3 cuts the nightly write from 2.3 TB to 120 GB, and the lookup opens exactly the same 60 files. The expensive version was buying nothing. Clustering writes the same 120 GB a night and the lookup opens 20 files, because the 60 daily partitions have been replaced by 20 ZCubes that each cover the whole key range once.

Where the clustering keys actually live

A clustered table has no directories. The keys are stored once, in a domainMetadata action in the transaction log under the domain delta.clustering:

{"domainMetadata":{
   "domain":"delta.clustering",
   "configuration":"{\"clusteringColumns\":[\"customer_id\"]}",
   "removed":false}}

That is the entire on-disk representation of "this table is clustered by customer_id." Nothing about a data file's path depends on it. Three consequences follow immediately, and they are the differences from partitioning that matter in practice.

Changing the keys is a metadata write. ALTER TABLE t CLUSTER BY (region, customer_id) appends one commit and returns. Changing partition columns is not possible at all: you rewrite the table into a new one. This is the headline benefit, and it is also the trap, because a command that returns in 200 milliseconds having moved no data is easy to mistake for a command that reorganised your table.

Cardinality stops being dangerous. A partition column with a million distinct values creates a million directories and a million-way file listing; this is the classic over-partitioning failure. A clustering key with a million distinct values creates nothing. The keys only have to be columns that carry statistics — by default Delta records minimum and maximum values for the first 32 columns of the schema, controlled by delta.dataSkippingNumIndexedCols — because file skipping reads those statistics, and that is the only channel through which clustering helps a query.

You cannot have both. The protocol states it flatly: writers must not define a clustered and a partitioned table at the same time. In practice you meet this as DELTA_CLUSTER_BY_WITH_PARTITIONED_BY — "Clustering and partitioning cannot both be specified" — or, on an existing table, as ALTER TABLE CLUSTER BY cannot be applied to a partitioned table. That exclusivity is not an oversight. It is what keeps clustered tables eligible for row-level conflict detection, which Delta's optimistic concurrency control switches off permanently on any partitioned table.

A clustered table is limited to four clustering keys (DELTA_CLUSTER_BY_INVALID_NUM_COLUMNS), and Databricks notes that on tables under 10 TB, filtering on one column of a four-key table performs worse than filtering on one column of a two-key table. More keys means each individual key gets less of the ordering.

Why ZORDER could not be incremental

Run OPTIMIZE t ZORDER BY (customer_id) twice in a row with no writes in between and the second run rewrites the whole table again. This is not a missing optimisation; it is a direct consequence of two lines in Delta's optimize planner.

First, candidate selection. Plain bin-packing OPTIMIZE keeps only files smaller than the target size, which is how it fixes the small-file problem. For any multi-dimensional clustering — Z-order or liquid — that filter is skipped entirely:

// Select all files in case of multi-dimensional clustering
if (isMultiDimClustering) return files

Second, bin packing. A bin is one group of input files that becomes one group of output files. ZOrderStrategy sets maxBinSize = Long.MaxValue, with the comment that this gives a single bin per partition. Every file in a partition goes into one bin, so every file in that partition is read and written. There is no bookkeeping anywhere that records "these files were already sorted together," so there is nothing a second run could skip.

ClusteringStrategy adds exactly that bookkeeping. When a clustering run writes files, it stamps each one with clusteringProvider: "liquid" in its add action and a shared ZCube identifier in the file's tags. The next run sorts candidate files by ZCube identifier and applies three filters, in this order:

  • Files whose ZCube was built from different clustering columns are dropped — unless OPTIMIZE FULL was requested. This is the rule behind the key-change trap below.
  • Files in a ZCube whose total size is at or above the minimum cube size are dropped. In open-source Delta that minimum defaults to 100 times the target file size, so 100 GB with the default 1 GB files, and new cubes are packed up to 1.5 times that. Sealed means sealed: those bytes are never read by OPTIMIZE again.
  • If everything left belongs to a single ZCube, drop it too. Merging a cube with itself would rewrite bytes and change nothing.

That is the entire difference. Not a better sort — the sort is a Hilbert curve instead of a Z-order curve, which is a real but second-order improvement, and for a single clustering key the code falls back to Z-order anyway. The difference is a rule about which files are candidates.

One consequence is worth stating on its own: OPTIMIZE on a clustered table refuses a WHERE clause, because the whole point is that it now decides for itself which files need work. You get DELTA_CLUSTERING_WITH_PARTITION_PREDICATE: "OPTIMIZE command for Delta table with clustering doesn't support partition predicates." The windowed-Z-order habit does not carry over, and it does not need to.

The cost you did not expect: small batches

Incremental does not mean cheap. Set the panel to clustering, leave the minimum ZCube size at 100 GB, and set the daily ingest to 10 GB. Write amplification climbs to 5.5×. Now raise the minimum to 400 GB with the same 10 GB per day and it reaches 17.2× by day 60.

The mechanism is visible in the decision log. An open ZCube is a candidate until it seals, so each night's run reads the whole open cube plus the new data and writes it all back out. With 10 GB of new data and a 100 GB minimum, the cube grows 10, 20, 30 … 100, and the run costs 10, 20, 30 … 100. That is 550 GB written to absorb 100 GB. Sealing happens once; the walk up to it is paid every night.

So the advice to run OPTIMIZE hourly on a busy table has a condition attached that the docs do not spell out: it is cheap when each run brings enough new data relative to the cube size that cubes seal quickly. The ratio you care about is new bytes per run divided by minimum cube size. Below about one, you are re-writing the same open cube repeatedly.

You cannot fix that by shrinking cubes, because the minimum cube size is a trade, not a tuning knob with a good side. Drag the ingest slider to 200 GB and the minimum to 25 GB: amplification falls to 1.0× — every run's output seals immediately and is never read again — and the lookup readout jumps from 20 files to 360. Small cubes are cheap to write and useless to read, because each cube independently covers the whole key range and a query has to open one file from every single one. Large cubes skip well and cost a long climb to seal. The default of 100 times the target file size sits deliberately near the expensive end.

Databricks reduces this in ways the open-source planner does not: some writes cluster their own output before committing, above published size thresholds — 64 MB of new data for a one-key Unity Catalog managed table, rising to 1 GB for four keys, and four times those figures for other Delta tables. Data below the threshold arrives unclustered and waits for OPTIMIZE. The simulation models the open-source planner, so treat its absolute numbers as illustrative of the mechanism rather than as a forecast for a managed table.

The boundary: changing the keys

Tick change clustering keys on day 30 with clustering selected and watch the last readout. Every ZCube built before day 30 turns hot, the lookup cost goes from 20 files to 1,171, and the nightly rewrite volume falls, from 120 GB to 40 GB. Nothing failed. Nothing was logged as a warning. OPTIMIZE ran and reported success every night.

The first filter did this: files whose ZCube was built from different clustering columns are not candidates. Those bytes are now permanently excluded from the incremental path. They are still clustered — by the old keys, which nobody queries by any more. The table quietly splits into a growing well-clustered region and a frozen badly-clustered one, and the only external symptom is that queries got slower.

Now tick run OPTIMIZE FULL that day. One night's bar goes off the scale — it rewrites everything written so far — and from the next day the incremental cycle resumes on the new keys. That is the intended workflow, and it is why Databricks documents OPTIMIZE FULL as the thing to run "when you enable clustering for the first time or change clustering keys." Available from Databricks Runtime 16.4 LTS; before that, changing keys meant rewriting the table yourself.

The same trap catches the more common case of enabling clustering on an existing table. ALTER TABLE t CLUSTER BY (customer_id) on a table with a terabyte of history commits instantly and marks none of that history. Those files have no clusteringProvider at all, so they are candidates — but they are candidates for the ordinary incremental path, which packs them into cubes a few at a time alongside new data. If the table is large and the daily ingest is small, that takes months. OPTIMIZE FULL is what compresses it into one deliberate, expensive run you can schedule.

Two smaller boundaries worth knowing. ALTER TABLE t CLUSTER BY NONE stops future runs from clustering but does not un-cluster anything. And enabling clustering moves the table to writer version 7 and reader version 3, turning on deletion vectors and row tracking by default; clients that do not implement those features stop being able to read the table, and the protocol version cannot be lowered again.

Checking it yourself

DESCRIBE DETAIL my_table returns a clusteringColumns array. Empty array plus a non-empty partitionColumns array means you are on the old path. A populated clusteringColumns tells you the keys are set — it does not tell you any data is clustered by them.

For that, read operationMetrics on the last OPTIMIZE row of DESCRIBE HISTORY my_table. A clustered table reports a clusteringStats structure whose fields are the candidate rule made visible: inputZCubeFiles are the files that came from cubes matching your current keys, inputOtherFiles are files in no cube or in a cube built from different keys, inputNumZCubes is how many distinct cubes were read, and numOutputZCubes is how many were written. A run that reports numFilesAdded: 0 on a table you believe needs work is telling you every candidate was filtered out — check whether inputOtherFiles is large, which means stale keys, or whether everything is already sealed.

At the file level, dump a commit from _delta_log and look at an add action. A clustered file carries "clusteringProvider": "liquid". A file without that field has never been through a clustering run, whatever the table properties say.

Finally, SHOW TBLPROPERTIES my_table exposes clusterByAuto. When it is true the keys were chosen for you: on Unity Catalog managed tables from Databricks Runtime 15.4 LTS, CLUSTER BY AUTO lets predictive optimization pick and change keys based on the table's own query history, and it only changes them when the predicted skipping gain outweighs the reclustering cost. If that is on, the clusteringColumns you see today are not necessarily the ones you set.

A 4 TB table has been clustered by (order_id) for a year and queries are fast. The team adds a new dashboard that filters on region, so they run ALTER TABLE t CLUSTER BY (region, order_id). Two weeks later the dashboard is still slow, nightly OPTIMIZE takes the same six minutes it always did, and DESCRIBE DETAIL confirms both clustering columns. What is happening?

Read next: the per-file statistics that clustering exists to sharpen, because clustering with no skipping is just an expensive sort; and why staying unpartitioned changes how concurrent writers conflict. The word "partition" in all of this means directories on object storage, not Spark's in-memory partitions.

Why this concept is on the site

Topics are chosen from places engineers visibly get stuck, and the sources are kept with the lesson so the claim is checkable.