andygrove opened a new issue, #6573:
URL: https://github.com/apache/datafusion-comet/issues/6573
### Describe the bug
With `spark.comet.convert.parquet.enabled=true`, `input_file_name()`,
`input_file_block_start()` and `input_file_block_length()` return `""`, `-1`
and `-1` for every row when Spark evaluates them above a Comet operator that
runs over a converted Spark Parquet scan. Those are `InputFileBlockHolder`'s
unset defaults. Spark returns each row's file path and block offsets. The query
succeeds, so the wrong values are silent.
#3312 fixed this for the native scan: `CometScanRule` keeps Spark's
`FileSourceScanExec` when the plan references any of these expressions. With
`spark.comet.convert.parquet.enabled=true`, `CometExecRule` then wraps that
scan in `CometSparkToColumnarExec`, operators above it run natively, and the
Spark `Project` that evaluates the expressions ends up above
`CometColumnarToRow`:
```
*(1) Project [input_file_name() AS f#6, input_file_block_start() AS s#7L,
input_file_block_length() AS l#8L, id#4L]
+- *(1) CometColumnarToRow
+- CometFilter [id#4L], (isnotnull(id#4L) AND (id#4L >= 0))
+- CometSparkColumnarToColumnar
+- FileScan parquet [id#4L] Batched: true, ...
```
### Steps to reproduce
```scala
spark.range(9000).repartition(3).write.parquet("/tmp/ifn")
spark.conf.set("spark.comet.convert.parquet.enabled", "true")
spark.read
.parquet("/tmp/ifn")
.where("id >= 0")
.selectExpr("input_file_name()", "input_file_block_start()",
"input_file_block_length()", "id")
.orderBy("id")
.show(3, false)
```
Comet returns `["", -1, -1, 0]`, `["", -1, -1, 1]`, `["", -1, -1, 2]`. The
filter matters: without a Comet operator between the scan and the project,
Comet drops the conversion and the project reads the scan directly.
### Expected behavior
Each row reports the file it was read from, as Spark does. Keeping such
plans on Spark would also be acceptable.
### Additional context
- Reproduced on `main` at `432523fb5` with the default Spark 4.1 profile, as
a `CometExecSuite` test using `checkSparkAnswer`.
- A likely fix is to apply the check `CometScanRule` already makes
(`InputFileName`, `InputFileBlockStart` or `InputFileBlockLength` anywhere in
the plan) before `CometExecRule.shouldApplySparkToColumnar` converts a scan.
- The CSV and JSON conversions (`spark.comet.convert.csv.enabled`,
`spark.comet.convert.json.enabled`) and DSv2 `BatchScanExec` conversion
probably behave the same way. I have not tested them.
- Found while reviewing #6566, which does not change this behavior.
--
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]
