From 44bb618490263fd1d859f253055959f9294c48b4 Mon Sep 17 00:00:00 2001 From: Burak KALAYCI Date: Thu, 24 Sep 2026 19:44:34 +0300 Subject: [PATCH 1/4] [GH-3398] Keep extra predicates on broadcast and nested KNN joins Broadcast KNN plans dropped every extra join predicate, and a nested AND hid the non-KNN conjuncts from the one-level extractor. Flatten the conjuncts and apply them on both broadcast sides. --- .../strategy/join/JoinQueryDetector.scala | 25 +++----- .../org/apache/sedona/sql/KnnJoinSuite.scala | 62 +++++++++++++++++++ 2 files changed, 71 insertions(+), 16 deletions(-) diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala index 68aad366b8b..dc9834cecd2 100644 --- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala +++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala @@ -904,21 +904,13 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { } private def extractExtraKNNJoinCondition(condition: Expression): Option[Expression] = { - condition match { - case and: And => - // Check both left and right sides for ST_KNN or ST_AKNN - if (and.left.isInstanceOf[ST_KNN]) { - Some(and.right) - } else if (and.right.isInstanceOf[ST_KNN]) { - Some(and.left) - } else { - None - } - case _: ST_KNN => - None - case _ => - Some(condition) + // A nested AND such as (ST_KNN AND p1) AND p2 does not have ST_KNN as a direct child. + // Flatten the conjuncts so every non-KNN predicate is kept. + val others = splitConjunctivePredicates(condition).filter { + case _: ST_KNN => false + case _ => true } + others.reduceOption(And(_, _)) } private def planBroadcastJoin( @@ -966,6 +958,7 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { case None => Nil } + val knnExtraCondition = extraCondition.flatMap(extractExtraKNNJoinCondition) if (querySide == broadcastSide.get) { // broadcast is on query side return BroadcastQuerySideKNNJoinExec( @@ -980,7 +973,7 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { spatialPredicate, isGeography, condition = null, - extraCondition = None) :: Nil + extraCondition = knnExtraCondition) :: Nil } else { // broadcast is on object side return BroadcastObjectSideKNNJoinExec( @@ -995,7 +988,7 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { spatialPredicate, isGeography, condition = null, - extraCondition = None) :: Nil + extraCondition = knnExtraCondition) :: Nil } } } diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala b/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala index 4ca17598d30..cd20cf402da 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala @@ -486,6 +486,68 @@ class KnnJoinSuite extends TestBaseScala with TableDrivenPropertyChecks { assert(knnJoined.count() > 0) } + + it("keeps non-KNN predicates on broadcast and nested AND plans") { + withConf( + Map( + "spark.sql.adaptive.enabled" -> "false", + "spark.sql.autoBroadcastJoinThreshold" -> "-1", + "spark.sedona.join.autoBroadcastJoinThreshold" -> "-1", + "spark.sedona.join.knn.includeTieBreakers" -> "false")) { + import sparkSession.implicits._ + val queries = Seq((1, 0.0, 0.0, 10), (2, 10.0, 10.0, 10)) + .toDF("id", "x", "y", "score") + .selectExpr("id", "ST_Point(x, y) AS g", "score") + val objects = Seq((101, 1.0, 1.0, 20), (102, 11.0, 11.0, 20)) + .toDF("id", "x", "y", "score") + .selectExpr("id", "ST_Point(x, y) AS g", "score") + queries.createOrReplaceTempView("knn_pred_queries") + objects.createOrReplaceTempView("knn_pred_objects") + + def pairs(sql: String): Seq[(Int, Int)] = + sparkSession + .sql(sql) + .collect() + .map(row => (row.getInt(0), row.getInt(1))) + .sorted + .toSeq + + val neighbors = + pairs( + "SELECT q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + + "ON ST_KNN(q.g, o.g, 1, false)") + neighbors should be(Seq((1, 101), (2, 102))) + + val filtered = + "SELECT q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + + "ON ST_KNN(q.g, o.g, 1, false) AND q.score > o.score" + pairs(filtered) should be(Seq.empty) + + val broadcastQueries = sparkSession.sql( + "SELECT /*+ BROADCAST(q) */ q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + + "ON ST_KNN(q.g, o.g, 1, false) AND q.score > o.score") + broadcastQueries.queryExecution.executedPlan.toString should include( + "BroadcastQuerySideKNNJoin") + broadcastQueries + .collect() + .map(row => (row.getInt(0), row.getInt(1))) + .sorted + .toSeq should be(Seq.empty) + + val broadcastObjects = sparkSession.sql( + "SELECT /*+ BROADCAST(o) */ q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + + "ON ST_KNN(q.g, o.g, 1, false) AND q.score > o.score") + broadcastObjects.queryExecution.executedPlan.toString should include( + "BroadcastObjectSideKNNJoin") + broadcastObjects + .collect() + .map(row => (row.getInt(0), row.getInt(1))) + .sorted + .toSeq should be(Seq.empty) + + pairs(filtered + " AND q.id < o.id") should be(Seq.empty) + } + } } private def withOptimizationMode(mode: String)(body: => Unit): Unit = { From 6f53cc9f74dd46a528df4e3a6137078c72619319 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Thu, 24 Sep 2026 21:36:49 -0700 Subject: [PATCH 2/4] [GH-3398] Reuse KNN join residuals and reject additional KNN predicates Use the residual condition already returned by OptimizableJoinCondition for regular and both broadcast KNN plans. Remove duplicate extraction that discarded every KNN marker, and reject additional markers before execution instead of silently ignoring their semantics. Replace all-false-only regression coverage with independently selective conditions, nullable values, both relation orders and broadcast sides, three- and four-argument KNN, and planning-time rejection cases. Validation: 337 join tests pass on Spark 3.5; 62 KNN tests pass on Spark 4.1.1/Scala 2.13. The nine new rejection tests failed on the original PR. --- .../strategy/join/JoinQueryDetector.scala | 26 ++- .../org/apache/sedona/sql/KnnJoinSuite.scala | 159 +++++++++++------- 2 files changed, 110 insertions(+), 75 deletions(-) diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala index dc9834cecd2..2a392142f58 100644 --- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala +++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala @@ -618,6 +618,11 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { Some(distance))) // ST_KNN + case _: ST_KNN if extraCondition.exists(_.exists(_.isInstanceOf[ST_KNN])) => + // The physical join implements only the selected KNN predicate. + throw new UnsupportedOperationException( + "Only one ST_KNN predicate is supported per join condition") + case ST_KNN(Seq(leftShape, rightShape, k)) => Some( JoinQueryDetection( @@ -627,7 +632,7 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { rightShape, spatialPredicate = SpatialPredicate.KNN, isGeography = false, - condition, + extraCondition, Some(k))) case ST_KNN(Seq(leftShape, rightShape, k, useSpheroid)) => @@ -640,7 +645,7 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { rightShape, spatialPredicate = SpatialPredicate.KNN, isGeography = useSpheroidUnwrapped, - condition, + extraCondition, Some(k))) case _ => None @@ -841,7 +846,7 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { spatialPredicate = null, isGeography, condition, - extractExtraKNNJoinCondition(condition)) :: Nil + extraCondition) :: Nil } private def planDistanceJoin( @@ -903,16 +908,6 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { } } - private def extractExtraKNNJoinCondition(condition: Expression): Option[Expression] = { - // A nested AND such as (ST_KNN AND p1) AND p2 does not have ST_KNN as a direct child. - // Flatten the conjuncts so every non-KNN predicate is kept. - val others = splitConjunctivePredicates(condition).filter { - case _: ST_KNN => false - case _ => true - } - others.reduceOption(And(_, _)) - } - private def planBroadcastJoin( left: LogicalPlan, right: LogicalPlan, @@ -958,7 +953,6 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { case None => Nil } - val knnExtraCondition = extraCondition.flatMap(extractExtraKNNJoinCondition) if (querySide == broadcastSide.get) { // broadcast is on query side return BroadcastQuerySideKNNJoinExec( @@ -973,7 +967,7 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { spatialPredicate, isGeography, condition = null, - extraCondition = knnExtraCondition) :: Nil + extraCondition = extraCondition) :: Nil } else { // broadcast is on object side return BroadcastObjectSideKNNJoinExec( @@ -988,7 +982,7 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { spatialPredicate, isGeography, condition = null, - extraCondition = knnExtraCondition) :: Nil + extraCondition = extraCondition) :: Nil } } } diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala b/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala index cd20cf402da..7ec2aace331 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala @@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.expressions.Literal import org.apache.spark.sql.functions.col import org.apache.spark.sql.sedona_sql.UDT.GeometryUDT import org.apache.spark.sql.sedona_sql.expressions.st_constructors.ST_GeomFromText -import org.apache.spark.sql.sedona_sql.strategy.join.KNNJoinExec +import org.apache.spark.sql.sedona_sql.strategy.join.{BroadcastObjectSideKNNJoinExec, BroadcastQuerySideKNNJoinExec, KNNJoinExec} import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType} import org.apache.spark.sql.{Column, DataFrame, Row, SparkSession} import org.apache.spark.sql.functions.expr @@ -487,65 +487,106 @@ class KnnJoinSuite extends TestBaseScala with TableDrivenPropertyChecks { assert(knnJoined.count() > 0) } - it("keeps non-KNN predicates on broadcast and nested AND plans") { - withConf( - Map( - "spark.sql.adaptive.enabled" -> "false", - "spark.sql.autoBroadcastJoinThreshold" -> "-1", - "spark.sedona.join.autoBroadcastJoinThreshold" -> "-1", - "spark.sedona.join.knn.includeTieBreakers" -> "false")) { - import sparkSession.implicits._ - val queries = Seq((1, 0.0, 0.0, 10), (2, 10.0, 10.0, 10)) - .toDF("id", "x", "y", "score") - .selectExpr("id", "ST_Point(x, y) AS g", "score") - val objects = Seq((101, 1.0, 1.0, 20), (102, 11.0, 11.0, 20)) - .toDF("id", "x", "y", "score") - .selectExpr("id", "ST_Point(x, y) AS g", "score") - queries.createOrReplaceTempView("knn_pred_queries") - objects.createOrReplaceTempView("knn_pred_objects") - - def pairs(sql: String): Seq[(Int, Int)] = - sparkSession - .sql(sql) - .collect() - .map(row => (row.getInt(0), row.getInt(1))) - .sorted - .toSeq - - val neighbors = - pairs( - "SELECT q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + - "ON ST_KNN(q.g, o.g, 1, false)") - neighbors should be(Seq((1, 101), (2, 102))) - - val filtered = - "SELECT q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + - "ON ST_KNN(q.g, o.g, 1, false) AND q.score > o.score" - pairs(filtered) should be(Seq.empty) - - val broadcastQueries = sparkSession.sql( - "SELECT /*+ BROADCAST(q) */ q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + - "ON ST_KNN(q.g, o.g, 1, false) AND q.score > o.score") - broadcastQueries.queryExecution.executedPlan.toString should include( - "BroadcastQuerySideKNNJoin") - broadcastQueries - .collect() - .map(row => (row.getInt(0), row.getInt(1))) - .sorted - .toSeq should be(Seq.empty) - - val broadcastObjects = sparkSession.sql( - "SELECT /*+ BROADCAST(o) */ q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + - "ON ST_KNN(q.g, o.g, 1, false) AND q.score > o.score") - broadcastObjects.queryExecution.executedPlan.toString should include( - "BroadcastObjectSideKNNJoin") - broadcastObjects - .collect() - .map(row => (row.getInt(0), row.getInt(1))) - .sorted - .toSeq should be(Seq.empty) + val residualJoinStrategies: Seq[(String, String, Class[_])] = Seq( + ("regular", "", classOf[KNNJoinExec]), + ("broadcast query", "/*+ BROADCAST(q) */", classOf[BroadcastQuerySideKNNJoinExec]), + ("broadcast object", "/*+ BROADCAST(o) */", classOf[BroadcastObjectSideKNNJoinExec])) + val knnPredicate = "ST_KNN(q.g, o.g, 1, false)" + val scorePredicate = "q.score > o.score" + val ceilingPredicate = "q.ceiling > o.floor" + val residualConditions = Seq( + ( + "single residual with default distance metric", + s"ST_KNN(q.g, o.g, 1) AND $scorePredicate", + Seq((1, 101), (4, 104))), + ("KNN first", s"($knnPredicate AND $scorePredicate) AND $ceilingPredicate", Seq((1, 101))), + ("KNN middle", s"$scorePredicate AND ($knnPredicate AND $ceilingPredicate)", Seq((1, 101))), + ("KNN last", s"($scorePredicate AND $ceilingPredicate) AND $knnPredicate", Seq((1, 101)))) + + for { + (strategy, hint, expectedPlan) <- residualJoinStrategies + queriesFirst <- Seq(true, false) + (conditionName, condition, expected) <- residualConditions + } { + it(s"KNN residuals: $strategy, queriesFirst=$queriesFirst, $conditionName") { + withKnnResidualInputs { + val relations = if (queriesFirst) { + "knn_pred_queries q JOIN knn_pred_objects o" + } else { + "knn_pred_objects o JOIN knn_pred_queries q" + } + val joined = sparkSession.sql(s"SELECT $hint q.id, o.id FROM $relations ON $condition") + val plan = joined.queryExecution.executedPlan + assert(plan.find(_.getClass == expectedPlan).isDefined, plan.toString) + val pairs = joined.collect().map(row => (row.getInt(0), row.getInt(1))).sorted.toSeq + pairs should be(expected) + } + } + } + + val additionalKnnPredicates = Seq( + ("different k", s"ST_KNN(q.g, o.g, 2, false) AND $knnPredicate"), + ("reciprocal", s"$knnPredicate AND ST_KNN(o.g, q.g, 1, false)"), + ("nested in OR", s"$knnPredicate AND (ST_KNN(q.g, o.g, 2, false) OR $scorePredicate)")) + for { + (strategy, hint, _) <- residualJoinStrategies + (conditionName, condition) <- additionalKnnPredicates + } { + it(s"KNN residuals reject multiple predicates during planning: $strategy, $conditionName") { + withKnnResidualInputs { + val exception = intercept[UnsupportedOperationException] { + sparkSession + .sql(s"SELECT $hint q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + + s"ON $condition") + .queryExecution + .executedPlan + } + exception.getMessage should be( + "Only one ST_KNN predicate is supported per join condition") + } + } + } + } - pairs(filtered + " AND q.id < o.id") should be(Seq.empty) + private def withKnnResidualInputs(body: => Unit): Unit = { + withConf( + Map( + "spark.sql.adaptive.enabled" -> "false", + "spark.sql.autoBroadcastJoinThreshold" -> "-1", + "spark.sedona.join.autoBroadcastJoinThreshold" -> "-1", + "spark.sedona.join.knn.includeTieBreakers" -> "false")) { + try { + sparkSession + .sql(""" + |SELECT id, ST_Point(x, x) AS g, score, ceiling + |FROM VALUES + | (1, 0.0D, 30, 100), + | (2, 10.0D, 10, 100), + | (3, 20.0D, CAST(NULL AS INT), 100), + | (4, 30.0D, 30, 0) + |AS q(id, x, score, ceiling) + |""".stripMargin) + .coalesce(1) + .createOrReplaceTempView("knn_pred_queries") + // Different column layouts expose incorrectly bound residual attributes. Query 2's + // nearest object fails its score predicate; the farther passing object must not replace it. + sparkSession + .sql(""" + |SELECT 'object' AS label, score, ST_Point(x, x) AS g, id, floor + |FROM VALUES + | (101, 1.0D, 20, 50), + | (102, 11.0D, 20, 50), + | (103, 21.0D, 20, 50), + | (104, 31.0D, 20, 50), + | (105, 12.0D, 0, 50) + |AS o(id, x, score, floor) + |""".stripMargin) + .coalesce(1) + .createOrReplaceTempView("knn_pred_objects") + body + } finally { + sparkSession.catalog.dropTempView("knn_pred_queries") + sparkSession.catalog.dropTempView("knn_pred_objects") } } } From 2b7784a5829ec261898f1d3738410079f853222c Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Thu, 24 Sep 2026 22:24:48 -0700 Subject: [PATCH 3/4] [GH-3398] Reject KNN residuals before planning spatial joins Validate every detected spatial join residual so an ST_KNN marker left after ST_Intersects or ST_DWithin fails during planning with an actionable ordering message. Preserve the existing error for multiple KNN predicates and the behavior of valid KNN-first joins. Add regression coverage for both matcher paths and all join strategies, including positive KNN-first spatial filters with reversed table order. Document residual filtering, the optimizer pushdown caveat, and the single-KNN and spatial predicate ordering requirements. --- docs/api/sql/NearestNeighbourSearching.md | 4 +++ .../strategy/join/JoinQueryDetector.scala | 14 ++++++---- .../org/apache/sedona/sql/KnnJoinSuite.scala | 27 ++++++++++++++++++- 3 files changed, 39 insertions(+), 6 deletions(-) diff --git a/docs/api/sql/NearestNeighbourSearching.md b/docs/api/sql/NearestNeighbourSearching.md index 26712a07a18..c081a2c2489 100644 --- a/docs/api/sql/NearestNeighbourSearching.md +++ b/docs/api/sql/NearestNeighbourSearching.md @@ -68,6 +68,10 @@ CACHE TABLE knnResult; SELECT * FROM knnResult WHERE condition; ``` +Additional `ON` predicates that remain in the kNN join condition are evaluated on the neighbors selected by `ST_KNN`. If a predicate rejects a selected pair, the join does not choose another neighbor to replace it, so it may return fewer than `k` neighbors for a query geometry. Spark may push predicates that reference only one input below the join, as described above. + +Use only one `ST_KNN` predicate per join condition. When combining it with another spatial predicate, place `ST_KNN` before predicates such as `ST_Intersects` in the `ON` condition; otherwise, query planning may reject the join. + ### Optimization Barrier Use the `barrier` function to prevent filter pushdown and control predicate evaluation order in complex spatial joins. This function creates an optimization barrier by evaluating boolean expressions at runtime. diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala index 2a392142f58..e228fca7ee8 100644 --- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala +++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala @@ -273,6 +273,15 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { val joinConditionMatcher = OptimizableJoinCondition(left, right) val queryDetection: Option[JoinQueryDetection] = condition.flatMap { case joinConditionMatcher(predicate, extraCondition) => + // ST_KNN must be implemented by the join, never evaluated as a per-pair filter. + if (extraCondition.exists(_.exists(_.isInstanceOf[ST_KNN]))) { + val message = if (predicate.isInstanceOf[ST_KNN]) { + "Only one ST_KNN predicate is supported per join condition" + } else { + "Place ST_KNN before other spatial predicates in the join condition" + } + throw new UnsupportedOperationException(message) + } predicate match { // ST_Contains / ST_Intersects / ST_Within / ST_Equals are InferredExpression (not // ST_Predicate) so they can't sit inside getJoinDetection; they're also the only @@ -618,11 +627,6 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy { Some(distance))) // ST_KNN - case _: ST_KNN if extraCondition.exists(_.exists(_.isInstanceOf[ST_KNN])) => - // The physical join implements only the selected KNN predicate. - throw new UnsupportedOperationException( - "Only one ST_KNN predicate is supported per join condition") - case ST_KNN(Seq(leftShape, rightShape, k)) => Some( JoinQueryDetection( diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala b/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala index 7ec2aace331..b21910a6dd4 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/KnnJoinSuite.scala @@ -501,7 +501,11 @@ class KnnJoinSuite extends TestBaseScala with TableDrivenPropertyChecks { Seq((1, 101), (4, 104))), ("KNN first", s"($knnPredicate AND $scorePredicate) AND $ceilingPredicate", Seq((1, 101))), ("KNN middle", s"$scorePredicate AND ($knnPredicate AND $ceilingPredicate)", Seq((1, 101))), - ("KNN last", s"($scorePredicate AND $ceilingPredicate) AND $knnPredicate", Seq((1, 101)))) + ("KNN last", s"($scorePredicate AND $ceilingPredicate) AND $knnPredicate", Seq((1, 101))), + ( + "spatial residual with KNN first", + s"$knnPredicate AND ST_DWithin(q.g, o.g, CAST(q.id AS DOUBLE))", + Seq((2, 102), (3, 103), (4, 104)))) for { (strategy, hint, expectedPlan) <- residualJoinStrategies @@ -546,6 +550,27 @@ class KnnJoinSuite extends TestBaseScala with TableDrivenPropertyChecks { } } } + + val spatialPredicatesBeforeKnn = + Seq("ST_Intersects(ST_Buffer(q.g, 2), o.g)", "ST_DWithin(q.g, o.g, 2)") + for { + (strategy, hint, _) <- residualJoinStrategies + spatialPredicate <- spatialPredicatesBeforeKnn + } { + it(s"KNN residuals reject KNN after $spatialPredicate during planning: $strategy") { + withKnnResidualInputs { + val exception = intercept[UnsupportedOperationException] { + sparkSession + .sql(s"SELECT $hint q.id, o.id FROM knn_pred_queries q JOIN knn_pred_objects o " + + s"ON $spatialPredicate AND $knnPredicate") + .queryExecution + .executedPlan + } + exception.getMessage should include( + "Place ST_KNN before other spatial predicates in the join condition") + } + } + } } private def withKnnResidualInputs(body: => Unit): Unit = { From e19400d6621e21bddd651f826354b1b36722a079 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Fri, 25 Sep 2026 00:03:27 -0700 Subject: [PATCH 4/4] [GH-3398] Build the MinIO test image from verified upstream binaries Quay no longer grants anonymous pull access to minio/minio, causing both S3 reader tests to time out fetching their container image and cancel the Spark build matrix. Build the test image from the same pinned MinIO release on GitHub and verify architecture-specific SHA-256 checksums. Share the image lazily between the reader suites while retaining separate containers per test. Write the Dockerfile build context against Spark's bundled tar API to avoid changing Spark 3.5's older compression dependencies for these tests. --- .../src/test/resources/minio/Dockerfile | 33 +++++++++++++++++++ .../sedona/sql/GeoPackageReaderTest.scala | 7 +--- .../org/apache/sedona/sql/OsmReaderTest.scala | 7 +--- .../org/apache/sedona/sql/TestBaseScala.scala | 33 +++++++++++++++++++ 4 files changed, 68 insertions(+), 12 deletions(-) create mode 100644 spark/common/src/test/resources/minio/Dockerfile diff --git a/spark/common/src/test/resources/minio/Dockerfile b/spark/common/src/test/resources/minio/Dockerfile new file mode 100644 index 00000000000..1672e48037f --- /dev/null +++ b/spark/common/src/test/resources/minio/Dockerfile @@ -0,0 +1,33 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +FROM alpine:3.22.1 + +# Keep the same upstream MinIO release without depending on its withdrawn registry image. +RUN set -eux; \ + case "$(uname -m)" in \ + x86_64) arch=amd64; sha256=7c5bd8512c6e966455b1d198209358b2d191c77a83ab377c4073281065fb855f ;; \ + aarch64) arch=arm64; sha256=5c83cd2cf151717ba0243f73e1c7802ff36e272b67144bdd7f1f7d684fd6f03d ;; \ + *) echo "Unsupported MinIO test image architecture: $(uname -m)" >&2; exit 1 ;; \ + esac; \ + wget -O /usr/bin/minio \ + "https://github.com/minio/minio/releases/download/RELEASE.2025-09-07T16-13-09Z/minio.linux-${arch}.RELEASE.2025-09-07T16-13-09Z"; \ + echo "${sha256} /usr/bin/minio" | sha256sum -c -; \ + chmod +x /usr/bin/minio + +EXPOSE 9000 9001 +ENTRYPOINT ["/usr/bin/minio"] diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala b/spark/common/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala index 2a873621831..49554c9fa8b 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala @@ -26,8 +26,6 @@ import org.apache.spark.sql.sedona_sql.UDT.GeometryUDT import org.apache.spark.sql.types.{BinaryType, BooleanType, DateType, DoubleType, IntegerType, StringType, StructField, StructType, TimestampType} import org.scalatest.matchers.should.Matchers import org.scalatest.prop.TableDrivenPropertyChecks._ -import org.testcontainers.containers.MinIOContainer -import org.testcontainers.utility.DockerImageName import java.io.FileInputStream import java.sql.{Date, Timestamp} @@ -375,10 +373,7 @@ class GeoPackageReaderTest extends TestBaseScala with Matchers { describe("Reading from S3") { it("should be able to read files from S3") { - val container = new MinIOContainer( - DockerImageName - .parse("quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z") - .asCompatibleSubstituteFor("minio/minio")) + val container = TestBaseScala.createMinioContainer() container.start() diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/OsmReaderTest.scala b/spark/common/src/test/scala/org/apache/sedona/sql/OsmReaderTest.scala index bf460f9e340..756a6723ba9 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/OsmReaderTest.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/OsmReaderTest.scala @@ -21,8 +21,6 @@ package org.apache.sedona.sql import io.minio.{ListObjectsArgs, MakeBucketArgs, MinioClient} import org.apache.spark.sql.functions.col import org.scalatest.matchers.should.Matchers -import org.testcontainers.containers.MinIOContainer -import org.testcontainers.utility.DockerImageName import java.io.FileInputStream @@ -111,10 +109,7 @@ class OsmReaderTest extends TestBaseScala with Matchers { } it("should be able to read from osm file on s3") { - val container = new MinIOContainer( - DockerImageName - .parse("quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z") - .asCompatibleSubstituteFor("minio/minio")) + val container = TestBaseScala.createMinioContainer() container.start() diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/TestBaseScala.scala b/spark/common/src/test/scala/org/apache/sedona/sql/TestBaseScala.scala index 0068da6060f..1760287143b 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/TestBaseScala.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/TestBaseScala.scala @@ -20,6 +20,7 @@ package org.apache.sedona.sql import com.google.common.math.DoubleMath import io.minio.{MinioClient, PutObjectArgs} +import org.apache.commons.compress.archivers.tar.{TarArchiveEntry, TarArchiveOutputStream} import org.apache.hadoop.fs.FileUtil import org.apache.hadoop.hdfs.{HdfsConfiguration, MiniDFSCluster} import org.apache.log4j.{Level, Logger} @@ -33,6 +34,8 @@ import org.locationtech.jts.geom._ import org.locationtech.jts.io.WKTReader import org.scalatest.{BeforeAndAfterAll, FunSpec} import org.testcontainers.containers.MinIOContainer +import org.testcontainers.images.builder.{ImageFromDockerfile, Transferable} +import org.testcontainers.utility.DockerImageName import java.io.{File, FileInputStream} import java.nio.file.Files @@ -417,3 +420,33 @@ trait TestBaseScala extends FunSpec with BeforeAndAfterAll { .set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") } } + +object TestBaseScala { + // Build once per test JVM from the pinned upstream binary; MinIO's registry image is unavailable. + private lazy val minioImage: DockerImageName = { + val resource = getClass.getResourceAsStream("/minio/Dockerfile") + val content = + try resource.readAllBytes() + finally resource.close() + val dockerfile = new Transferable { + override def getSize: Long = content.length + override def getBytes: Array[Byte] = content + + // Compile against Spark's commons-compress API. Testcontainers' precompiled implementation + // calls a newer overload that is absent from Spark 3.5's commons-compress version. + override def transferTo(output: TarArchiveOutputStream, path: String): Unit = { + val entry = new TarArchiveEntry(path) + entry.setMode(getFileMode) + entry.setSize(content.length) + output.putArchiveEntry(entry) + output.write(content) + output.closeArchiveEntry() + } + } + DockerImageName + .parse(new ImageFromDockerfile().withFileFromTransferable("Dockerfile", dockerfile).get()) + .asCompatibleSubstituteFor("minio/minio") + } + + def createMinioContainer(): MinIOContainer = new MinIOContainer(minioImage) +}