WIP: End-to-end spatial file pruning from Spark SQL (ST_INTERSECTS pushdown) - #17175
Draft
huan233usc wants to merge 6 commits into
Draft
WIP: End-to-end spatial file pruning from Spark SQL (ST_INTERSECTS pushdown)#17175huan233usc wants to merge 6 commits into
huan233usc wants to merge 6 commits into
Conversation
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.
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
force-pushed
the
geo-spatial-pushdown-poc
branch
from
July 13, 2026 21:36
0b8d25d to
0a579ed
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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_intersectsfunction 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:
TestSpatialFilterPushdown(spark-extensions) writes two geometry files with disjointclusters and shows:
filters=list;skipped during planning.
The full chain (each link was the tricky part)
api):Operation.ST_INTERSECTS,Unbound/BoundSpatialPredicatecarrying a query
BoundingBox(a spatial window, not a scalarLiteral),Expressions.stIntersects(...), and dispatch inExpressionVisitors.InclusiveMetricsEvaluator): reads the file's geometrylower_bounds/upper_boundsand callsGeospatialPredicateEvaluators.intersects(query, fileBbox); disjoint ⇒ROWS_CANNOT_MATCH.(
TestSpatialMetricsEvaluatorcovers this in isolation.)ExpressionVisitorsubclass —
Projections,Binder,RewriteNot,ExpressionUtilsanitizers,Spark3Utildescribe, and the three Parquet row-group filters all need aspatialPredicatecase (both Bound and Unbound overloads), or they NPE. Enumerated here.BoundingBox/GeospatialBoundmadeSerializableso a pushedpredicate ships to executors.
system.st_intersects(geom, minX, minY, maxX, maxY)registered as aScalarFunction(with the magicinvokestatic method and aCLASS_TO_FUNCTIONSentry — both required for pushdown).
ReplaceStaticInvoke, spark-extensions): a top-level booleanscalar-function filter is left by Spark as a
StaticInvoke; the rule now rewrites it toan
ApplyFunctionExpressionthat Spark can translate to a pushable DSv2 predicate.(Previously it only rewrote functions inside comparisons like
bucket(col) = 5.)SparkV2Filters): the boolean UDF arrives wrapped asV2Predicate("BOOLEAN_EXPRESSION", [UserDefinedScalarFunc]); unwrap it and build theIceberg
ST_INTERSECTSexpression.SparkParquetWriters(inaddition 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
(minX, minY, maxX, maxY)window (doubles) so thepushdown extraction is trivial; a real design would take a constant geometry.
the group); file-level pruning does the work. The row-level
invokereturns true (a PoCstub); correctness for the returned rows relies on file pruning here.
ReplaceStaticInvoke).Opening as a draft to anchor the spatial-predicate API and the pushdown design.