karuppayya commented on code in PR #52213:
URL: https://github.com/apache/spark/pull/52213#discussion_r2371201064
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -205,6 +206,9 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
sql("analyze table bf5part compute statistics for columns a5, b5, c5, d5,
e5, f5")
sql("analyze table bf5filtered compute statistics for columns a5, b5, c5,
d5, e5, f5")
+ // Tests depend on intermediate results that would otherwise be cleaned up
when
Review Comment:
Excellent catch! Thanks @cloud-fan
I don't believe the root cause is with shuffle file cleanup itself, but
rather with how Adaptive Query Execution handles subquery synchronization.
My theory (partially verified):
1. [During codegen
phase](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BloomFilterMightContain.scala#L105-L106),
FilterExec looks for subquery results but doesn't find them (at times), so it
skips using the Bloom filter optimization
```
// The bloom filter created from `bloomFilterExpression`.
@transient private lazy val bloomFilter = {
val bytes = bloomFilterExpression.eval().asInstanceOf[Array[Byte]]
if (bytes == null) null else deserialize(bytes)
}
```
3. The main query finishes execution while the subquery is still running in
the background (separate execution context)
4. As part of query completion, shuffle cleanup removes all shuffle files,
including those needed by the still-running subquery(while subquery results are
also no longer needed as main query has completed, this is a bug in that it
doesn't use the bloom filters)
5. Subquery fails with FetchFailedException - The subquery execution fails
after the main query has already completed, trying to access cleaned-up shuffle
data.
This suites verifies only the logical plan for the presence of
BloomfilterAggregate and does not the verify if the code indeed used Bllom
filter based filtering.
This can be easily reproduced by running this suite. (Its not consistent,
and fails based on when the subquery completes. But I am sure atleast one test
would fail and cause a ripple since sc gets stopped)
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -67,6 +67,18 @@ class AdaptiveQueryExecSuite
setupTestData()
+ protected override def beforeAll(): Unit = {
+ super.beforeAll()
+ // Tests depend on intermediate results that would otherwise be cleaned up
when
Review Comment:
I think my wording might not have not been clear.
Thge query execution indeed succeeds and there is no correctness issue.
The issue is with how we assert in tests
In
`org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite#checkNumLocalShuffleReads`
```
private def checkNumLocalShuffleReads(
plan: SparkPlan, numShufflesWithoutLocalRead: Int = 0): Unit = {
val numShuffles = collect(plan) {
case s: ShuffleQueryStageExec => s
}.length
val numLocalReads = collect(plan) {
case read: AQEShuffleReadExec if read.isLocalRead => read
}
numLocalReads.foreach { r =>
val rdd = r.execute()
val parts = rdd.partitions
assert(parts.forall(rdd.preferredLocations(_).nonEmpty))
}
assert(numShuffles === (numLocalReads.length +
numShufflesWithoutLocalRead))
}
```
Specifically `rdd.preferredLocations(_).nonEmpty)` will be empty after the
cleanup and the assertion fails.
if shuffle clean up is enabled, i dont think this assertion should pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
