ErikBPF commented on code in PR #5903:
URL: https://github.com/apache/datafusion-comet/pull/5903#discussion_r4367827433
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -366,6 +366,54 @@ class CometAggregateSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
classOf[LocalTableScanExec])
}
+ test("collect_list and collect_set over non-nullable nested fields") {
+ withTempPath { path =>
+ // Keep every group in one input and final partition so aggregation
outgrows its memory pool.
+ spark
+ .createDataFrame((0 until 32768).map { i =>
+ (i % 4096, i % 8192, s"${i % 8192}-" + "x" * 256)
+ })
+ .coalesce(1)
+ .write
+ .parquet(path.getAbsolutePath)
+ withParquetTable(path.getAbsolutePath, "tbl") {
+ // The literal and coalesce create required fields after the Parquet
scan, which
+ // otherwise widens field nullability. Sorting removes collect order
differences.
+ val struct = "named_struct('flag', true, 'id', coalesce(_2, 0),
'value', _3)"
+ // Ungrouped aggregation uses Accumulator rather than
GroupsAccumulator and fails
+ // on mismatched nested nullability even without spilling.
+ checkSparkAnswerAndOperator(
+ sql(s"""
+ SELECT sort_array(collect_list(s)), sort_array(collect_set(s))
+ FROM (SELECT $struct AS s FROM tbl WHERE _1 < 8)"""),
+ Seq(classOf[CometHashAggregateExec]))
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "1",
+ CometConf.COMET_BATCH_SIZE.key -> "1024",
+ CometConf.COMET_OFFHEAP_MEMORY_POOL_FRACTION.key -> "0.002",
+ CometConf.COMET_RESPECT_DATAFUSION_CONFIGS.key -> "true",
+ "spark.comet.datafusion.execution.sort_spill_reservation_bytes" ->
"65536") {
+ Seq(struct, s"array($struct)").foreach { value =>
+ val (_, cometPlan) = checkSparkAnswerAndOperator(
+ sql(s"""
+ SELECT _1, sort_array(collect_list(s)),
sort_array(collect_set(s))
+ FROM (
+ SELECT _1, $value AS s
+ FROM tbl
+ ) GROUP BY _1"""),
+ Seq(classOf[CometHashAggregateExec]))
+ val aggregates = collect(cometPlan) { case agg:
CometHashAggregateExec => agg }
+ assert(
+ aggregates.map(_.metrics("spill_count").value).sum > 0L,
+ s"Expected the native collection aggregate to
spill:\n$cometPlan")
Review Comment:
Correction to the reply above, after re-running RED on apollo against the
delivered revision `72cd19892`.
The config-free version of the grouped test **passes with `asNullable`
removed**, so it was not a gate. My earlier RED came from a revision that still
carried `withDistinctMergeMemoryPressure`, and that helper's settings change
the plan, not just the budget: with AQE on and the default shuffle partition
count, Spark's plan inserts a shuffle-boundary cast that hides the mismatch.
So `withDistinctMergeMemoryPressure` is restored, and only the spill
assertion and the two DataFusion settings stay out. Measured both ways with the
fix removed: helper kept -> fails with `column types must match schema types`;
helper dropped -> passes.
The spill measurement above is unchanged: neither stage spills at 4 MiB, 2
MiB or 400 KiB.
--
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]
