weiqingy commented on code in PR #2458:
URL: https://github.com/apache/auron/pull/2458#discussion_r3746755657
##########
spark-extension-shims-spark/src/test/scala/org/apache/auron/exec/AuronExecSuite.scala:
##########
@@ -43,6 +44,34 @@ class AuronExecSuite extends AuronQueryTest with
BaseAuronSQLSuite {
}
}
+ test("CollectLimit batches partition scans") {
+ withTempPath { path =>
+ spark.range(0, 80, 1, 8).write.parquet(path.getCanonicalPath)
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ "spark.sql.files.maxPartitionBytes" -> "4096",
+ "spark.sql.limit.initialNumPartitions" -> "100") {
+ val df = spark.read.parquet(path.getCanonicalPath).where("id <
0").limit(1)
+ val collectLimit = collectFirst(df.queryExecution.executedPlan) {
+ case exec: NativeCollectLimitExec => exec
+ }.get
+ val numPartitions = collectLimit.child.execute().getNumPartitions
+ assert(numPartitions > 1 && numPartitions <= 100)
+
+ val jobGroup = s"collect-limit-${System.nanoTime()}"
+ spark.sparkContext.setJobGroup(jobGroup, "test CollectLimit job count")
+ try {
+ assert(collectLimit.executeCollect().isEmpty)
+ val jobCount =
spark.sparkContext.statusTracker.getJobIdsForGroup(jobGroup).length
Review Comment:
`getJobIdsForGroup` reads from `AppStatusStore`, which is filled by a
listener on Spark's async event bus. So it counts jobs whose events have been
processed by now, not jobs that were submitted.
The race falls on the unhelpful side. If a `JobStart` is still queued, the
count comes back low, and a low count passes this assertion. On the old
one-job-per-partition code, one late event would be enough to make this test go
green.
I have not seen it flake, this is just from reading the path. Would draining
the bus before the read make it deterministic?
`AuronAdaptiveQueryExecSuite.scala:109` uses
`spark.sparkContext.listenerBus.waitUntilEmpty()` for the same reason. It is
`private[spark]`, so the caller has to sit under `org.apache.spark`, and
`AuronQueryTest` already does.
##########
spark-extension-shims-spark/src/test/scala/org/apache/auron/exec/AuronExecSuite.scala:
##########
@@ -43,6 +44,34 @@ class AuronExecSuite extends AuronQueryTest with
BaseAuronSQLSuite {
}
}
+ test("CollectLimit batches partition scans") {
+ withTempPath { path =>
+ spark.range(0, 80, 1, 8).write.parquet(path.getCanonicalPath)
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ "spark.sql.files.maxPartitionBytes" -> "4096",
+ "spark.sql.limit.initialNumPartitions" -> "100") {
Review Comment:
`spark.sql.limit.initialNumPartitions` was added in Spark 3.4, with a
default of 1, and `SparkPlan.executeTake` only reads it from 3.4 on. So on the
spark-3.0 to 3.3 profiles this line is set but never read.
On 3.4 and later it is read, and 100 is larger than the partition count, so
the first `runJob` takes every partition. That run is a single job with no
batching. Only the older profiles go through the scale-up loop.
The regression is caught either way, since `jobCount < numPartitions` holds
in both shapes. Is there a reason to pin the value here? Without it the 3.4+
default of 1 applies, and every profile would exercise the batching this test
is named for.
--
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]
