sunchao commented on code in PR #7261:
URL: https://github.com/apache/datafusion-comet/pull/7261#discussion_r4394079733
##########
spark/src/main/scala/org/apache/comet/serde/datetime.scala:
##########
@@ -436,6 +436,85 @@ object CometNextDay extends CometExpressionSerde[NextDay]
with CodegenDispatchFa
object CometMakeDate extends CometExpressionSerde[MakeDate] {
+ private val sharedArgumentReason =
+ "make_date requires Spark evaluation when a computed argument after a
nullable " +
+ "argument cannot preserve Spark's shared-subexpression evaluation."
+
+ override def getUnsupportedReasons(): Seq[String] = Seq(sharedArgumentReason)
+
+ override def getSupportLevel(expr: MakeDate): SupportLevel = {
+ val firstNullable = expr.children.indexWhere(_.nullable)
+ val hasComputedMaskedArgument = firstNullable >= 0 &&
+ expr.children.drop(firstNullable +
1).exists(!_.isInstanceOf[LeafExpression])
+ if (hasComputedMaskedArgument && !canShortCircuitNulls(expr)) {
+ Unsupported(Some(sharedArgumentReason))
+ } else {
+ Compatible()
+ }
+ }
+
+ // Native IN evaluates dynamic candidates for the whole batch. Only these
constructors and
+ // leaves are safe on rows where Spark skips a candidate after a match or a
NULL probe.
+ private def canEvaluateInCandidateEagerly(expr: Expression): Boolean = expr
match {
+ case _: Literal | _: Attribute => true
+ case _: CreateArray | _: CreateNamedStruct =>
+ expr.children.forall(canEvaluateInCandidateEagerly)
+ case _ => false
Review Comment:
[P2] Admit nullability-only casts around safe IN candidates
Could this recognize casts that only widen nested nullability around an
otherwise safe candidate? With nullable `y`, Spark coerces the literal struct
candidates in `(make_date(y, 1, 1), 0D) IN ((DATE '2024-01-01', 0D), (DATE
'2025-01-01', 0D))` into `Cast(CreateNamedStruct(...), ...)` so their field
nullability matches the probe. These casts cannot introduce an evaluation
error, but the catch-all rejects them and forces the whole projection back to
Spark. This is the failure in the new `make_date_ansi.sql:91` test in the
[current expression
job](https://github.com/apache/datafusion-comet/actions/runs/40561224738/job/122427367686),
and a Spark 4.1.3 analyzer probe confirms the inserted casts. The hand-built
serde tests bypass this coercion. Please preserve native execution for these
harmless wrappers while retaining fallback for casts that can throw.
##########
spark/src/main/scala/org/apache/comet/serde/datetime.scala:
##########
@@ -436,6 +436,85 @@ object CometNextDay extends CometExpressionSerde[NextDay]
with CodegenDispatchFa
object CometMakeDate extends CometExpressionSerde[MakeDate] {
+ private val sharedArgumentReason =
+ "make_date requires Spark evaluation when a computed argument after a
nullable " +
+ "argument cannot preserve Spark's shared-subexpression evaluation."
+
+ override def getUnsupportedReasons(): Seq[String] = Seq(sharedArgumentReason)
+
+ override def getSupportLevel(expr: MakeDate): SupportLevel = {
+ val firstNullable = expr.children.indexWhere(_.nullable)
+ val hasComputedMaskedArgument = firstNullable >= 0 &&
+ expr.children.drop(firstNullable +
1).exists(!_.isInstanceOf[LeafExpression])
+ if (hasComputedMaskedArgument && !canShortCircuitNulls(expr)) {
+ Unsupported(Some(sharedArgumentReason))
+ } else {
+ Compatible()
+ }
+ }
+
+ // Native IN evaluates dynamic candidates for the whole batch. Only these
constructors and
+ // leaves are safe on rows where Spark skips a candidate after a match or a
NULL probe.
+ private def canEvaluateInCandidateEagerly(expr: Expression): Boolean = expr
match {
+ case _: Literal | _: Attribute => true
+ case _: CreateArray | _: CreateNamedStruct =>
+ expr.children.forall(canEvaluateInCandidateEagerly)
+ case _ => false
+ }
+
+ /** A native parent must preserve Spark's evaluation of a date and its
sibling expressions. */
+ private[serde] def unsupportedParentReason(expr: Expression): Option[String]
= {
+ var pending: List[(Expression, Option[String])] = (expr, None) :: Nil
+ while (pending.nonEmpty) {
+ val (current, parent) = pending.head
+ pending = pending.tail
+ if (current.isInstanceOf[MakeDate] && parent.isDefined) {
+ return Some(s"make_date requires Spark evaluation under ${parent.get}")
+ }
+ def pushChildren(mask: Option[String]): Unit = {
+ current.children.reverseIterator.foreach(child => pending = (child,
mask) :: pending)
+ }
+ if (parent.isDefined) {
+ pushChildren(parent)
+ } else {
+ current match {
+ // These native expressions already select the rows passed to their
children.
+ case _: If | _: CaseWhen | _: MakeDate => pushChildren(None)
+ // COALESCE lowers to CASE and evaluates an earlier child again in
its result branch.
+ case coalesce: Coalesce if coalesce.deterministic =>
pushChildren(None)
+ // These parents evaluate every child in Spark, including NULL
values.
+ case _: EqualNullSafe | _: CreateArray | _: CreateNamedStruct =>
+ pushChildren(None)
+ // These parents evaluate left first and skip right only when left
is NULL.
+ case _: BinaryComparison | _: DateAdd | _: DateSub | _: DateDiff |
_: DateFormatClass =>
+ val left = current.children.head
+ val right = current.children(1)
+ val rightMask = if (left.nullable) Some(current.nodeName) else None
+ pending = (left, None) :: (right, rightMask) :: pending
+ case _: SortOrder => pushChildren(Some(current.nodeName))
+ // Empty membership can skip its sole operand in Spark.
+ case in: In if in.list.isEmpty =>
pushChildren(Some(current.nodeName))
+ case inSet: InSet if inSet.hset.isEmpty =>
pushChildren(Some(current.nodeName))
+ // Nonempty IN always evaluates its probe, but may skip candidates
on individual rows.
+ case in: In =>
+ in.list.reverseIterator.foreach { child =>
+ pending = (child, Some(in.nodeName)) :: pending
+ }
+ val probeParent = if
(in.list.forall(canEvaluateInCandidateEagerly)) {
+ None
+ } else {
+ Some("In with computed candidates")
+ }
+ pending = (in.value, probeParent) :: pending
+ case _ if current.children.size <= 1 => pushChildren(None)
+ // Other native multi-child expressions may evaluate operands that
Spark skips.
+ case _ => pushChildren(Some(current.nodeName))
Review Comment:
[P2] Preserve native execution for always-eager parents
Could `Greatest` and `Least` be handled as eager parents? For example,
`greatest(make_date(y, m, 1), DATE '2024-01-01')` now returns `make_date
requires Spark evaluation under Greatest` and sends the entire projection back
to Spark. Spark evaluates every operand of both functions in interpreted and
generated evaluation, so there is no skipped date operand to protect. Both
functions already support DateType natively on the PR base and branch-1.2. The
new guard also rejects these shapes in a Spark 4.1.3 analyzer probe. Please
admit these parents and add native-plan regression coverage.
##########
native/spark-expr/src/datetime_funcs/make_date.rs:
##########
@@ -17,15 +17,85 @@
use arrow::array::{Array, Date32Array, Int32Array};
use arrow::compute::cast;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, FieldRef, Schema};
+use arrow::record_batch::RecordBatch;
use datafusion::common::{utils::take_function_args, DataFusionError, Result};
use datafusion::logical_expr::{
ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
};
+use datafusion::physical_expr::PhysicalExpr;
+use std::fmt::{Display, Formatter};
use std::sync::Arc;
use crate::{conversion_funcs::ymd_to_epoch_day, SparkError};
+/// Skips the entire date expression on empty input, before any argument can
raise an error.
+/// `inner` retains the planner's argument evaluation, including NULL
short-circuiting when enabled.
+#[derive(Debug, Eq, Hash)]
Review Comment:
[P2] Implement Hash explicitly alongside the manual equality
Could this use an explicit `Hash` implementation, as `NullShortCircuit`
does? Deriving `Hash` while implementing `PartialEq` manually triggers Clippy's
default-deny `derived_hash_with_manual_eq` lint. The [Rust job for this
head](https://github.com/apache/datafusion-comet/actions/runs/40561224738/job/122408703995)
fails here for both the library and test target, so the subsequent native
validation never runs. Delegating `hash` to `self.inner.hash(state)` alongside
the existing equality follows the local wrapper pattern.
--
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]
