Describe the bug
With spark.comet.convert.typedDataset.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 reads the converted output of a typed Dataset operation. Spark returns each row's file path and block offsets. The query succeeds, so the wrong values are silent.
This is #6573 for the conversion that #6564 added. CometSparkToColumnarExec fills an Arrow batch from SerializeFromObjectExec, which pulls rows from the scan below the typed operation. By the time the Spark Project evaluates the expressions, the reader has moved on or unset InputFileBlockHolder. The SerializeFromObjectExec case in CometExecRule.transform calls convertTypedDatasetOutput without going through shouldApplySparkToColumnar, so the guard that #6703 adds there does not cover it.
*(2) Project [input_file_name() AS input_file_name()#576, input_file_block_start() AS input_file_block_start()#577L, input_file_block_length() AS input_file_block_length()#578L, value#568L]
+- *(2) CometColumnarToRow
+- CometFilter [value#568L], (value#568L >= 0)
+- CometSparkRowToColumnar
+- *(1) SerializeFromObject [input[0, bigint, false] AS value#568L]
+- *(1) MapElements ..., obj#566: bigint
+- *(1) DeserializeToObject assertnotnull(id#558L), obj#564: bigint
+- *(1) ColumnarToRow
+- FileScan parquet [id#558L] Batched: true, ...
Steps to reproduce
spark.range(9000).repartition(3).write.parquet("/tmp/ifn")
spark.conf.set("spark.comet.convert.typedDataset.enabled", "true")
spark.read
.parquet("/tmp/ifn")
.as[Long]
.map(_ + 1)
.where("value >= 0")
.selectExpr("input_file_name()", "input_file_block_start()", "input_file_block_length()", "value")
.show(3, false)
Comet returns ["", -1, -1, <value>] for all 9000 rows.
Expected behavior
Each row reports the file it was read from, as Spark does. Leaving the typed operation's output unconverted when the plan uses these expressions would also be acceptable.
Additional context
Describe the bug
With
spark.comet.convert.typedDataset.enabled=true,input_file_name(),input_file_block_start()andinput_file_block_length()return"",-1and-1for every row when Spark evaluates them above a Comet operator that reads the converted output of a typed Dataset operation. Spark returns each row's file path and block offsets. The query succeeds, so the wrong values are silent.This is #6573 for the conversion that #6564 added.
CometSparkToColumnarExecfills an Arrow batch fromSerializeFromObjectExec, which pulls rows from the scan below the typed operation. By the time the SparkProjectevaluates the expressions, the reader has moved on or unsetInputFileBlockHolder. TheSerializeFromObjectExeccase inCometExecRule.transformcallsconvertTypedDatasetOutputwithout going throughshouldApplySparkToColumnar, so the guard that #6703 adds there does not cover it.Steps to reproduce
Comet returns
["", -1, -1, <value>]for all 9000 rows.Expected behavior
Each row reports the file it was read from, as Spark does. Leaving the typed operation's output unconverted when the plan uses these expressions would also be acceptable.
Additional context
mainatb80bf4e08with the default Spark 4.1 profile, comparing against the same query withspark.comet.enabled=false. It still reproduces with fix: keep a Spark scan unconverted when the plan reads input_file_name #6703 merged intomain.CometScanRule.readsInputFileBlock(plan)from fix: keep a Spark scan unconverted when the plan reads input_file_name #6703 in theSerializeFromObjectExeccase, next to the partial-reader check, and add the case to the typed Dataset paragraph indocs/source/contributor-guide/adding_a_new_operator.md.