coderfender commented on code in PR #7456:
URL: https://github.com/apache/datafusion-comet/pull/7456#discussion_r4571923256
##########
native/spark-expr/src/conditional_funcs/case_when.rs:
##########
@@ -443,7 +443,8 @@ pub(super) fn is_infallible(expr: &Arc<dyn PhysicalExpr>,
input_schema: &Schema)
/// Whether `expr` can be evaluated for rows that Spark would not evaluate it
for, at any cost: what
/// [`is_infallible`] accepts, and casts that [`Cast::cannot_fail`] accepts
beyond it, such as a
-/// LEGACY or TRY cast from a string to an integer, which returns NULL for a
string it cannot parse.
+/// LEGACY or TRY cast from a string to an integer, which returns NULL for a
string it cannot parse,
+/// or a cast from a number to a string.
Review Comment:
Following up on Andy's point about the LEGACY decimal shape. `p * (1.00 -
d)` is still wrapped in LEGACY mode (1.14x slower than eager at 50% NULL in
your table), but nothing in it can raise. A `CheckOverflow` with `fail_on_error
= false` returns NULL on overflow instead. Could `infallible` below accept it?
Something like this before the `NormalizeNaNAndZero` arm, with `CheckOverflow`
added to the `use crate::{...}` at the top:
```rust
} else if let Some(check) = expr.downcast_ref::<CheckOverflow>() {
// A CheckOverflow that does not fail on error returns NULL on
overflow
!check.fail_on_error
```
On its own that isn't enough, because the child `1.00 - d` is a decimal
`BinaryExpr`, and `binary_is_infallible` only accepts integer and floating
point `+`, `-` and `*`. Could you check whether the decimal arithmetic the
planner emits outside ANSI can raise, and if it can't, accept it there too? If
you'd rather keep wrapping the decimal case, a sentence in the code saying why
would help.
##########
spark/src/main/scala/org/apache/comet/serde/strings.scala:
##########
@@ -393,6 +393,13 @@ object CometConcat
override def getIncompatibleReasons(): Seq[String] = Seq(collationReason)
+ // Spark's generated code for Concat evaluates each argument only where
every earlier one is not
+ // NULL, and the native concat returns NULL wherever an argument is NULL.
+ override def convert(expr: Concat, inputs: Seq[Attribute], binding:
Boolean): Option[Expr] = {
+ val proto = super.convert(expr, inputs, binding)
+ if (expr.children.length > 1) withNullShortCircuit(expr, proto) else proto
Review Comment:
I said a follow-up was fine for the string functions, but Andy's numbers
change my mind. `concat(q, upper(s))` is 1.28x slower than the release at 50%
NULL and 1.17x at 70%, in every eval mode, and `upper` can't fail. That's an
existing query getting slower with nothing gained, so I'd rather it not ship in
this PR. Could the string functions you measured (`upper`, `lower`, `trim`,
`substring` and similar) go into `cannot_fail` here, so a `concat` over them
isn't wrapped at all, and #7587 can cover the rest?
##########
native/spark-expr/src/conditional_funcs/null_short_circuit.rs:
##########
@@ -234,49 +293,160 @@ impl PhysicalExpr for NullShortCircuit {
}
}
+/// How an argument is evaluated, decided when it is planned.
+struct ArgPlan {
+ /// See [`Arg::field`]
+ field: Option<FieldRef>,
+ /// Whether the argument is evaluated for only the rows where no earlier
argument is NULL
+ masked: bool,
+ /// See [`Masked::cheap`]
+ cheap: bool,
+}
+
/// An argument, and how it is evaluated.
#[derive(Debug)]
struct Arg {
expr: Arc<dyn PhysicalExpr>,
/// The field of the column that `body` reads the argument from, or `None`
for a literal,
/// which `body` keeps
field: Option<FieldRef>,
- /// For an argument evaluated for only the rows where no earlier argument
is NULL, the input
- /// columns it reads, and the argument reading them from a batch of just
those columns, so
- /// that skipping rows filters no other column. It is evaluated in place
of `expr` for every
- /// batch, even one with no NULL, because rebuilding the argument gave it
its own copy of any
- /// state kept across batches, such as the random generator of `shuffle`
- masked: Option<(Vec<usize>, Arc<dyn PhysicalExpr>)>,
+ /// The argument, for one evaluated for only the rows where no earlier
argument is NULL
+ masked: Option<Masked>,
}
impl Arg {
- fn try_new(
- expr: &Arc<dyn PhysicalExpr>,
- field: Option<FieldRef>,
- masked: bool,
- ) -> Result<Self> {
- let masked = masked.then(|| project(expr)).transpose()?;
- Ok(Self {
+ fn new(expr: &Arc<dyn PhysicalExpr>, plan: ArgPlan) -> Self {
+ Self {
expr: Arc::clone(expr),
- field,
- masked,
- })
+ field: plan.field,
+ masked: plan
+ .masked
+ .then(|| Masked::new(Arc::clone(expr), plan.cheap)),
+ }
}
/// Evaluates the argument, for just the rows outside `nulls` if it is
masked, returning a
/// value for every row of the batch.
fn evaluate(&self, batch: &RecordBatch, nulls: Option<&NullBuffer>) ->
Result<ColumnarValue> {
- let Some((projection, projected)) = &self.masked else {
- return self.expr.evaluate(batch);
+ match &self.masked {
+ Some(masked) => masked.evaluate(batch, nulls),
+ None => self.expr.evaluate(batch),
+ }
+ }
+}
+
+/// An expression evaluated for only some rows of each batch, as an argument of
+/// [`NullShortCircuit`] or the right operand of [`super::LazyLogical`].
+#[derive(Debug)]
+pub(super) struct Masked {
+ expr: Arc<dyn PhysicalExpr>,
+ /// The input columns that `expr` reads, and `expr` reading them from a
batch of just those
+ /// columns, so that skipping rows filters no other column. It is
evaluated in place of `expr`
+ /// for every batch, even one where every row needs it, because rebuilding
the expression gave
+ /// it its own copy of any state kept across batches, such as the random
generator of
+ /// `shuffle`. It is built when first evaluated rather than when planned:
rebuilding an
+ /// expression plans its children again, so building it eagerly would
rebuild each nested
+ /// [`NullShortCircuit`] or [`super::LazyLogical`] once for every level
above it, which doubles
+ /// the work at each level.
+ projected: OnceLock<(Vec<usize>, Arc<dyn PhysicalExpr>)>,
+ /// Whether to first try the expression over the whole batch, as it would
be without masking,
+ /// for a batch where at least half the rows need it, or for any batch if
it is
+ /// [`Masked::cheap`]. Selecting the rows costs more than evaluating the
skipped ones when they
+ /// are few, and less when they are many, and the crossover moves with the
cost of the
+ /// expression, so half is where neither choice is much slower than the
other for the costlier
+ /// expressions measured in `benches/null_short_circuit.rs`. Whatever it
returns for a skipped
+ /// row is ignored, so it only matters if it fails there, and it requires
an expression that
+ /// returns the same for a row whichever rows it sees. If it fails, the
rows are selected
+ /// instead, here and for the batches after it.
+ speculate: AtomicBool,
+ /// Whether the expression is so cheap that evaluating it over the whole
batch is about as fast
+ /// as selecting the rows that need it, however few they are, as
[`is_cheap`] decides
+ cheap: bool,
+}
+
+impl Masked {
+ pub(super) fn new(expr: Arc<dyn PhysicalExpr>, cheap: bool) -> Self {
+ let speculate = AtomicBool::new(is_deterministic(&expr));
+ Self {
+ expr,
+ projected: OnceLock::new(),
+ speculate,
+ cheap,
+ }
+ }
+
+ /// Evaluates the expression for just the valid rows of `rows`, or for
every row if it is
+ /// `None`, returning a value for every row of the batch.
+ pub(super) fn evaluate(
+ &self,
+ batch: &RecordBatch,
+ rows: Option<&NullBuffer>,
+ ) -> Result<ColumnarValue> {
+ let (projection, projected) = self.projected()?;
+ let Some(rows) = rows else {
+ return evaluate_projected(projection, projected, batch, None);
};
- let rows = nulls.map(|nulls| BooleanArray::new(nulls.inner().clone(),
None));
- evaluate_projected(projection, projected, batch, rows.as_ref())
+ let needed = rows.len() - rows.null_count();
+ if (self.cheap || needed * 2 >= rows.len()) &&
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:
Andy asked about this too. Once one whole-batch attempt fails, `speculate`
stays off for every later batch in the task, even for a cheap argument where
almost every row needs it. So one malformed skipped row early on makes the rest
of the task select rows. Is that the intended trade-off? If it is, a short note
on the `speculate` field would help the next reader. If not, would retrying
after some number of batches be worth measuring?
--
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]
