comphead commented on code in PR #7456:
URL: https://github.com/apache/datafusion-comet/pull/7456#discussion_r4460105300


##########
native/core/src/execution/planner.rs:
##########
@@ -1541,23 +1523,36 @@ impl PhysicalPlanner {
                         None,
                         eval_mode,
                     )?;
-                    let scalar_expr = Arc::new(ScalarFunctionExpr::new(
-                        op_str,
-                        fun_expr,
-                        vec![left, right],
-                        Arc::new(Field::new(op_str, data_type, true)),
-                        Arc::new(ConfigOptions::default()),
-                    ));
-
-                    if query_context.is_some() {
-                        Ok(Arc::new(CheckedBinaryExpr::new(scalar_expr, 
query_context)))
-                    } else {
-                        Ok(scalar_expr)
-                    }
+                    (
+                        Arc::new(ScalarFunctionExpr::new(
+                            op_str,
+                            fun_expr,
+                            vec![left, right],
+                            Arc::new(Field::new(op_str, data_type, true)),
+                            Arc::new(ConfigOptions::default()),
+                        )),
+                        true,
+                    )
                 } else {
-                    Ok(Arc::new(BinaryExpr::new(left, op, right)))
+                    (Arc::new(BinaryExpr::new(left, op, right)), false)
                 }
             }
+        };
+        let expr = if options.null_short_circuit {
+            // Spark evaluates the divisor of a division first
+            let order: &[usize] = if op == DataFusionOperator::Divide {
+                &[1, 0]
+            } else {
+                &[0, 1]
+            };
+            NullShortCircuit::wrap_in_order(expr, order, &input_schema)?
+        } else {
+            expr
+        };

Review Comment:
   I think two things change for decimal operands once this wrapper is in the 
tree. I have not run it, this is from reading the code.
   
   1. `DecimalPrecision.promote` puts every decimal `+`, `-`, `*`, `/` and `%` 
under `CheckOverflow`. The `CheckOverflow` arm skips the check for a 
`WideDecimalBinaryExpr` (#3619), but `unwrap_checked` only looks through 
`CheckedBinaryExpr`. For 
`CheckedBinaryExpr(NullShortCircuit(WideDecimalBinaryExpr))`, `is_wide_decimal` 
is now false and the extra `CheckOverflow` pass comes back. `CAST(n AS 
DECIMAL(38, 0)) + CAST(txt AS DECIMAL(38, 0))` in the new fixture takes that 
path.
   2. A decimal operand reaches `NullShortCircuit` as a `CheckOverflow`, which 
is neither in `cannot_fail` nor in `is_deterministic`. So `price * (1 - 
discount)` is wrapped even in LEGACY mode, where Spark returns NULL on overflow 
rather than failing, and speculation never applies to it. A batch with NULL 
prices then takes the selection path that the description measured at up to 2x 
for `BIGINT`.
   
   Could `unwrap_checked` look through `NullShortCircuit`, and 
`is_deterministic` accept `CheckOverflow`? A decimal shape in the new benchmark 
would show whether that is enough.



##########
spark/src/test/resources/sql-tests/expressions/math/arithmetic_ansi_skipped_operands.sql:
##########
@@ -0,0 +1,124 @@
+-- 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.
+
+-- Spark evaluates the second operand of an arithmetic operator only for the 
rows where the first
+-- one is not NULL, so an operand that fails under ANSI mode must not fail on 
the rows Spark skips
+-- (https://github.com/apache/datafusion-comet/issues/7246). Every table is 
one Parquet file, so its
+-- rows arrive in one batch.
+
+-- Config: spark.sql.ansi.enabled=true
+-- Config: spark.sql.adaptive.enabled=false
+
+-- `txt` is malformed only on the rows where `n` is NULL
+statement
+CREATE TABLE skipped(id int, n int, txt string) USING parquet
+
+statement
+INSERT INTO skipped SELECT /*+ REPARTITION(1) */ * FROM VALUES
+  (1, CAST(NULL AS INT), 'bad'), (2, 3, '6'), (3, CAST(NULL AS INT), 'x'), (4, 
2, '7')
+  AS t(id, n, txt)
+
+-- `n` is NULL on every row
+statement
+CREATE TABLE all_null(id int, n int, txt string) USING parquet
+
+statement
+INSERT INTO all_null SELECT /*+ REPARTITION(1) */ * FROM VALUES
+  (1, CAST(NULL AS INT), 'bad'), (2, CAST(NULL AS INT), '2') AS t(id, n, txt)
+
+-- `n` is not NULL, so Spark evaluates every operand
+statement
+CREATE TABLE filled(id int, n int, txt string) USING parquet
+
+statement
+INSERT INTO filled SELECT /*+ REPARTITION(1) */ * FROM VALUES (1, 1, 'bad'), 
(2, 2, '2') AS t(id, n, txt)
+
+-- Add, subtract and multiply evaluate the right operand only where the left 
one is not NULL

Review Comment:
   Could this file also cover the other paths the planner now wraps?
   
   - `TINYINT` and `SMALLINT`, which use the `checked_*` kernels on `Int8` and 
`Int16`
   - `DOUBLE` for `+`, `-` and `*`, which stays a plain `BinaryExpr`
   - `try_add` and `try_divide`, which Spark plans as `EvalMode.TRY` arithmetic
   - a zero divisor under ANSI. Spark returns NULL for a NULL dividend over 
`0`, and for `CAST(txt AS INT) DIV n` with `txt = 'bad'` and `n = 0` it raises 
`CAST_INVALID_INPUT` rather than `DIVIDE_BY_ZERO`, because `DivModLike` 
evaluates the dividend before its zero check.
   
   `null_short_circuit_skips_the_dividend_where_the_divisor_is_null` in 
`modulo_expr.rs` repeats the `DECIMAL(38, 10) % DECIMAL(38, 0)` query in this 
file, so it could go.



##########
native/spark-expr/src/math_funcs/modulo_expr.rs:
##########
@@ -155,6 +155,28 @@ pub fn create_modulo_expr(
     input_schema: SchemaRef,
     fail_on_error: bool,
     registry: &dyn FunctionRegistry,
+) -> Result<Arc<dyn PhysicalExpr>, DataFusionError> {
+    create_modulo_expr_with_null_short_circuit(
+        left,
+        right,
+        data_type,
+        input_schema,
+        fail_on_error,
+        false,
+        registry,
+    )
+}
+
+/// Like [`create_modulo_expr`], but with `null_short_circuit` evaluates the 
dividend only for the
+/// rows where the divisor, which Spark evaluates first, is not NULL. See 
[`NullShortCircuit`].
+pub fn create_modulo_expr_with_null_short_circuit(

Review Comment:
   After this change `RemainderBuilder` calls 
`create_modulo_expr_with_null_short_circuit`, so `create_modulo_expr` has no 
caller left outside its tests. The PR removes `create_binary_expr` for the same 
reason. Could `create_modulo_expr` take the `null_short_circuit` flag itself, 
so the crate keeps one public function instead of two?



##########
native/spark-expr/src/conditional_funcs/null_short_circuit.rs:
##########
@@ -269,6 +319,12 @@ impl Arg {
         let Some((projection, projected)) = &self.masked else {
             return self.expr.evaluate(batch);
         };
+        if nulls.is_some() && self.speculate.load(Ordering::Relaxed) {
+            if let Ok(value) = evaluate_projected(projection, projected, 
batch, None) {
+                return Ok(value);
+            }
+            self.speculate.store(false, Ordering::Relaxed);
+        }

Review Comment:
   `LazyLogical` only speculates when more than half of the rows need the right 
operand (`pending_rows * 2 > num_rows` in `lazy_logical.rs`). Here any NULL is 
enough, so when nearly every row is skipped the masked argument runs over the 
whole batch. That also applies to the existing users from #6716, such as 
`array_contains`, `slice`, `make_date` and `GetMapValue`, which on `main` 
evaluate an argument like `CAST(s AS INT)` over only the rows that need it. 
Could this use the same rule, for example `nulls.null_count() * 2 < 
batch.num_rows()`, and could the benchmark add a point at 90% NULL? I have not 
measured it.
   
   The struct doc also still says the function sees NULL for each skipped 
argument. With speculation it sees the computed value there instead, so the 
function must also not fail on a row where an earlier argument is NULL, 
whatever the later ones hold. `SparkArraySlice` and `SparkMakeDate` check for 
NULL first, so I do not see a current bug, but could the doc here and the 
`withNullShortCircuit` Scaladoc say so, so that a new kernel does not validate 
first?



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

Reply via email to