Josh Rosen created SPARK-58485:
----------------------------------
Summary: 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
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 (QueryPlan.scala:327-341).
* {{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
(subquery.scala:99/108). 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]
