Databricks / delta lake / data layout / optimize
Databricks Liquid Clustering
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.
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.
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".
rewrote no more than the day's ingest · rewrote more than the day's ingest · the most expensive night so far
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 FULLwas 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
OPTIMIZEagain. - 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.