viirya commented on code in PR #6035:
URL: https://github.com/apache/datafusion-comet/pull/6035#discussion_r4342677938
##########
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:
This is the right place for it, and the wording does the job: it says the
gap exists, names the shape, and points at the issue rather than leaving a
future reader to re-derive what I had to go and check.
One thought for #7147 rather than this PR: the mechanism you found — that
the native local limit slices the projection's output, so it cannot bound that
projection's input — is the part that makes the non-collect shape non-obvious,
and it isn't captured in this sentence. If #7147's fix ends up extending this
rule, that sentence is where the next person will start.
##########
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:
Confirmed equivalent to the two early `return`s it replaces: `alreadyCapped`
is true only for a `CometLocalLimitExec` child with `cap.limit <= limit.limit`
and false for any other child, which is exactly when the old code returned
`op`. The wider-existing-cap case (inner `LIMIT 10` under an outer collect
`LIMIT 2`) still falls through and gets the tighter cap.
Reads better than what I was commenting on. No further changes from me.
##########
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:
This closing paragraph is the most useful sentence in the section. "An error
on a row absent from the final result is not always a mismatch" is exactly the
distinction that makes the rest actionable — without it, a user hitting an ANSI
error under a `LIMIT` has no way to tell whether they found a Comet bug or
correct Spark behavior they should expect, and would reasonably file either way.
Same for noting that SQL without `ORDER BY` does not guarantee the input
order the example relies on. That keeps the example honest about being a
physical-order illustration rather than a contract.
--
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]
