Neuw84 opened a new issue, #6133:
URL: https://github.com/apache/datafusion-comet/issues/6133
### Describe the bug
TPC-DS **q64** returns **0 rows** at scale factor 1000 with Comet native
execution, where Spark returns 12,185. The same `CometNativeScan` feeding
Spark's own operators (every `spark.comet.exec.<operator>.enabled=false`, so
the scan output goes through `ColumnarToRow`) returns the correct 12,185 rows,
and at SF10 every configuration agrees with Spark. So the scan's values are
right when read through the `ColumnVector` accessors, and the rows are lost
only when the scan's batches are consumed by native operators.
Where the rows go: with operators disabled one at a time, the loss sits in
the join of the second `cs_ui` instance of the `cross_sales` CTE against the
broadcast of `store_sales (ss_sold_date_sk in 2000) ⋈ store_returns`; before
that join the intermediate row counts agree with Spark. Turning AQE off does
not change the result. An unrelated third-party columnar operator set that
consumes Comet's scan output through the Arrow C data interface loses the rows
in exactly the same place, which suggests the difference is in the exported
batches rather than in a particular operator.
Environment: Comet 1.0.0, Spark 4.1.3 (`spark-4.1.3-bin-hadoop3`), Java 25
(Corretto), Kubernetes (Spark operator), TPC-DS SF1000 parquet on S3 read
through S3A. 8 executors × 13 cores, 18 GB heap + 16 GB overhead,
`spark.memory.offHeap.size=16g`.
| configuration | q64 rows |
|---|---|
| Spark (no Comet) | 12,185 |
| Comet scan + Comet native operators + Comet shuffle | **0** (at 200 and
300 shuffle partitions) |
| Comet scan, Spark operators and shuffle
(`spark.comet.exec.*.enabled=false`) | 12,185 |
| any of the above at SF10 | 12,185-equivalent, all agree |
### Steps to reproduce
1. Generate TPC-DS SF1000 as parquet (we used the `tpcds-kit` `dsdgen`,
`char`/`varchar` columns stored as strings; SF10 does not reproduce).
2. Run q64 (the official TPC-DS text, parameters `i_color in
('purple','burlywood','indian','spring','floral','medium')`, `i_current_price
between 64 and 74`, `cs_ui.syear = 1999` / `cs2.syear = 2000`) with:
```
--conf spark.plugins=org.apache.spark.CometPlugin
--conf spark.comet.enabled=true --conf spark.comet.scan.enabled=true
--conf spark.comet.exec.enabled=true --conf
spark.comet.exec.shuffle.enabled=true --conf spark.comet.exec.shuffle.mode=auto
--conf
spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager
--conf spark.comet.explainFallback.enabled=true --conf
spark.comet.cast.allowIncompatible=true
--conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=16g
--conf spark.sql.shuffle.partitions=300
```
→ 0 rows.
3. Same, adding `--conf spark.comet.exec.shuffle.enabled=false` and
`spark.comet.exec.{project,filter,aggregate,sort,localLimit,globalLimit,takeOrderedAndProject,hashJoin,sortMergeJoin,broadcastHashJoin,broadcastExchange,expand,union,window,coalesce,collectLimit,explode,sample}.enabled=false`,
default shuffle manager → 12,185 rows, the plan still showing
`CometNativeScan` for all 48 scans.
### Expected behavior
12,185 rows, as Spark and as Comet's scan under Spark's operators.
### Additional context
I can attach `explain formatted` output of both plans and the per-join
intermediate row counts if useful. Related to the much older #74 (q64 among the
queries once disabled for result differences), but here the scan alone is fine
and the difference appears only at scale.
--
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]
