Skip to content

WIP: End-to-end spatial file pruning from Spark SQL (ST_INTERSECTS pushdown) - #17175

Draft
huan233usc wants to merge 6 commits into
apache:mainfrom
huan233usc:geo-spatial-pushdown-poc
Draft

WIP: End-to-end spatial file pruning from Spark SQL (ST_INTERSECTS pushdown)#17175
huan233usc wants to merge 6 commits into
apache:mainfrom
huan233usc:geo-spatial-pushdown-poc

Conversation

@huan233usc

@huan233usc huan233usc commented Jul 12, 2026

Copy link
Copy Markdown
Contributor

Important

Draft / proof-of-concept, stacked on #17161 (geometry bounding-box metrics — the first
commit here is that PR; review it there). This branch demonstrates end-to-end spatial
file pruning from a Spark SQL query
. API shapes (predicate, operation, the
st_intersects function signature) are a sketch for discussion, not a final proposal.
Not intended to merge as-is.

What it proves

A Spark SQL query with a spatial filter now prunes files by their geometry bounding box:

SELECT id FROM t WHERE system.st_intersects(geom, minX, minY, maxX, maxY)

TestSpatialFilterPushdown (spark-extensions) writes two geometry files with disjoint
clusters and shows:

  • the plan carries the predicate in the Iceberg scan's pushed filters= list;
  • a query window over one cluster returns only that cluster's row — the disjoint file is
    skipped during planning.

The full chain (each link was the tricky part)

  1. Expression API (api): Operation.ST_INTERSECTS, Unbound/BoundSpatialPredicate
    carrying a query BoundingBox (a spatial window, not a scalar Literal),
    Expressions.stIntersects(...), and dispatch in ExpressionVisitors.
  2. File pruning (InclusiveMetricsEvaluator): reads the file's geometry
    lower_bounds/upper_bounds and calls
    GeospatialPredicateEvaluators.intersects(query, fileBbox); disjoint ⇒ ROWS_CANNOT_MATCH.
    (TestSpatialMetricsEvaluator covers this in isolation.)
  3. Every visitor: a new predicate kind ripples through every ExpressionVisitor
    subclass — Projections, Binder, RewriteNot, ExpressionUtil sanitizers,
    Spark3Util describe, and the three Parquet row-group filters all need a
    spatialPredicate case (both Bound and Unbound overloads), or they NPE. Enumerated here.
  4. Serialization: BoundingBox/GeospatialBound made Serializable so a pushed
    predicate ships to executors.
  5. Spark function: system.st_intersects(geom, minX, minY, maxX, maxY) registered as a
    ScalarFunction (with the magic invoke static method and a CLASS_TO_FUNCTIONS
    entry — both required for pushdown).
  6. Optimizer rewrite (ReplaceStaticInvoke, spark-extensions): a top-level boolean
    scalar-function filter is left by Spark as a StaticInvoke; the rule now rewrites it to
    an ApplyFunctionExpression that Spark can translate to a pushable DSv2 predicate.
    (Previously it only rewrote functions inside comparisons like bucket(col) = 5.)
  7. Pushdown conversion (SparkV2Filters): the boolean UDF arrives wrapped as
    V2Predicate("BOOLEAN_EXPRESSION", [UserDefinedScalarFunc]); unwrap it and build the
    Iceberg ST_INTERSECTS expression.
  8. Spark writer: wired the geometry bbox metrics into SparkParquetWriters (in
    addition to the generic-data path from Parquet: Compute geometry bounding box metrics #17161), so Spark-written geo files carry a bbox
    to prune against.

Scope / caveats

  • Function signature takes an explicit (minX, minY, maxX, maxY) window (doubles) so the
    pushdown extraction is trivial; a real design would take a constant geometry.
  • Row-group-level spatial pruning is not implemented (row-group filters conservatively keep
    the group); file-level pruning does the work. The row-level invoke returns true (a PoC
    stub); correctness for the returned rows relies on file pruning here.
  • Requires the Iceberg Spark extensions (for ReplaceStaticInvoke).

Opening as a draft to anchor the spatial-predicate API and the pushdown design.

The Parquet footer's lexicographic min/max over WKB bytes is not meaningful for
geometry, so geometry columns previously wrote counts only and could not be data
skipped. Scan each value's coordinates as it is written and accumulate a 2D (XY)
bounding box, emitting it as a FieldMetrics through the writer-side metrics
channel (the same path float and double use for NaN counts). The box's lower and
upper corners serialize into the existing lower_bounds and upper_bounds maps via
the geometry conversion.

A new pure-Java WKB coordinate scanner (WKBBoundingBox, in api/geospatial, no JTS
dependency) walks all OGC geometry types, skips Z and M, ignores NaN coordinates,
and validates the buffer defensively. Geography, higher dimensions, and the
content_stats geo struct bounds are left as follow-ups.
@huan233usc huan233usc changed the title WIP: ST_INTERSECTS spatial predicate for metrics-based file pruning WIP: End-to-end spatial file pruning (ST_INTERSECTS predicate + Spark function) Jul 13, 2026
ebyhr and others added 5 commits July 13, 2026 14:27
Add a spatial predicate to the Expression API and wire it into InclusiveMetricsEvaluator
so file-level pruning can use geometry/geography bounds.

 - Expression.Operation.ST_INTERSECTS
 - UnboundSpatialPredicate / BoundSpatialPredicate carrying a query BoundingBox (the
   constant is a spatial window, not a scalar Literal, so it is its own predicate kind)
 - Expressions.stIntersects(name, BoundingBox) factory
 - ExpressionVisitors: spatialPredicate() hooks + dispatch in visit()/visitEvaluator();
   Binder/RewriteNot handle the new predicate
 - InclusiveMetricsEvaluator reads the file's geo lower/upper bounds and calls
   GeospatialPredicateEvaluators.intersects(query, fileBbox); disjoint => ROWS_CANNOT_MATCH

TestSpatialMetricsEvaluator proves a bound ST_INTERSECTS expression prunes a disjoint
file and keeps an overlapping one. This is a WIP proof of concept; the Spark pushdown
side (register iceberg.st_intersects, SparkV2Filters) is the next commit.
Add the Spark side of spatial pushdown:
 - STIntersectsFunction: a registered system function st_intersects(geom, minX, minY,
   maxX, maxY) -> boolean, so a spatial filter can be written in SQL.
 - SparkV2Filters.convertSpatial: converts that UDF predicate into the Iceberg
   ST_INTERSECTS expression (extracting the geo column ref and the four window doubles).

Finding recorded by TestSparkGeospatialPushdown: Spark 4.1 leaves a bare boolean function
used as a filter (WHERE st_intersects(...)) as an ApplyFunctionExpression in a post-scan
Filter node and does NOT hand it to DSv2 filter pushdown, so SparkV2Filters.convert is
never reached for it. Iceberg's ReplaceStaticInvoke rule only rewrites a function call
inside a comparison/IN, not a standalone boolean function. Driving this from SQL needs the
st_intersects(...) = true comparison form plus the Iceberg extensions, or an extension of
that rule. The conversion path itself is exercised by the core TestSpatialMetricsEvaluator.
Handle BoundSpatialPredicate/UnboundSpatialPredicate in every ExpressionVisitor the
scan/pushdown path exercises (Projections, ExpressionUtil sanitizers, Spark3Util
describe, Parquet row-group filters), make BoundingBox/GeospatialBound Serializable so
a pushed spatial predicate ships to executors, and extend ReplaceStaticInvoke to rewrite
a top-level boolean scalar-function predicate.

Finding: adding a new predicate KIND ripples through every ExpressionVisitor subclass —
the base default must be handled or each visitor NPEs. Enumerated the full set here.
Wire the geometry bounding-box metrics into the Spark Parquet writer
(SparkParquetWriters.GeometryWriter) so Spark-written geo files carry a bbox, and add
an end-to-end test (spark-extensions) proving a SQL spatial filter prunes files.

TestSpatialFilterPushdown: with the Iceberg extensions enabled, a query
  WHERE system.st_intersects(geom, minX, minY, maxX, maxY)
is rewritten by ReplaceStaticInvoke into a pushable ApplyFunctionExpression, translated
to an Iceberg ST_INTERSECTS expression by SparkV2Filters, and used by
InclusiveMetricsEvaluator to skip files whose geometry bbox is disjoint from the window.
The plan shows the predicate in the scan's pushed filters, and the query returns only the
rows from the intersecting file (the disjoint file is pruned).
@huan233usc
huan233usc force-pushed the geo-spatial-pushdown-poc branch from 0b8d25d to 0a579ed Compare July 13, 2026 21:36
@huan233usc huan233usc changed the title WIP: End-to-end spatial file pruning (ST_INTERSECTS predicate + Spark function) WIP: End-to-end spatial file pruning from Spark SQL (ST_INTERSECTS pushdown) Jul 13, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants