tlfua commented on code in PR #7456:
URL: https://github.com/apache/datafusion-comet/pull/7456#discussion_r4461139044
##########
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:
Thanks, measured it before picking a cutoff. The benchmark now also runs at
70% and 90% NULL, has a decimal shape, and I ran it on a GitHub runner always
speculating and never speculating
([run](https://github.com/tlfua/datafusion-comet/actions/runs/41721000607),
from the measurement-only branch
https://github.com/tlfua/datafusion-comet/tree/bench/7246, which is not part of
this PR). The table is in the PR description.
Where selecting the rows starts to win depends on how much the argument
costs. For `a + CAST(s AS BIGINT)` it is already faster at 50% NULL (65 vs 77
µs). For `a * (b - c)` it only wins at 90% (7.9 vs 8.6 µs), and at 50% it is
1.8x slower (20.2 vs 11.2 µs). So `null_count * 2 < num_rows` would cost the
cheap arithmetic 1.8x at exactly half NULL.
I went with `null_count * 2 <= num_rows`, the same split as `LazyLogical`.
For the three shapes it is at most 1.2x off the better choice up to half NULL
and 1.4x above it, and at 90% NULL it is 4.5x faster than always speculating
for the cast. A batch above the cutoff selects the rows without turning
speculation off, so later batches still speculate. That also keeps the existing
`NullShortCircuit` users from #6716 on the old path for mostly-NULL batches.
The doc now says the function must return NULL wherever an earlier argument
is NULL without checking the later ones, both on `NullShortCircuit` and on
`withNullShortCircuit`.
##########
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:
You're right on both, thanks for reading it that closely.
1. `unwrap_checked` now also looks through `NullShortCircuit`, so the
`CheckOverflow` over a wide decimal add, subtract or multiply is dropped again.
`test_check_overflow_skipped_over_short_circuited_wide_decimal` in `planner.rs`
plans `CheckOverflow(a + CAST(s AS DECIMAL(38, 0)))` under ANSI and checks that
the result is the wrapped `WideDecimalBinaryExpr` with no `CheckOverflow`, and
that it returns NULL for the row where `a` is NULL and `s` is malformed.
2. `is_deterministic` now accepts `CheckOverflow`, and a `NullShortCircuit`
whose inner expression is deterministic, so decimal and nested operands
speculate. `decimal_and_nested_short_circuited_arguments_are_speculated` covers
both, and the benchmark has a decimal shape now.
I did not add decimal arithmetic or `CheckOverflow` to `cannot_fail`,
because I can't show that the decimal kernels never return an error. So a
LEGACY decimal operand is still wrapped, but with speculation it costs about
the same as unwrapped (decimal rows in the table).
##########
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:
Added all of them:
- `TINYINT` `+` and `%`, `SMALLINT` `*` and `DIV`
- `DOUBLE` `+`, `-` and `*`
- `try_add` and `try_divide` over skipped rows, and `try_add` with a
malformed operand that Spark does evaluate, which still raises
`CAST_INVALID_INPUT`
- a zero divisor: `DIV`, `%` and `/` return NULL for a NULL dividend over
`0`, and `CAST(txt AS INT) DIV n` with `txt = 'bad'`, `n = 0` raises
`CAST_INVALID_INPUT`
Comet matches Spark on all of them. I also removed the duplicate test in
`modulo_expr.rs`.
##########
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:
Done. `create_modulo_expr` now takes `null_short_circuit: bool`,
`create_modulo_expr_with_null_short_circuit` is gone, and the tests pass
`false`.
--
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]
