You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "kazuyukitanimura (via GitHub)" <gi...@apache.org> on 2026/10/06 01:01:35 UTC
[PR] fix: keep a Spark scan unconverted when the plan reads input_file_name [datafusion-comet]
kazuyukitanimura opened a new pull request, #6703:
URL: https://github.com/apache/datafusion-comet/pull/6703
fix: keep a Spark scan unconverted when the plan reads input_file_name
## Which issue does this PR close?
Closes #6573.
## Rationale for this change
`input_file_name()`, `input_file_block_start()` and `input_file_block_length()` read
`InputFileBlockHolder`, a thread-local that the reader producing the rows (`FileScanRDD`, the V2
file readers, `HadoopRDD`) sets as it moves from file to file. #3312 made `CometScanRule` keep
Spark's `FileSourceScanExec` when a plan uses them. With `spark.comet.convert.parquet.enabled=true`,
though, `CometExecRule` then wrapped that scan in `CometSparkToColumnarExec`, ran the operators above
it natively, and left the Spark `Project` that evaluates the expressions above `CometColumnarToRow`.
The conversion fills a batch, and Comet operators above it pull more input before they emit. By the
time Spark evaluates the expressions, the reader has moved on to a later file or reached the end of
its input and unset the holder. Rows then report another file's values or `""`, `-1`, `-1`. The query
succeeds, so the wrong answer is silent.
The issue notes that keeping such plans on Spark is acceptable, and that is what this PR does.
## What changes are included in this PR?
- `CometScanRule.readsInputFileBlock(plan)`: the check `CometScanRule` already made for the native
Parquet scan, moved into the companion object so both rules share it.
- `CometExecRule.shouldApplySparkToColumnar` now refuses to convert a leaf when the plan reads these
expressions, and records the fallback reason "Spark to Arrow conversion is not compatible with
input_file_name, input_file_block_start, or input_file_block_length". The previous body is
unchanged, renamed to `canApplySparkToColumnar`. The plan walk is lazy, so it only runs once a leaf
could otherwise be converted.
- The guard applies to every leaf, not just Parquet. That covers the CSV and JSON conversions, V1
and V2 scans, and other `spark.comet.sparkToColumnar` leaves such as `RDDScan`, whose `HadoopRDD`
sets the same holder.
- A note in the user guide's data sources page.
Not in this PR: the native Iceberg scan and the native CSV V2 scan
(`spark.comet.scan.csv.v2.enabled`) have no such guard and have the same problem. That gap predates
this change and needs its own fix in `CometScanRule.transformV2Scan`, which I will file separately.
## How are these changes tested?
Two new tests in `CometExecSuite`. Each puts a filter between the scan and the projection, so a Comet
operator would sit above the conversion, then compares results with Spark using
`checkSparkAnswerAndFallbackReason` and asserts the plan has no `CometSparkToColumnarExec`:
- `input_file_name above a converted Spark file scan keeps the scan on Spark`: Parquet, CSV and JSON,
each as a V1 and a V2 scan.
- `input_file_name above a converted RDD scan keeps the scan on Spark`: an `RDDScan` over
`sparkContext.textFile`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: keep a Spark scan unconverted when the plan reads input_file_name [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #6703:
URL: https://github.com/apache/datafusion-comet/pull/6703
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: keep a Spark scan unconverted when the plan reads input_file_name [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on code in PR #6703:
URL: https://github.com/apache/datafusion-comet/pull/6703#discussion_r4200435804
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -447,7 +450,7 @@ case class CometExecRule(session: SparkSession)
case c: CometSparkToColumnarExec =>
convertToComet(c, CometScanWrapper).getOrElse(c)
- case op if shouldApplySparkToColumnar(conf, op) =>
+ case op if shouldApplySparkToColumnar(conf, op, readsInputFileBlock) =>
Review Comment:
addressed
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: keep a Spark scan unconverted when the plan reads input_file_name [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on PR #6703:
URL: https://github.com/apache/datafusion-comet/pull/6703#issuecomment-6042107459
Thank you @sunchao @andygrove
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: keep a Spark scan unconverted when the plan reads input_file_name [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6703:
URL: https://github.com/apache/datafusion-comet/pull/6703#discussion_r4194522169
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -447,7 +450,7 @@ case class CometExecRule(session: SparkSession)
case c: CometSparkToColumnarExec =>
convertToComet(c, CometScanWrapper).getOrElse(c)
- case op if shouldApplySparkToColumnar(conf, op) =>
+ case op if shouldApplySparkToColumnar(conf, op, readsInputFileBlock) =>
Review Comment:
When you rebase, #6553 added a `case op: LeafExecNode if hasEnabledHandler(op)` just above this one, and its fallback calls `shouldApplySparkToColumnar` too, so it needs `readsInputFileBlock` passed through as well.
Just below, the typed Dataset conversion from #6564 has the same bug. The `SerializeFromObjectExec` case calls `convertTypedDatasetOutput` without going through `shouldApplySparkToColumnar`, so with `spark.comet.convert.typedDataset.enabled=true` every row still comes back as `""`, -1, -1 after this PR. I filed #6708 for it. Since the conflict is in that block anyway, would you be up for adding the same check there next to the partial-reader tag, with a test, and closing #6708 here too? After the rebase, the new paragraph in `datasources.md` lands right under the typed Dataset paragraph, so it reads as if it covers that case.
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -4248,6 +4248,52 @@ class CometExecSuite extends CometTestBase {
}
}
+ // https://github.com/apache/datafusion-comet/issues/6573
+ private val inputFileColumns =
+ Seq("input_file_name()", "input_file_block_start()", "input_file_block_length()", "id")
+ private val inputFileFallbackReason =
+ "Spark to Arrow conversion is not compatible with input_file_name"
+
+ test("input_file_name above a converted Spark file scan keeps the scan on Spark") {
+ for (format <- Seq("parquet", "csv", "json"); v1List <- Seq("", format)) {
+ withTempPath { dir =>
+ spark.range(9000).repartition(3).write.format(format).save(dir.toString)
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
+ CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true",
+ CometConf.COMET_CONVERT_FROM_CSV_ENABLED.key -> "true",
+ CometConf.COMET_CONVERT_FROM_JSON_ENABLED.key -> "true") {
+ // The filter would run in Comet above the conversion. Without a Comet operator there,
+ // Comet drops the conversion and Spark's project reads the scan directly.
+ val df = spark.read
+ .schema("id LONG")
+ .format(format)
+ .load(dir.toString)
+ .where("id >= 0")
+ .selectExpr(inputFileColumns: _*)
+ val (_, cometPlan) = checkSparkAnswerAndFallbackReason(df, inputFileFallbackReason)
+ assert(collect(cometPlan) { case c: CometSparkToColumnarExec => c }.isEmpty)
+ }
+ }
+ }
+ }
+
+ test("input_file_name above a converted RDD scan keeps the scan on Spark") {
+ withTempPath { dir =>
+ spark.range(9000).repartition(3).selectExpr("CAST(id AS STRING)").write.text(dir.toString)
+ withSQLConf(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true") {
Review Comment:
Since #6602, RDDs have their own `spark.comet.convert.rdd.enabled`, and converting them through `spark.comet.sparkToColumnar.enabled` is the deprecated route. `CometTestBase` already turns the new key on, so this setting does nothing after the rebase. Could this set `CometConf.COMET_CONVERT_FROM_RDD_ENABLED` instead, so the test doesn't lean on the base class?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: keep a Spark scan unconverted when the plan reads input_file_name [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on code in PR #6703:
URL: https://github.com/apache/datafusion-comet/pull/6703#discussion_r4200296047
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -4248,6 +4248,52 @@ class CometExecSuite extends CometTestBase {
}
}
+ // https://github.com/apache/datafusion-comet/issues/6573
+ private val inputFileColumns =
+ Seq("input_file_name()", "input_file_block_start()", "input_file_block_length()", "id")
+ private val inputFileFallbackReason =
+ "Spark to Arrow conversion is not compatible with input_file_name"
+
+ test("input_file_name above a converted Spark file scan keeps the scan on Spark") {
+ for (format <- Seq("parquet", "csv", "json"); v1List <- Seq("", format)) {
+ withTempPath { dir =>
+ spark.range(9000).repartition(3).write.format(format).save(dir.toString)
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
+ CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true",
+ CometConf.COMET_CONVERT_FROM_CSV_ENABLED.key -> "true",
+ CometConf.COMET_CONVERT_FROM_JSON_ENABLED.key -> "true") {
+ // The filter would run in Comet above the conversion. Without a Comet operator there,
+ // Comet drops the conversion and Spark's project reads the scan directly.
+ val df = spark.read
+ .schema("id LONG")
+ .format(format)
+ .load(dir.toString)
+ .where("id >= 0")
+ .selectExpr(inputFileColumns: _*)
+ val (_, cometPlan) = checkSparkAnswerAndFallbackReason(df, inputFileFallbackReason)
+ assert(collect(cometPlan) { case c: CometSparkToColumnarExec => c }.isEmpty)
+ }
+ }
+ }
+ }
+
+ test("input_file_name above a converted RDD scan keeps the scan on Spark") {
+ withTempPath { dir =>
+ spark.range(9000).repartition(3).selectExpr("CAST(id AS STRING)").write.text(dir.toString)
+ withSQLConf(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true") {
Review Comment:
Addressed
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org