[
https://issues.apache.org/jira/browse/SPARK-58485?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58485:
-----------------------------------
Labels: pull-request-available (was: )
> Broadcast hash join fails with "Subquery has not finished" when the join key
> contains an uncorrelated scalar subquery
> ---------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-58485
> URL: https://issues.apache.org/jira/browse/SPARK-58485
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.0.0
> Reporter: Josh Rosen
> Priority: Major
> Labels: pull-request-available
>
> The following query deterministically crashes:
> {code:java}
> CREATE TABLE t1(a INT) USING PARQUET; INSERT INTO t1 VALUES (1), (2);
> CREATE TABLE t2(b INT) USING PARQUET; INSERT INTO t2 VALUES (1), (NULL);
> CREATE TABLE t3(c INT) USING PARQUET; INSERT INTO t3 VALUES (1);
> SELECT * FROM t1 JOIN t2 ON t1.a = coalesce(t2.b, (SELECT min(c) FROM t3));
> {code}
> with error
> {code:java}
> org.apache.spark.sql.execution.QueryExecutionException:
> java.lang.IllegalArgumentException: requirement failed: Subquery
> subquery#186, [id=#527] has not finished
> at scala.Predef$.require(Predef.scala:337)
> at
> org.apache.spark.sql.execution.ScalarSubquery.eval(subquery.scala:99)
> at
> org.apache.spark.sql.catalyst.expressions.Coalesce.eval(nullExpressions.scala:90)
> at
> org.apache.spark.sql.catalyst.expressions.UnaryExpression.eval(Expression.scala:606)
> at
> org.apache.spark.sql.catalyst.expressions.InterpretedUnsafeProjection.apply(InterpretedUnsafeProjection.scala:84)
> at
> org.apache.spark.sql.execution.joins.LongHashedRelation$.apply(HashedRelation.scala:1084)
> at
> org.apache.spark.sql.execution.joins.HashedRelation$.apply(HashedRelation.scala:158)
> at
> org.apache.spark.sql.execution.joins.HashedRelationBroadcastMode.transform(HashedRelation.scala:1164)
> at
> org.apache.spark.sql.execution.joins.HashedRelationBroadcastMode.transform(HashedRelation.scala:1152)
> at
> org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.$anonfun$relationFuture$1(BroadcastExchangeExec.scala:197)
> at
> org.apache.spark.sql.execution.SQLExecutionThreadLocalCaptured.$anonfun$runWith$2(SQLExecution.scala:63)
> [...] {code}
> Root cause:
> * {{ExtractEquiJoinKeys}} accepts a join key containing an uncorrelated
> scalar subquery (it has no references, so {{canEvaluate}} passes for either
> side), and the key lands in {{{}HashedRelationBroadcastMode(key:
> Seq[Expression]){}}}.
> * Scalar subqueries are normally started and awaited by
> {{SparkPlan.prepareSubqueries}} and {{{}waitForSubqueries{}}}, which discover
> them through {{{}QueryPlan.expressions{}}}. That walk only surfaces
> Expression, Option and Iterable constructor arguments
> ([src|https://github.com/apache/spark/blob/4dcdd4dba544210898180a410b8b30e0b85157cb/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala#L327-L341]).
> * {{BroadcastMode}} is an opaque case class, so the subquery inside
> {{mode.key}} is never registered or awaited on the exchange node.
> * The broadcast thread then evaluates/code-generates the un-updated
> {{execution.ScalarSubquery}} and hits the {{require(updated)}} guard
> ([src|https://github.com/apache/spark/blob/4dcdd4dba544210898180a410b8b30e0b85157cb/sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala#L99]).
> The join node's own {{waitForSubqueries}} does see its
> {{{}leftKeys{}}}/{{{}rightKeys{}}}, but has no happens-before relationship
> with the broadcast thread, which {{doPrepare}} launches eagerly.
> This is the same bug class as SPARK-35874 (AQE shuffle materialization not
> waiting for its subqueries), SPARK-42937 (InSubqueryExec broadcast), and
> SPARK-45584 (TakeOrderedAndProjectExec.executeCollect bypassing
> executeQuery): an execution entry point that evaluates operator expressions
> outside of executeQuery
> Possible fixes: have {{BroadcastExchangeExec}} surface {{mode.key}} to the
> subquery (register in prepareSubqueries and wait before {{mode.transform}} in
> {{{}relationFuture{}}}, mirroring SPARK-35874), or ensure planning never
> places unexecuted subquery expressions into a {{BroadcastMode}} (e.g. reject
> them in {{ExtractEquiJoinKeys}} or substitute executed results).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
