andygrove commented on code in PR #6763: URL: https://github.com/apache/datafusion-comet/pull/6763#discussion_r4212559109
########## spark/src/test/resources/sql-tests/expressions/conditional/in_case_when_candidate.sql: ########## @@ -0,0 +1,48 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- IN builds its list once when every candidate is a constant, which it decides by evaluating the +-- candidates on an empty batch and checking for a scalar. A CASE or IF that depends on a column +-- must not return a scalar NULL there, or IN compares every row against NULL. + +-- Config: spark.comet.exec.range.enabled=true +-- Config: spark.comet.sparkToColumnar.enabled=true +-- Config: spark.comet.sparkToColumnar.supportedOperatorList=Range + +query +SELECT id, id IN (IF(id = 1, NULL, id)), id IN (nullif(id, 1), 5L) FROM range(0, 3) + +query +SELECT id, id IN (CASE WHEN id = 1 THEN NULL ELSE id END) FROM range(0, 3) + +-- no ELSE +query +SELECT id, id IN (CASE WHEN id <> 1 THEN id END), id NOT IN (CASE WHEN id <> 1 THEN id END) +FROM range(0, 3) Review Comment: `CometCoalesce` builds the same `CaseWhen` proto, so `coalesce` hit this bug too, and the fixture has no query for it. On `main` before this PR, `id IN (coalesce(v, id))` over a nullable column `v` returns NULL for every row. Could we add `SELECT id, id IN (coalesce(nullif(id, 1), 5L)) FROM range(0, 3)` and a CASE with two WHEN branches and no ELSE, such as `SELECT id, id IN (CASE WHEN id = 0 THEN id WHEN id = 2 THEN id END) FROM range(0, 3)`? The first fails without the fix, and both pass with it. ########## native/spark-expr/src/array_funcs/nested_comparison.rs: ########## @@ -376,7 +377,9 @@ pub fn spark_in_list( let constants = candidates .iter() .map(|child| { - if is_volatile(child) { + // A candidate that reads a column is not a constant, even if it returns a scalar + // for the empty batch + if is_volatile(child) || !collect_columns(child).is_empty() { Review Comment: The new `collect_columns` check only protects nested operands. Any other type returns into DataFusion's `in_list` at the top of `spark_in_list`, and `InListExpr::try_new` decides there by evaluating the candidates on an empty batch. A column-reading expression that returns a scalar for zero rows would bring this bug back for flat `IN`, with no test to notice. Could we say that in a comment on that early return, and add a sentence to `adding_a_new_expression.md` saying an expression that reads a column must return an array for an empty batch? ########## native/spark-expr/src/conditional_funcs/case_when.rs: ########## @@ -349,6 +350,13 @@ impl PhysicalExpr for CaseWhenExpr { } fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> { + // No row chooses a branch, which both the eager and the lazy evaluation answer with a + // scalar NULL. A scalar from an empty batch is taken to mean the expression is constant, + // as IN does to build its list once, so return an empty array instead. + if batch.num_rows() == 0 { + let data_type = self.data_type(&batch.schema())?; + return Ok(ColumnarValue::Array(new_empty_array(&data_type))); Review Comment: I reproduced this on Spark 4.1 with ANSI on, and it does not need a CASE. `SELECT id, id IN (0L, CAST(s AS BIGINT)) FROM t` raises `CAST_INVALID_INPUT` when `s` is `'bad'` on the row where `id = 0`, and it does the same on `main` before this PR. `InListExpr` and `NestedPredicate` evaluate every later candidate over the whole batch and only stop once every row has matched, where Spark stops per row. So this PR does not add the behavior. A CASE candidate now goes through the same loop instead of being read as a constant NULL. That happened to be right for this shape and wrong for others. `id IN (0L, CASE WHEN id = 0 THEN CAST(s AS BIGINT) WHEN id = 1 THEN 1L END)` gave NULL for `id = 1` where Spark returns true. Could you open an issue for the whole-batch evaluation of later `IN` candidates, link it from #6006, and reference it here? I'd rather track it than grow this PR into a per-row `IN` evaluator, but Spark 4 runs with ANSI on by default, so I don't want it lost. -- 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]
