andygrove opened a new issue, #6667:
URL: https://github.com/apache/datafusion-comet/issues/6667
### Describe the bug
On Spark 3.4, a global aggregate over a native Iceberg scan returns no row
when the table has a single input split and dynamic partition pruning removes
that split. Spark returns one row.
Spark 3.4's `DataSourceV2ScanExecBase.outputPartitioning` reports
`SinglePartition` when the scan has one input partition, so
`EnsureRequirements` leaves out the exchange between the partial and final
aggregate. When DPP then filters out the only split, `BatchScanExec.inputRDD`
returns `sparkContext.parallelize(empty, 1)` so the plan still runs on one
partition. `CometIcebergNativeScan.serializePartitions` treats that
`ParallelCollectionRDD` as "DPP pruned everything" and emits no per-partition
data, so the native scan's RDD has 0 partitions and the final aggregate never
runs.
It happens with AQE on and off. Spark 3.5 and later are not affected,
because `DataSourceV2ScanExecBase` no longer reports `SinglePartition` there
and the plan gets an exchange. The empty-RDD branch came in with #4215, so 1.0
and 1.1 are affected too.
### Steps to reproduce
Spark 3.4, `spark.comet.scan.icebergNative.enabled=true`, Iceberg `hadoop`
catalog `probe_cat`:
```scala
val dimPath = "/tmp/probe_dim"
spark.sql("CREATE TABLE probe_cat.db.fact (amount INT, store_id INT) USING
iceberg PARTITIONED BY (store_id)")
// every row in one partition, so the table is a single data file and a
single split
spark.range(100).selectExpr("cast(id as int) as amount", "5 as store_id")
.coalesce(1).writeTo("probe_cat.db.fact").append()
spark.range(1).selectExpr("99 as store_id", "'X' as
country").write.parquet(dimPath)
spark.read.parquet(dimPath).createOrReplaceTempView("probe_dim")
spark.sql("""SELECT /*+ BROADCAST(d) */ count(*) AS c, sum(f.amount) AS s
FROM probe_cat.db.fact f JOIN probe_dim d ON f.store_id = d.store_id
WHERE d.country = 'X'""").collect()
```
### Expected behavior
`[0,null]`, as Spark returns. Comet returns no rows. The final plan has
`HashAggregate(Final)` directly over `HashAggregate(Partial)`, with no
exchange, above the `CometIcebergNativeScan`, whose RDD has 0 partitions.
### Additional context
Found while reviewing #6268, which does not change this (the scan's
execution partition count is the same before and after). Emitting one empty
partition in the `ParallelCollectionRDD` case, to match the partition count
Spark's `inputRDD` has, looks like the fix.
--
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]
