sunchao opened a new issue, #7147: URL: https://github.com/apache/datafusion-comet/issues/7147
## Problem Native batch evaluation can raise `CAST_INVALID_INPUT` on a row that Spark never evaluates because a `LIMIT` stops consumption earlier. This issue records two concrete remaining cases from [PR #6035](https://github.com/apache/datafusion-comet/pull/6035): a throwing filter predicate, and a projection beneath a non-collect `LocalLimit` / `GlobalLimit` plan. This is a focused follow-up under [#6006](https://github.com/apache/datafusion-comet/issues/6006), which tracks the broader evaluation-mask problem. These are preexisting limitations left outside #6035's fix for an immediately adjacent `CollectLimit -> CometProject`. That PR retains existing operator support and [#6826](https://github.com/apache/datafusion-comet/pull/6826)'s cast dispatch; it does not add fallback for these remaining shapes. ## Reproduction Reproduced with Spark **4.0.4** and PR #6035 head **`1dc89673272b2fe452ba4fbd11ac2b79ce0a9967`**, using Comet's test harness with native execution and native shuffle enabled. The PR has no native-code changes relative to its main base, `80b518173`. Start Spark with the Comet extension and Comet shuffle manager configured. Create the input with Comet disabled: ```scala val path = java.nio.file.Files.createTempDirectory("comet-limit-casts") .resolve("rows").toString spark.conf.set("spark.sql.adaptive.enabled", "false") spark.conf.set("spark.sql.ansi.enabled", "true") spark.conf.set("spark.comet.exec.enabled", "true") spark.conf.set("spark.comet.exec.scalaUDF.codegen.enabled", "true") spark.conf.set("spark.comet.batchSize", "8192") spark.conf.set("spark.comet.enabled", "false") spark.range(2) .selectExpr("id", "CAST(id AS STRING) AS value") .coalesce(1) .write.parquet(path) spark.read.parquet(path).createOrReplaceTempView("scope_rows") ``` The Parquet input has `(id, value) = (0, '0'), (1, '1')` in one file, with the valid row read first and both rows fitting in one batch. This is the physical order used by the reproduction; SQL without `ORDER BY` does not guarantee that order. Reading Parquet also keeps the expressions in the execution plan instead of reducing a constant `VALUES` query during optimization. Run each query with `spark.comet.enabled=false`, then with `spark.comet.enabled=true`. Construct a fresh DataFrame after each setting change, inspect `queryExecution.executedPlan`, and call `collect()`. ### Throwing predicate below a collect limit ```sql SELECT value FROM scope_rows WHERE CAST(IF(id = 1, 'bad', value) AS INT) >= 0 LIMIT 1 ``` Spark returns the string `'0'`. Comet raises `CAST_INVALID_INPUT` while evaluating the predicate on the second row. The observed Comet plan, with expression IDs omitted, is: ```text CometCollectLimit 1, 0 +- CometProject [value] +- CometFilter [id, value], if (id = 1) (cast(bad as int) >= 0) else (cast(value as int) >= 0) +- CometNativeScan parquet [id, value] ``` The cap introduced by #6035 does not bound filter input: a filter may need to read more than `limit` input rows to produce `limit` matching rows. ### Throwing projection beneath a non-collect limit ```sql SELECT parsed FROM ( SELECT CAST(IF(id = 1, 'bad', value) AS INT) AS parsed FROM scope_rows LIMIT 1 ) q WHERE parsed >= 0 ``` Spark returns the integer `0`. Comet raises `CAST_INVALID_INPUT`. The outer predicate keeps the inner limit in a non-collect plan: ```text CometColumnarToRow +- CometFilter (isnotnull(parsed) AND parsed >= 0) +- CometGlobalLimit 1, 0 +- CometExchange SinglePartition, CometNativeShuffle +- CometLocalLimit 1 +- CometProject [if (id = 1) cast(bad as int) else cast(value as int) AS parsed] +- CometNativeScan parquet [id, value] ``` This reproduces the LocalLimit/GlobalLimit plan family, specifically the `CometLocalLimit -> CometProject` edge. Native planning wraps the already-built projection in a limit, so the projection can evaluate the entire batch before the limit slices its output. A standalone `GlobalLimit -> Project` edge still needs its own coverage. ## Follow-up scope - Fix the two reproduced cast cases while preserving errors when Spark actually consumes the invalid value, and preserve native execution for valid-input controls. - Add coverage for `LocalLimit -> Project` and direct `GlobalLimit -> Project`, including offsets, multiple batches/partitions, and AQE. Confirm the physical plan for each case. - Investigate **already-ordered input** to `TakeOrderedAndProject`: Spark can stop consuming each partition after enough rows, so a throwing child projection or filter may expose the same issue. This is not yet a reproduced mismatch. The operator's own projection runs after selection in both Spark and Comet. - Keep blocking operations distinct. An unordered top-K or sort must consume its input in Spark too. `CollectTail` also reads every row of each visited partition to find its tail; an error on a discarded row there is not by itself a mismatch. No `CollectTail` defect is established by these reproductions. - Keep the compatibility guide linked to the remaining confirmed limitations while the follow-up is open. Any broader fallback policy belongs to a deliberate design and performance assessment under #6006. The recorded local probe passed its assertions for both Spark results and both Comet errors. It used AQE disabled; it does not establish behavior across other Spark versions or every physical strategy. ## Review context - [viirya's sibling-limit question](https://github.com/apache/datafusion-comet/pull/6035#discussion_r4321727663) - [andygrove's filter reproducer and documentation request](https://github.com/apache/datafusion-comet/pull/6035#pullrequestreview-5585125855) - [#5532: related unbase64 failures under partial consumption](https://github.com/apache/datafusion-comet/issues/5532) -- 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]
