[BUG] NDS query 5 fails with AdaptiveSparkPlanExec assertion #4625
Comments
jlowe
added
bug
|
The assertion is for the child of |
|
@sperlingxx and @andygrove can you take a look at this? I think it may be related to #4440. It looks like Spark tries to reuse a GpuBroadcastExchange for a context that isn't going to execute on the GPU (or at least it doesn't know yet whether it will), so a ColumnarToRow transition gets inserted which then screws up AdaptiveSparkPlanExec's assumptions about the child node. That seems like a bug in either Spark's planning of columnar plans or AdaptiveSparkPlanExec's handling of columnar plans, as Spark's rules are injecting the plan node that makes AdaptiveSparkPlanExec upset. I suspect we may be able to fix this specific instance by peeling off the ColumnarToRowExec, as the plan after the AdaptiveSparkPlanExec may ultimately be planned on the GPU later, but I do wonder about the case where the GpuBroadcastExchange really is trying to be reused by a non-GPU plan. |
|
I reproduced the issue and tried just removing the This was called from: This code path assumes row-based results and I am not sure if we can override this behavior here. I then tried another approach where I pushed the ColumnarToRow to inside the BroadcastQueryStage (which possibly partly defeats the point of reusing GPU broadcasts in the first place?) using this logic: This failed with: caused by this logic in @transient val broadcast = plan match {
case b: BroadcastExchangeLike => b
case ReusedExchangeExec(_, b: BroadcastExchangeLike) => b
case _ =>
throw new IllegalStateException(s"wrong plan for broadcast stage:\n ${plan.treeString}")
}I do not see an obvious solution to this issue yet but will keep looking. |
|
Hi @jlowe @andygrove, I am really sorry for causing this bug. I think I found the root cause: when we start to replace SubqueryBroadcast with GPU overrides, we missed a potenial condition when searching SubqueryBroadcast through pattern matching: code link. private lazy val partitionFilters = wrapped.partitionFilters.map { filter =>
filter.transformDown {
case dpe @ DynamicPruningExpression(inSub: InSubqueryExec)
if inSub.plan.isInstanceOf[SubqueryBroadcastExec] =>
val subBcMeta = GpuOverrides.wrapAndTagPlan(inSub.plan, conf)
subBcMeta.tagForExplain()
val gpuSubBroadcast = subBcMeta.convertIfNeeded().asInstanceOf[BaseSubqueryExec]
dpe.copy(inSub.copy(plan = gpuSubBroadcast))
}
}There exists another possible pattern: the SubqueryBroadcastExec may be used. If it is reused before applying GPU overrides rules, the pattern becomes If I am not wrong, this pattern only appears when AQE is on. When AQE is off, no rules other than Under this circumstance, Perhaps it was a little bit reckless to simply remove |
|
Once this is fixed, can we try to do some fuzz testing to see if there is anything else that we might have missed? I know it is a large task, but it would really help me feel more comfortable that we have both DPP and AQE covered. |
Running NDS query 5 with Spark 3.2.0 on partitioned input results in the following exception:
The text was updated successfully, but these errors were encountered: