Optimizing Spatial Joins with Iceberg Z-Ordering

In production lakehouse architectures, spatial join failures rarely stem from raw compute exhaustion. They originate from cross-partition shuffle skew. When joining high-cardinality vector layers (cadastral parcels, sensor footprints, road networks) against time-series telemetry or raster tilesets, standard Iceberg partitioning by ingestion timestamp or administrative region fails to localize spatial predicates. The query planner defaults to broadcast or sort-merge joins that trigger full table scans, materializing intermediate datasets that exceed executor memory limits, saturate network I/O, and breach SLAs. The engineering objective is deterministic: enforce spatial locality at the file level using multi-dimensional Z-ordering to enable aggressive predicate pruning and eliminate unnecessary shuffle.

The Partition Blindness Failure Mode

Traditional Spatial Partitioning & Indexing Strategies rely on hierarchical grids, temporal buckets, or categorical boundaries that rarely align with runtime spatial query patterns. When a join executes ST_Intersects(a.geometry, b.geometry), the optimizer cannot map bounding box overlap to physical file boundaries if partitioning is purely temporal or categorical. Iceberg’s metadata layer tracks column-level min/max statistics per data file, but geometry columns stored as raw BINARY (WKB) lack scalar bounds. Without spatial clustering on explicit bounding box columns, every join degenerates into a Cartesian product across partitions, forcing the execution engine to deserialize and evaluate geometries that fall entirely outside the target region.

Configuring Z-Order as a Sort Primitive

Z-ordering maps multi-dimensional spatial coordinates into a single scalar sort key by interleaving the binary representations of X and Y dimensions. In Iceberg, this is applied via scheduled compaction. By materializing a scalar Z-value column (or equivalent bbox columns sorted together), spatial locality is transformed into a linear sort key. The Z-Ordering for Geospatial Queries methodology ensures that spatially proximate features land in the same Parquet row groups, enabling the query planner to prune files using simple range predicates on the bounding box columns instead of evaluating complex spatial functions at runtime.

Step 1: Compute Deterministic Bounding Box Columns

Extract min/max bounding box coordinates during ingestion. This is the primary vehicle for Z-order clustering in Iceberg:

sql
-- Compute bbox columns from geometry (Spark SQL with Apache Sedona)
CREATE OR REPLACE TEMPORARY VIEW spatial_prepped AS
SELECT
  id,
  geometry,
  ST_XMin(ST_GeomFromWKB(geometry)) AS bbox_min_x,
  ST_YMin(ST_GeomFromWKB(geometry)) AS bbox_min_y,
  ST_XMax(ST_GeomFromWKB(geometry)) AS bbox_max_x,
  ST_YMax(ST_GeomFromWKB(geometry)) AS bbox_max_y
FROM raw_vector_feed;

Step 2: Register Sort Order in Iceberg Metadata

Define the bbox columns as the primary sort key. Iceberg uses this metadata to guide file layout during writes and compaction.

sql
ALTER TABLE prod.spatial_assets SET TBLPROPERTIES (
  'write.sort-order' = 'bbox_min_x ASC, bbox_min_y ASC, bbox_max_x ASC, bbox_max_y ASC'
);

For Spark with Iceberg 1.3+, you can also enforce sort order at the DataFrame level:

python
df.sortWithinPartitions("bbox_min_x", "bbox_min_y", "bbox_max_x", "bbox_max_y") \
  .writeTo("prod.spatial_assets") \
  .append()

Production Compaction & Metadata Alignment

Z-ordering degrades as data accumulates. Without scheduled compaction, file boundaries diverge from the sort key, reintroducing shuffle skew. Implement a daily compaction job that rewrites small files and re-clusters on bbox columns:

sql
CALL catalog.system.rewrite_data_files(
  table => 'prod.spatial_assets',
  strategy => 'sort',
  sort_order => 'bbox_min_x ASC, bbox_min_y ASC, bbox_max_x ASC, bbox_max_y ASC',
  options => map('min-input-files', '10', 'target-file-size-bytes', '536870912')
);

Critical Parameters:

  • target-file-size-bytes: Set to 536870912 (512MB) for S3/GCS optimal read block size.
  • spark.sql.adaptive.enabled=true: Enables Adaptive Query Execution (AQE) to dynamically coalesce skewed partitions post-shuffle.
  • spark.sql.adaptive.skewJoin.enabled=true: Splits skewed partitions into smaller tasks.
  • Verify bbox column statistics are tracked: DESCRIBE EXTENDED prod.spatial_assets should show min/max values for bbox_min_x, bbox_max_x, etc.

Debugging Predicate Pruning & Resolving Skew

Failure Mode 1: Full Table Scan Despite Sort Order

Symptom: EXPLAIN shows FileScan parquet with PartitionFilters: [] and DataFilters: [] for bbox columns. Root Cause: Missing sort order metadata or query predicate does not reference the bbox columns. Resolution:

  1. Verify sort order registration: SHOW TBLPROPERTIES prod.spatial_assets ('write.sort-order')
  2. Rewrite query to explicitly filter on bbox range before spatial evaluation:
sql
WITH pruned AS (
  SELECT * FROM prod.spatial_assets
  WHERE bbox_min_x >= -74.1 AND bbox_max_x <= -73.8
    AND bbox_min_y >= 40.6  AND bbox_max_y <= 40.9
)
SELECT * FROM pruned a
JOIN telemetry b ON ST_Intersects(
  ST_GeomFromWKB(a.geometry),
  ST_GeomFromWKB(b.footprint)
);

Failure Mode 2: Executor OOM on Join Stage

Symptom: java.lang.OutOfMemoryError: Java heap space during SortMergeJoin or BroadcastHashJoin. Root Cause: Z-order clustering is misaligned with join keys, or broadcast threshold is exceeded. Resolution:

  1. Disable broadcast for large spatial tables: spark.sql.autoBroadcastJoinThreshold=-1
  2. Increase shuffle partitions to match data skew: spark.sql.shuffle.partitions=400
  3. Enable AQE skew handling:
properties
spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.advisoryPartitionSizeInBytes=134217728
  1. Validate row group alignment using Parquet metadata inspection:
bash
parquet-tools meta s3://bucket/path/to/file.parquet | grep -A 5 "bbox_min_x"

Ensure min and max values are tightly bounded per row group. Wide ranges indicate poor clustering.

Failure Mode 3: Manifest File Bloat & Metadata Latency

Symptom: Query planning exceeds 30s; table.refresh() triggers frequent catalog calls. Root Cause: High write frequency without compaction creates thousands of small manifests. Resolution:

  • Run CALL catalog.system.expire_snapshots('prod.spatial_assets', older_than => TIMESTAMPADD(DAY, -30, CURRENT_TIMESTAMP))
  • Schedule rewrite_data_files to target max-concurrent-file-group-rewrites=5
  • Enable write.metadata.delete-after-commit.enabled=true to limit manifest accumulation

Production Checklist

  • Bbox columns (bbox_min_x, bbox_min_y, bbox_max_x, bbox_max_y) materialized as DOUBLE NOT NULL
  • write.sort-order registered in Iceberg table properties on bbox columns
  • Compaction job scheduled daily with strategy => 'sort'
  • AQE and skew join handling enabled in Spark config
  • Query predicates explicitly reference bbox range before ST_Intersects
  • Manifest count monitored; threshold alert set at >5,000 per snapshot

For authoritative Iceberg configuration references, consult the official Apache Iceberg Sort Order documentation and Spark AQE performance tuning guidelines.

Why a Join Benefits More Than a Filter

Sorting helps a filtered scan by letting the reader skip files. It helps a join by a second, larger mechanism: it makes the two sides of the join co-located, so that matching partitions of the two inputs can be paired without a full exchange.

Co-location turns a many-to-many exchange into pairs unsorted: every pair compared shuffle volume grows with the product both sorted on the same curve shuffle volume grows with the data, not the product

This is why a sorted layout produces a larger speedup on joins than on filters, and why it is worth sorting both sides of a frequently-joined pair on the same columns with the same curve. The saving is in the exchange, which is the part of a distributed join that does not parallelise away.

The caveat is that co-location only helps when the engine knows about it. Some planners detect matching sort orders and elide the shuffle; others do not and will shuffle regardless. Check the plan for an exchange operator before and after, because sorting both sides and still shuffling costs the sort with none of the benefit — in which case an explicit spatial partitioner, or a broadcast of the smaller side, is the better lever.

Sorting the Right Side of the Join

Where the sorting effort belongs small side broadcast thousands of polygons sort the large side only the small side is in memory cheapest case by far large versus large both sides at scale sort both, same columns, same curve, same resolution check the exchange is elided repeated reference join same table, many queries sort it once, permanently cost amortises immediately easiest win available

The left-hand case covers the majority of production spatial joins — telemetry against administrative boundaries, events against service areas, observations against a fixed set of regions — and it is worth checking whether a join really is large-versus-large before investing in sorting both sides. A polygon table of a few thousand rows fits comfortably in memory on every executor, and broadcasting it removes the exchange entirely, which is a bigger saving than any sort order can deliver.

Keeping the Sort Order Alive

Clustering decays between rewrites — schedule accordingly high 1.0 time → appends raise the overlap factor each rewrite returns it near 1

The sawtooth is normal and it is the reason the rewrite cadence matters more than the rewrite quality. A perfect sort run monthly leaves the table poorly clustered for most of the month; a good-enough sort run every few hours on the active partition keeps the average close to the peak. Where the format offers incremental clustering that maintains the ordering as data arrives, it flattens the curve entirely and is worth adopting for any continuously written table.

Set the cadence from the measurement rather than from a convention. A table whose overlap factor rises from 1.2 to 3 in six hours needs a six-hourly rewrite; one that takes a week to reach the same point needs a weekly one. Both numbers come from the same one-line metadata query, and running it for a fortnight before fixing the schedule produces a cadence that fits the workload instead of one inherited from a template.

Record the cadence and the observed overlap factor as table properties so the next person to look at the table inherits both the setting and the reason for it, rather than a maintenance job whose schedule appears arbitrary.

The pattern generalises to every maintenance setting on a spatial table: measure the property the setting is meant to preserve, choose the cadence from the measured decay, and store both alongside the data so the decision remains auditable.

A schedule chosen this way tends to survive staff changes, because the reasoning is recoverable from the table rather than from whoever configured it.

For the mechanics of the sort itself and how it interacts with partition granularity, see Z-ordering for geospatial queries. That page also covers the overlap-factor metric referenced throughout this guide, and the query that computes it from table metadata alone.