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: [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]
