Skip to content

Commit a7fff26

Browse files
authored
[GH-3031] Box3D spatial join: index ST_Intersects / ST_Contains on Box3D (#3032)
1 parent ffefe5b commit a7fff26

4 files changed

Lines changed: 268 additions & 57 deletions

File tree

spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala

Lines changed: 62 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -77,14 +77,37 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy {
7777
left.dataType.isInstanceOf[Box2DUDT] && right.dataType.isInstanceOf[Box2DUDT]
7878

7979
/**
80-
* Either shape expression resolves to Box3DUDT. The join executors only know how to materialise
81-
* Geometry / Geography / Box2D values; Box3D inputs would fall into the generic shape path,
82-
* cast as a Geometry binary, and fail at runtime. Skip join planning so the predicate falls
83-
* back to row-by-row evaluation where the scalar Box3D overload of `ST_Intersects` /
84-
* `ST_Contains` handles it correctly.
80+
* Both shape expressions resolve to Box3DUDT. The join executor materialises each Box3D as the
81+
* XY footprint rectangle so the 2D partitioner / R-tree pipeline can prune candidate pairs on
82+
* X/Y overlap. The Z axis is re-checked per candidate via the scalar Box3D predicate carried in
83+
* `extraCondition`.
8584
*/
86-
private def hasBox3DInput(left: Expression, right: Expression): Boolean =
87-
left.dataType.isInstanceOf[Box3DUDT] || right.dataType.isInstanceOf[Box3DUDT]
85+
private def isBox3DPair(left: Expression, right: Expression): Boolean =
86+
left.dataType.isInstanceOf[Box3DUDT] && right.dataType.isInstanceOf[Box3DUDT]
87+
88+
/**
89+
* Construct a `JoinQueryDetection` for a Box3D-on-Box3D `ST_Intersects` / `ST_Contains`. The
90+
* R-tree pre-filter runs against the XY footprint with the supplied 2D `SpatialPredicate`; the
91+
* full 3D predicate is ANDed into `extraCondition` so `TraitJoinQueryExec`'s per-pair filter
92+
* enforces the Z axis after the index probe.
93+
*/
94+
private def box3dJoinDetection(
95+
left: LogicalPlan,
96+
right: LogicalPlan,
97+
leftShape: Expression,
98+
rightShape: Expression,
99+
spatial: Expression,
100+
extraCondition: Option[Expression],
101+
spatialPredicate: SpatialPredicate): Option[JoinQueryDetection] =
102+
Some(
103+
JoinQueryDetection(
104+
left,
105+
right,
106+
leftShape,
107+
rightShape,
108+
spatialPredicate,
109+
isGeography = false,
110+
extraCondition = Some(extraCondition.map(And(_, spatial)).getOrElse(spatial))))
88111

89112
/**
90113
* Build a JoinQueryDetection for an InferredExpression predicate (ST_Contains, ST_Within, ...)
@@ -280,16 +303,38 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy {
280303
SpatialPredicate.COVERS,
281304
isGeography = false,
282305
extraCondition))
283-
// Box3D inputs short-circuit out of join planning. The join executors only handle
284-
// Geometry / Geography / Box2D; treating a Box3D shape as a Geometry binary fails at
285-
// runtime. Returning None drops to row-by-row evaluation where the scalar Box3D
286-
// overload of ST_Intersects / ST_Contains handles the predicate correctly.
287-
case ST_Intersects(Seq(leftShape, rightShape))
288-
if hasBox3DInput(leftShape, rightShape) =>
289-
None
290-
case ST_Contains(Seq(leftShape, rightShape))
291-
if hasBox3DInput(leftShape, rightShape) =>
292-
None
306+
// Box3D-on-Box3D variants of ST_Intersects / ST_Contains. The R-tree pass uses the
307+
// XY footprint of each Box3D (materialised in TraitJoinQueryBase.shapeToGeometry),
308+
// which is a coarse-grained 2D filter. The Z axis is checked per surviving candidate
309+
// by folding the original predicate back into `extraCondition` — TraitJoinQueryExec
310+
// applies it to every joined row after the index probe, so candidate pairs that
311+
// overlap in XY but not Z are filtered out.
312+
//
313+
// R-tree predicate choice mirrors the Box2D arms (line 263 / 274):
314+
// - ST_Intersects → INTERSECTS (3D intersection ⟹ XY intersection).
315+
// - ST_Contains → COVERS (3D containment ⟹ XY covering; tighter prune than
316+
// INTERSECTS at no correctness cost since the Z
317+
// refine runs unchanged).
318+
case spatial @ ST_Intersects(Seq(leftShape, rightShape))
319+
if isBox3DPair(leftShape, rightShape) =>
320+
box3dJoinDetection(
321+
left,
322+
right,
323+
leftShape,
324+
rightShape,
325+
spatial,
326+
extraCondition,
327+
SpatialPredicate.INTERSECTS)
328+
case spatial @ ST_Contains(Seq(leftShape, rightShape))
329+
if isBox3DPair(leftShape, rightShape) =>
330+
box3dJoinDetection(
331+
left,
332+
right,
333+
leftShape,
334+
rightShape,
335+
spatial,
336+
extraCondition,
337+
SpatialPredicate.COVERS)
293338
// ST_Contains: when either operand is GeographyUDT we still detect the join here and
294339
// set `geographyShape = true`; planBroadcastJoin will route the work to the
295340
// Geography-aware index/refine path. Non-broadcast plans bail out in `apply` below

spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/TraitJoinQueryBase.scala

Lines changed: 30 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import org.apache.spark.rdd.RDD
2727
import org.apache.spark.sql.catalyst.InternalRow
2828
import org.apache.spark.sql.catalyst.expressions.{Expression, UnsafeRow}
2929
import org.apache.spark.sql.execution.SparkPlan
30-
import org.apache.spark.sql.sedona_sql.UDT.{Box2DUDT, RasterUDT}
30+
import org.apache.spark.sql.sedona_sql.UDT.{Box2DUDT, Box3DUDT, RasterUDT}
3131
import org.locationtech.jts.geom.{Geometry, GeometryFactory}
3232

3333
trait TraitJoinQueryBase {
@@ -246,8 +246,11 @@ object TraitJoinQueryBase {
246246

247247
/**
248248
* Materialise a shape column value as a JTS [[Geometry]]. Box2D-typed columns are turned into
249-
* the closed rectangular polygon implied by their `(xmin, ymin, xmax, ymax)` bounds; all other
250-
* shape columns are deserialised from the Sedona geometry binary form.
249+
* the closed rectangular polygon implied by their `(xmin, ymin, xmax, ymax)` bounds; Box3D
250+
* columns are projected to their XY footprint (`xmin, ymin, xmax, ymax`) and the Z axis is
251+
* re-checked per candidate pair by the scalar Box3D predicate that `JoinQueryDetector` folds
252+
* into the join's `extraCondition`. All other shape columns are deserialised from the Sedona
253+
* geometry binary form.
251254
*
252255
* Producing a JTS rectangle here lets the rest of the join machinery — partitioner, R-tree
253256
* `IndexBuilder`, refine evaluator — stay shape-agnostic. JTS already short-circuits
@@ -259,7 +262,8 @@ object TraitJoinQueryBase {
259262
* `IllegalArgumentException` raised by `Predicates.boxIntersects` / `boxContains`. Inverted
260263
* bounds have no defined planar meaning today (they are reserved for future
261264
* antimeridian-wraparound semantics on Geography bboxes) and would silently mis-prune the
262-
* R-tree if accepted here.
265+
* R-tree if accepted here. Box3D applies the same check on all three axes — Z has no wraparound
266+
* convention.
263267
*
264268
* Returns `null` when the shape column evaluates to NULL; the caller is expected to either skip
265269
* the row or substitute an empty geometry.
@@ -283,6 +287,28 @@ object TraitJoinQueryBase {
283287
"reserved for future antimeridian wraparound semantics.")
284288
}
285289
Constructors.polygonFromEnvelope(xmin, ymin, xmax, ymax)
290+
case _: Box3DUDT =>
291+
val box = evaluated.asInstanceOf[InternalRow]
292+
val xmin = box.getDouble(0)
293+
val ymin = box.getDouble(1)
294+
val zmin = box.getDouble(2)
295+
val xmax = box.getDouble(3)
296+
val ymax = box.getDouble(4)
297+
val zmax = box.getDouble(5)
298+
// Eager validation: every row in the dataset is inspected during index build, so any
299+
// inverted-bound row throws here even if it would not have matched anything. This is
300+
// a slightly broader failure surface than the prior row-by-row fallback (which only
301+
// threw on rows the join actually evaluated), but it matches the Box2D contract on
302+
// line above and the scalar Box3D predicate contract — those reject inverted bounds
303+
// outright on the grounds that Z has no wraparound convention.
304+
if (xmin > xmax || ymin > ymax || zmin > zmax) {
305+
throw new IllegalArgumentException(
306+
"Box3D join input has inverted bounds (xmin > xmax, ymin > ymax, or " +
307+
"zmin > zmax). Box3D predicates require ordered intervals on all three axes.")
308+
}
309+
// XY footprint only — the Z axis is rechecked per candidate by the scalar Box3D
310+
// predicate folded into `extraCondition`.
311+
Constructors.polygonFromEnvelope(xmin, ymin, xmax, ymax)
286312
case _ =>
287313
GeometrySerializer.deserialize(evaluated.asInstanceOf[Array[Byte]])
288314
}

spark/common/src/test/scala/org/apache/sedona/sql/Box3DIntersectsContainsSuite.scala

Lines changed: 0 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -103,41 +103,5 @@ class Box3DIntersectsContainsSuite extends TestBaseScala {
103103
assert(row.isNullAt(1))
104104
}
105105

106-
// Regression for the JoinQueryDetector / TraitJoinQueryBase Box3D mishandling: prior to the
107-
// hasBox3DInput guard, a Box3D-on-Box3D join would fall into the generic shape path and try
108-
// to deserialise the Box3D struct as a Geometry binary, throwing at runtime. With the guard
109-
// in place, the join silently falls back to row-by-row evaluation and the scalar Box3D
110-
// overload of ST_Intersects / ST_Contains produces the correct boolean.
111-
it("Box3D-on-Box3D join falls back to row-by-row evaluation without a cast error") {
112-
import sparkSession.implicits._
113-
val left = Seq(("a1", 0.0, 0.0, 0.0, 5.0, 5.0, 5.0)).toDF(
114-
"id",
115-
"xmin",
116-
"ymin",
117-
"zmin",
118-
"xmax",
119-
"ymax",
120-
"zmax")
121-
val right = Seq(("b1", 1.0, 1.0, 1.0, 2.0, 2.0, 2.0)).toDF(
122-
"id",
123-
"xmin",
124-
"ymin",
125-
"zmin",
126-
"xmax",
127-
"ymax",
128-
"zmax")
129-
left.createOrReplaceTempView("left_box3d")
130-
right.createOrReplaceTempView("right_box3d")
131-
val df = sparkSession.sql("""
132-
SELECT L.id AS l_id, R.id AS r_id
133-
FROM left_box3d L
134-
JOIN right_box3d R ON ST_Intersects(
135-
ST_3DMakeBox(ST_PointZ(L.xmin, L.ymin, L.zmin), ST_PointZ(L.xmax, L.ymax, L.zmax)),
136-
ST_3DMakeBox(ST_PointZ(R.xmin, R.ymin, R.zmin), ST_PointZ(R.xmax, R.ymax, R.zmax)))
137-
""")
138-
val rows = df.collect()
139-
assert(rows.length == 1)
140-
assert(rows(0).getString(0) == "a1" && rows(0).getString(1) == "b1")
141-
}
142106
}
143107
}
Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.sedona.sql
20+
21+
import org.apache.spark.sql.DataFrame
22+
import org.apache.spark.sql.functions.{broadcast, expr}
23+
import org.apache.spark.sql.sedona_sql.strategy.join.{BroadcastIndexJoinExec, RangeJoinExec}
24+
25+
class Box3DJoinSuite extends TestBaseScala {
26+
27+
import Box3DJoinSuite.TestBox3D
28+
29+
/**
30+
* Left and right boxes wired so result counts are predictable across the XY filter and the
31+
* Z-axis refine step. L4 is the discriminating row: same XY footprint as L1/L2 so the 2D R-tree
32+
* pairs it with every right in the (0..10) XY cluster, but its Z range is disjoint from every
33+
* right's Z (above R1/R2's Z, below R4's Z) so the refine must reject all those candidates.
34+
*
35+
* - L1=(0..10, 0..10, 0..10), L2=same — dense cluster on Z=(0..10).
36+
* - L3=(20..30, 20..30, 0..10) — XY-disjoint outlier.
37+
* - L4=(0..10, 0..10, 50..60) — XY in the cluster, Z disjoint from every right side. Every
38+
* L4-R* candidate the R-tree emits has to be killed by the Z refine.
39+
*
40+
* - R1=(5..15, 5..15, 5..15), R2=(2..8, 2..8, 2..8) — overlap L1/L2 in XY *and* Z.
41+
* - R3=(40..50, 40..50, 0..10) — XY-disjoint.
42+
* - R4=(2..8, 2..8, 100..200) — XY in the cluster, Z separated from every left.
43+
*
44+
* True intersection pairs: 4 — (L1,R1), (L1,R2), (L2,R1), (L2,R2). Containment pairs
45+
* (closed-interval, all three axes): 2 — (L1,R2), (L2,R2). R1 extends past L1/L2 on all three
46+
* axes; R4 and L4 are Z-disjoint from anything they could contain or be contained by.
47+
*/
48+
private def leftBoxes: DataFrame = {
49+
import sparkSession.implicits._
50+
Seq(
51+
TestBox3D(1, 0, 0, 0, 10, 10, 10),
52+
TestBox3D(2, 0, 0, 0, 10, 10, 10),
53+
TestBox3D(3, 20, 20, 0, 30, 30, 10),
54+
TestBox3D(4, 0, 0, 50, 10, 10, 60))
55+
.toDF("id", "xmin", "ymin", "zmin", "xmax", "ymax", "zmax")
56+
.selectExpr(
57+
"id",
58+
"ST_3DMakeBox(ST_PointZ(xmin, ymin, zmin), ST_PointZ(xmax, ymax, zmax)) AS box")
59+
}
60+
61+
private def rightBoxes: DataFrame = {
62+
import sparkSession.implicits._
63+
Seq(
64+
TestBox3D(11, 5, 5, 5, 15, 15, 15),
65+
TestBox3D(12, 2, 2, 2, 8, 8, 8),
66+
TestBox3D(13, 40, 40, 0, 50, 50, 10),
67+
TestBox3D(14, 2, 2, 100, 8, 8, 200))
68+
.toDF("id", "xmin", "ymin", "zmin", "xmax", "ymax", "zmax")
69+
.selectExpr(
70+
"id",
71+
"ST_3DMakeBox(ST_PointZ(xmin, ymin, zmin), ST_PointZ(xmax, ymax, zmax)) AS box")
72+
}
73+
74+
describe("Box3D spatial join") {
75+
76+
it("ST_Intersects: broadcast index join filters Z-disjoint candidates") {
77+
val joined = leftBoxes
78+
.alias("L")
79+
.join(broadcast(rightBoxes.alias("R")), expr("ST_Intersects(L.box, R.box)"))
80+
81+
assert(joined.queryExecution.executedPlan.collectFirst { case _: BroadcastIndexJoinExec =>
82+
true
83+
}.isDefined)
84+
// 4 true intersections. The L4 row produces three XY-overlapping candidates (L4-R1,
85+
// L4-R2, L4-R4) that the 2D R-tree emits and the Z refine rejects.
86+
assert(joined.count() == 4)
87+
}
88+
89+
it("ST_Intersects: argument order is symmetric") {
90+
val joined = leftBoxes
91+
.alias("L")
92+
.join(broadcast(rightBoxes.alias("R")), expr("ST_Intersects(R.box, L.box)"))
93+
assert(joined.count() == 4)
94+
}
95+
96+
it("ST_Contains: broadcast index join uses Box3D containment semantics") {
97+
val joined = leftBoxes
98+
.alias("L")
99+
.join(broadcast(rightBoxes.alias("R")), expr("ST_Contains(L.box, R.box)"))
100+
101+
assert(joined.queryExecution.executedPlan.collectFirst { case _: BroadcastIndexJoinExec =>
102+
true
103+
}.isDefined)
104+
// Only R2 is fully inside L1 / L2 on all three axes. R1 sticks out in X/Y/Z; R4 is
105+
// disjoint in Z.
106+
assert(joined.count() == 2)
107+
}
108+
109+
it("ST_Contains: edge-touching boxes count (closed-interval semantics)") {
110+
import sparkSession.implicits._
111+
val outer = Seq(TestBox3D(1, 0, 0, 0, 10, 10, 10))
112+
.toDF("id", "xmin", "ymin", "zmin", "xmax", "ymax", "zmax")
113+
.selectExpr(
114+
"id",
115+
"ST_3DMakeBox(ST_PointZ(xmin, ymin, zmin), ST_PointZ(xmax, ymax, zmax)) AS box")
116+
// Inner shares all three faces with the outer on the high side.
117+
val inner = Seq(TestBox3D(2, 5, 5, 5, 10, 10, 10))
118+
.toDF("id", "xmin", "ymin", "zmin", "xmax", "ymax", "zmax")
119+
.selectExpr(
120+
"id",
121+
"ST_3DMakeBox(ST_PointZ(xmin, ymin, zmin), ST_PointZ(xmax, ymax, zmax)) AS box")
122+
val joined = outer
123+
.alias("O")
124+
.join(broadcast(inner.alias("I")), expr("ST_Contains(O.box, I.box)"))
125+
assert(joined.count() == 1)
126+
}
127+
128+
it("ST_Intersects: non-broadcast range join produces the same count") {
129+
val joined = leftBoxes
130+
.alias("L")
131+
.join(rightBoxes.alias("R"), expr("ST_Intersects(L.box, R.box)"))
132+
133+
assert(joined.queryExecution.executedPlan.collectFirst { case _: RangeJoinExec =>
134+
true
135+
}.isDefined)
136+
assert(joined.count() == 4)
137+
}
138+
139+
it("Inverted Box3D bounds in a join throw IllegalArgumentException") {
140+
import sparkSession.implicits._
141+
val ordered = Seq(TestBox3D(1, 0, 0, 0, 10, 10, 10))
142+
.toDF("id", "xmin", "ymin", "zmin", "xmax", "ymax", "zmax")
143+
.selectExpr(
144+
"id",
145+
"ST_3DMakeBox(ST_PointZ(xmin, ymin, zmin), ST_PointZ(xmax, ymax, zmax)) AS box")
146+
// Z inverted on the right side.
147+
val invertedZ = Seq(TestBox3D(2, 0, 0, 10, 10, 10, 0))
148+
.toDF("id", "xmin", "ymin", "zmin", "xmax", "ymax", "zmax")
149+
.selectExpr(
150+
"id",
151+
"ST_3DMakeBox(ST_PointZ(xmin, ymin, zmin), ST_PointZ(xmax, ymax, zmax)) AS box")
152+
val ex = intercept[Exception] {
153+
ordered
154+
.alias("L")
155+
.join(broadcast(invertedZ.alias("R")), expr("ST_Intersects(L.box, R.box)"))
156+
.collect()
157+
}
158+
assert(
159+
Iterator
160+
.iterate(ex: Throwable)(_.getCause)
161+
.takeWhile(_ != null)
162+
.exists(_.isInstanceOf[IllegalArgumentException]))
163+
}
164+
}
165+
}
166+
167+
object Box3DJoinSuite {
168+
case class TestBox3D(
169+
id: Int,
170+
xmin: Double,
171+
ymin: Double,
172+
zmin: Double,
173+
xmax: Double,
174+
ymax: Double,
175+
zmax: Double)
176+
}

0 commit comments

Comments
 (0)