What is Z-ordering?

How it works

Data skipping relies on each file’s min and max per column. If rows are written in random order, every file holds almost the full range of every column, and a filter such as customer_id = 4242 must read every file.

Sorting by customer_id would fix that column but leave product_id random. A Z-order curve interleaves the bits of several columns’ values into one sort key, so nearby values in each column tend to stay close. The result is a compromise: skipping works on every Z-ordered column, a bit less well than a pure sort would for any one of them.

In Delta Lake you run OPTIMIZE table ZORDER BY (col1, col2), which rewrites files in Z-order as part of compaction. Choose columns that appear often in filters and have many distinct values (IDs, timestamps); Z-ordering a low-cardinality column or a partition column gains nothing, and effectiveness falls with each extra column. Z-ordering is not incremental: new data stays unordered until the next OPTIMIZE. Databricks now recommends liquid clustering (CLUSTER BY) for new tables, which clusters incrementally and lets you change keys without rewriting the table.

Example

200,000 rows written in random order: every file spans every value, so both filters must read all 16 files. After OPTIMIZE … ZORDER BY (customer_id, product_id), the file statistics exclude most files for either column (PySpark 4.2.0 with delta-spark 4.4.0; a small target file size makes the effect visible on a laptop).

import glob, json, os, tempfile
from pyspark.sql import SparkSession, functions as F
from delta import configure_spark_with_delta_pip
from delta.tables import DeltaTable

builder = (SparkSession.builder.master("local[2]").appName("glossary-zorder")
           .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
           .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
           .config("spark.databricks.delta.optimize.maxFileSize", 256 * 1024))  # small files, to see skipping
spark = configure_spark_with_delta_pip(builder).getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
path = os.path.join(tempfile.mkdtemp(), "sales")

(spark.range(200_000)
   .select((F.rand(1) * 10_000).cast("int").alias("customer_id"),
           (F.rand(2) * 5_000).cast("int").alias("product_id"),
           F.rand(3).alias("amount"))
   .repartition(16).write.format("delta").save(path))

def files_to_read(column, value):
    """Data skipping: count live files whose min/max range for `column` contains `value`."""
    live = {}
    for log in sorted(glob.glob(f"{path}/_delta_log/*.json")):
        for action in map(json.loads, open(log)):
            if "add" in action:
                live[action["add"]["path"]] = json.loads(action["add"]["stats"])
            if "remove" in action:
                live.pop(action["remove"]["path"], None)
    hits = sum(s["minValues"][column] <= value <= s["maxValues"][column] for s in live.values())
    return f"{hits} of {len(live)} files"

print("before: customer_id = 4242 ->", files_to_read("customer_id", 4242),
      "| product_id = 777 ->", files_to_read("product_id", 777))
DeltaTable.forPath(spark, path).optimize().executeZOrderBy("customer_id", "product_id")
print("after:  customer_id = 4242 ->", files_to_read("customer_id", 4242),
      "| product_id = 777 ->", files_to_read("product_id", 777))
spark.stop()

before: customer_id = 4242 -> 16 of 16 files | product_id = 777 -> 16 of 16 files
after:  customer_id = 4242 -> 4 of 11 files | product_id = 777 -> 6 of 11 files

Common mistakes

• Z-ordering by a partition column, which is already pruned by folder.
• Listing many columns; each one dilutes the clustering of the others.
• Expecting new appends to stay ordered; they need another OPTIMIZE.
• Using Z-ordering on new Databricks tables where liquid clustering is the recommended replacement.