sunchao commented on code in PR #6035:
URL: https://github.com/apache/datafusion-comet/pull/6035#discussion_r4353919707
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -724,6 +724,64 @@ case class CometExecRule(session: SparkSession,
queryStagePrep: Boolean = false)
}
}
+ /**
+ * Spark's collect limit consumes at most `limit` rows per partition, but a
native projection
+ * evaluates its whole input batch before that limit can stop reading. Limit
the projection's
+ * input first so an expression on an unconsumed row cannot raise an error.
Keep the outer
+ * limit: it still combines partitions and drops OFFSET rows, which Spark
evaluates before
+ * dropping.
+ *
+ * Only handle an immediately adjacent native projection. A filter or join
between the limit and
+ * projection can require more input rows, and a blocking operator must
consume its input.
+ * Non-collect LocalLimit/GlobalLimit plans can still evaluate a projection
before limiting its
+ * output. That preexisting gap and throwing filter predicates are tracked
separately in
+ * https://github.com/apache/datafusion-comet/issues/7147.
Review Comment:
Agreed. #7147 already records that native planning wraps the projection in a
limit, allowing the projection to evaluate the whole batch before its output is
sliced. I added an explicit follow-up item to carry that mechanism into the
rule's scaladoc if the fix extends `limitCollectProjectionInput`. Keeping that
change with the follow-up as suggested.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -724,6 +724,64 @@ case class CometExecRule(session: SparkSession,
queryStagePrep: Boolean = false)
}
}
+ /**
+ * Spark's collect limit consumes at most `limit` rows per partition, but a
native projection
+ * evaluates its whole input batch before that limit can stop reading. Limit
the projection's
+ * input first so an expression on an unconsumed row cannot raise an error.
Keep the outer
+ * limit: it still combines partitions and drops OFFSET rows, which Spark
evaluates before
+ * dropping.
+ *
+ * Only handle an immediately adjacent native projection. A filter or join
between the limit and
+ * projection can require more input rows, and a blocking operator must
consume its input.
+ * Non-collect LocalLimit/GlobalLimit plans can still evaluate a projection
before limiting its
+ * output. That preexisting gap and throwing filter predicates are tracked
separately in
+ * https://github.com/apache/datafusion-comet/issues/7147.
+ */
+ private def limitCollectProjectionInput(op: SparkPlan): SparkPlan = op match
{
+ case limit: CollectLimitExec if limit.limit >= 0 =>
+ limit.child match {
+ case project: CometProjectExec =>
+ // Selecting or renaming columns cannot throw. Keep Project ->
Filter adjacent so
+ // native planning can prune columns used only by the filter's
predicate.
+ val onlyColumns = project.projectList.forall {
+ case _: Attribute | Alias(_: Attribute, _) => true
+ case _ => false
+ }
+ val alreadyCapped = project.child match {
+ case cap: CometLocalLimitExec => cap.limit <= limit.limit
+ case _ => false
+ }
+ if (onlyColumns || alreadyCapped) {
+ op
Review Comment:
Thanks for confirming. Keeping the combined conditional and the existing
`cap.limit <= limit.limit` guard unchanged.
##########
docs/source/user-guide/latest/compatibility/index.md:
##########
@@ -106,6 +106,32 @@ Separately, Comet's native Parquet scan currently rejects
string columns whose s
valid UTF-8 rather than reading them like Spark
([#4121](https://github.com/apache/datafusion-comet/issues/4121)).
+## ANSI errors from rows skipped by LIMIT
+
+Native operators evaluate batches of rows. With `spark.sql.ansi.enabled=true`,
some `LIMIT`
+plan shapes can therefore raise an expression error from a row that Spark
stops before evaluating.
+For example, with a Parquet batch read in the order `(id, value) = (0, '0'),
(1, '1')`:
+
+```sql
+SELECT value FROM t
+WHERE CAST(IF(id = 1, 'bad', value) AS INT) >= 0
+LIMIT 1
+```
+
+Spark returns `'0'` after the first row matches. Comet can evaluate the filter
predicate on the
+second row in the same batch and raise `CAST_INVALID_INPUT`. SQL without
`ORDER BY` does not
+guarantee this input order; the example describes a physical scan order that
exposes the difference.
+
+Comet caps projection input when a collect limit directly consumes the
projection. This protects
+that projection's expressions, but does not bound evaluation inside its child
operators. Throwing
+filter predicates and projections beneath non-collect `LocalLimit` /
`GlobalLimit` plans, such as
+a `LIMIT` inside a subquery, remain affected. These existing limitations and
the investigation of
+already-ordered `TakeOrderedAndProject` inputs are tracked in
+[#7147](https://github.com/apache/datafusion-comet/issues/7147).
+
+An error on a row absent from the final result is not always a mismatch: Spark
can also need to
+evaluate that row, for example to sort the input or process an `OFFSET` prefix.
Review Comment:
Thanks. Keeping both the physical-order qualification and the distinction
between a discarded row and a row Spark never evaluates.
--
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]
