diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index a2945aa95f2..582cc900f3c 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -35,6 +35,7 @@ - Spark 4.0.1 (audited 2026-05-27): now `RuntimeReplaceable` and rewritten to `ArrayInsert(arr, Literal(-1), elem)`. `CometArrayAppend` is therefore unreachable; dispatch goes through `CometArrayInsert` (which carries its own `Incompatible` notes documented at the `array_insert` entry). - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1. - Spark 3.4, 3.5 (2026-10-15, [#6086](https://github.com/apache/datafusion-comet/issues/6086)): Spark disagrees with itself here. Interpreted `ArrayAppend.eval` returns NULL for a NULL array without evaluating the item, but `doGenCode` emits the item's code outside `nullSafeExec`, so generated code evaluates the item on every row, including rows whose array is `NULL`, and an item that raises under ANSI raises there. Comet follows the generated code, which is what runs under whole-stage codegen. Its `CASE WHEN IS NOT NULL THEN array_append(...) ELSE null` guard skipped the item on those rows; the `ELSE` branch now evaluates it too, when it is not a literal or a column read, and returns a NULL typed as the list. A nondeterministic array or item with a nullable array falls back to Spark, because the guard serializes the operands more than once. A non-nullable array gets no guard at all, so each operand is evaluated once over the whole batch. +- Spark 3.4, 3.5 (2026-10-21, [#5867](https://github.com/apache/datafusion-comet/pull/5867)): a nondeterministic operand over a nullable array now runs through the JVM codegen dispatcher instead of falling back to Spark, unless the result holds a calendar interval or an argument the dispatched tree can skip holds a subexpression the operator shares; those still fall back. A non-nullable array stays native whatever its operands. ## array_compact diff --git a/docs/source/contributor-guide/sql-file-tests.md b/docs/source/contributor-guide/sql-file-tests.md index 78bb8806c95..196b2e1d30c 100644 --- a/docs/source/contributor-guide/sql-file-tests.md +++ b/docs/source/contributor-guide/sql-file-tests.md @@ -479,8 +479,9 @@ common ones include: - **Signed zero** -- Spark parses a bare `-0.0` as `decimal(1,1)`, which has no signed zero, so coercion to `float`/`double` yields `+0.0`. `CAST(-0.0 AS DOUBLE)` and `CAST(-0.0 AS FLOAT)` have the same problem because the cast source is still the - decimal literal. Use `double('-0.0')` or `float('-0.0')` (equivalently - `CAST('-0.0' AS DOUBLE)`). Spark's array comparator also treats `+0.0` and `-0.0` as + decimal literal. Use the double literal `-0.0D`, `CAST(-0.0D AS FLOAT)` for a float, + or `double('-0.0')` / `float('-0.0')` (equivalently `CAST('-0.0' AS DOUBLE)`), all of + which keep the sign. Spark's array comparator also treats `+0.0` and `-0.0` as equal, so `sort_array(...)` is not a unique projection when both signs are present (the SQL test comparator distinguishes the bits). Prefer a sign-aware form such as `sort_array(transform(arr, x -> cast(x AS string)))`. A `query tolerance=...` check diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index f9292b27343..fd833dc15bc 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -154,7 +154,7 @@ The tables below list every Spark built-in expression with its current status. | Function | Status | Implementation | Notes | | --- | --- | --- | --- | | `array` | ✅ | Native | | -| `array_append` | ✅ | Native | | +| `array_append` | ✅ | Hybrid | A nondeterministic operand over a nullable array routes through the JVM codegen dispatcher | | `array_compact` | ✅ | — | | | `array_contains` | ✅ | Hybrid | Flat float/double element arrays use a native kernel with Spark's equality; nested float elements, non-default string collations, and a value that is not a literal or column over a nullable array, route through the JVM codegen dispatcher | | `array_distinct` | ✅ | Native | Floating-point elements fall back on Spark versions before 4.0.5, 4.1.4, and 4.2; signed-zero results may differ with native opt-in ([details](compatibility/floating-point.md)). Non-default string collations fall back | @@ -170,8 +170,8 @@ The tables below list every Spark built-in expression with its current status. | `array_repeat` | ✅ | Native | | | `array_union` | ✅ | Native | Floating-point elements fall back on Spark versions before 4.0.5, 4.1.4, and 4.2; signed-zero results may differ with native opt-in ([details](compatibility/floating-point.md)). Non-default string collations fall back | | `arrays_overlap` | ✅ | Hybrid | Non-default string collations use the JVM codegen dispatcher | -| `arrays_zip` | ✅ | Native | | -| `element_at` | ✅ | Native | | +| `arrays_zip` | ✅ | Hybrid | | +| `element_at` | ✅ | Hybrid | | | `flatten` | ✅ | Native | Binary/struct/map elements fall back | | `get` | ✅ | — | | | `sequence` | ✅ | Hybrid | Integral types run natively; date/timestamp sequences use codegen dispatch | @@ -205,10 +205,10 @@ The tables below list every Spark built-in expression with its current status. | Function | Status | Implementation | Notes | | --- | --- | --- | --- | | `array_size` | ✅ | — | | -| `cardinality` | ✅ | Native | | +| `cardinality` | ✅ | Hybrid | | | `concat` | ✅ | Hybrid | Binary/array children and non-UTF8_BINARY collations route through the JVM codegen dispatcher | | `reverse` | ✅ | Hybrid | Binary input (Spark 4.2), arrays with binary, struct, or map elements, and collated strings, route through the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | -| `size` | ✅ | Native | | +| `size` | ✅ | Hybrid | | --- @@ -404,12 +404,12 @@ to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enab | Function | Status | Implementation | Notes | | --- | --- | --- | --- | -| `element_at` | ✅ | Native | | +| `element_at` | ✅ | Hybrid | | | `map` | ✅ | Codegen dispatch | Routed through the JVM codegen dispatcher | | `map_concat` | ✅ | Codegen dispatch | | | `map_contains_key` | ✅ | — | | | `map_entries` | ✅ | Native | | -| `map_from_arrays` | ✅ | Hybrid | `FLOAT` or `DOUBLE` keys and `spark.sql.mapKeyDedupPolicy=LAST_WIN` route through the JVM codegen dispatcher ([details](compatibility/expressions/map.md)) | +| `map_from_arrays` | ✅ | Hybrid | A nondeterministic operand, `FLOAT` or `DOUBLE` keys, and `spark.sql.mapKeyDedupPolicy=LAST_WIN` route through the JVM codegen dispatcher ([details](compatibility/expressions/map.md)) | | `map_from_entries` | ✅ | Hybrid | BinaryType keys/values, `FLOAT` or `DOUBLE` keys and `spark.sql.mapKeyDedupPolicy=LAST_WIN` route through the JVM codegen dispatcher ([details](compatibility/expressions/map.md)) | | `map_keys` | ✅ | Native | | | `map_values` | ✅ | Native | | diff --git a/pom.xml b/pom.xml index 292fc21d702..40114c59eab 100644 --- a/pom.xml +++ b/pom.xml @@ -982,6 +982,7 @@ under the License. -Wconf:cat=unused-params&site=org\.apache\.comet\.rules\.CometScanContrib\.tryTransformV[12]\..*:s -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CometExpressionSerde\.getSupportLevel\..*:s + -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CodegenDispatchFallback\.dispatchDeclineReason\..*:s -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CometAggregateExpressionSerde\.getSupportLevel\..*:s -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CometOperatorSerde\.getSupportLevel\..*:s -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CometAggregateExpressionSerde\.supportsSparkPartialToNativeFinal\..*:s @@ -1027,6 +1028,7 @@ under the License. -Wconf:cat=unused-params&site=org\.apache\.comet\.rules\.CometScanContrib\.tryTransformV[12]\..*:s -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CometExpressionSerde\.getSupportLevel\..*:s + -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CodegenDispatchFallback\.dispatchDeclineReason\..*:s -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CometAggregateExpressionSerde\.getSupportLevel\..*:s -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CometOperatorSerde\.getSupportLevel\..*:s -Wconf:cat=unused-params&site=org\.apache\.comet\.serde\.CometAggregateExpressionSerde\.supportsSparkPartialToNativeFinal\..*:s diff --git a/spark/src/main/scala/org/apache/comet/serde/CometExpressionSerde.scala b/spark/src/main/scala/org/apache/comet/serde/CometExpressionSerde.scala index 206bd7c8207..80f55967f6a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/CometExpressionSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/CometExpressionSerde.scala @@ -20,6 +20,9 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression} +import org.apache.spark.sql.types.CalendarIntervalType + +import org.apache.comet.serde.QueryPlanSerde.skipsSharedSubexpression /** * Trait for providing serialization logic for expressions. @@ -112,7 +115,8 @@ trait CometExpressionSerde[T <: Expression] { * `Unsupported`, the expression always routes through the dispatcher -- the serde is declaring * "no native path exists for this case; run Spark's code in-pipeline." Spark fallback is reserved * for the case where the dispatcher itself cannot handle the expression (e.g. the global codegen - * flag is off, or the kernel rejects the bound tree). + * flag is off, or the kernel rejects the bound tree), or an expression that + * `dispatchDeclineReason` declines. * * Contract for `Unsupported` reasons on a `CodegenDispatchFallback` serde: the case must be * something `Expression.doGenCode` can compile. If you mark something `Unsupported` because Spark @@ -122,7 +126,43 @@ trait CometExpressionSerde[T <: Expression] { * dispatcher. Every other `Incompatible` expression falls back to Spark, and every other * `Unsupported` expression falls back to Spark. */ -trait CodegenDispatchFallback extends NativeOptInAvailable { self: CometExpressionSerde[_] => } +trait CodegenDispatchFallback extends NativeOptInAvailable { self: CometExpressionSerde[_] => + + /** Why `expr` must fall back to Spark rather than run through the dispatcher, or `None`. */ + def dispatchDeclineReason(expr: Expression): Option[String] = None +} + +/** Sends the two cases below, which the dispatcher would get wrong, to Spark. */ +trait GatedCodegenDispatchFallback extends CodegenDispatchFallback { + self: CometExpressionSerde[_] => + + override def dispatchDeclineReason(expr: Expression): Option[String] = + CalendarIntervalOutput + .declineReason(expr) + .orElse(SharedSubexpressionArgument.declineReason(expr)) +} + +/** See https://github.com/apache/datafusion-comet/issues/5279. */ +private[serde] object CalendarIntervalOutput { + + private val reason: String = + "the result holds a calendar interval, whose microseconds overflow the dispatcher's Arrow " + + "nanoseconds beyond about 292 years" + + def declineReason(expr: Expression): Option[String] = + if (SupportLevel.containsType(expr.dataType, classOf[CalendarIntervalType])) Some(reason) + else None +} + +private[serde] object SharedSubexpressionArgument { + + private val reason: String = + "an argument this expression can skip holds a subexpression that the operator's other " + + "expressions share, which Spark evaluates for every row before this expression's NULL check" + + def declineReason(expr: Expression): Option[String] = + if (skipsSharedSubexpression(expr)) Some(reason) else None +} /** * Marker for serdes that have a native implementation the user can opt into. Normally these diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 7ee6b85fd05..4ba00c5af8c 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -1331,9 +1331,10 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { * an aggregation always, and in a projection, such as a project or a shuffle's partitioning * keys, only when `spark.sql.subexpressionElimination.enabled` is on. Other operators, such as * a filter or a sort, are treated like a projection, which at worst keeps them evaluated for - * every row where Spark does not. They are found on the first `canShortCircuitNulls` call for - * the operator, so an operator without an argument to skip never runs `EquivalentExpressions`. - * `None` outside `withSharedSubexpressions`, where they are unknown. + * every row where Spark does not, or sends to Spark an expression the dispatcher could run. + * They are found on the first `canShortCircuitNulls` call for the operator (or + * `skipsSharedSubexpression`), so an operator without an argument to skip never runs + * `EquivalentExpressions`. `None` outside `withSharedSubexpressions`, where they are unknown. */ private val sharedSubexpressions = new DynamicVariable[Option[SharedSubexpressions]](None) @@ -1341,14 +1342,22 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { // Find shared nodes in Spark's original trees, then compare them after decimal promotion. lazy val common: Seq[Expression] = if (alwaysShared || SQLConf.get.subexpressionEliminationEnabled) { - val equivalence = new EquivalentExpressions - expressions.foreach(equivalence.addExprTree(_)) - equivalence.getCommonSubexpressions.map(DecimalPrecision.promote) + commonSubexpressions(expressions) } else { Nil } } + /** + * The subexpressions that occur more than once in `expressions`, decimal-promoted so they + * compare equal to the promoted trees the serde converts. + */ + private def commonSubexpressions(expressions: Seq[Expression]): Seq[Expression] = { + val equivalence = new EquivalentExpressions + expressions.foreach(equivalence.addExprTree(_)) + equivalence.getCommonSubexpressions.map(DecimalPrecision.promote) + } + /** * Runs `f`, which serializes an operator whose expressions are `expressions`. Code that * serializes expressions outside an operator's conversion, such as when the plan runs, calls it @@ -1381,13 +1390,51 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { def canSkipEvaluation(skippable: Seq[Expression]): Boolean = { // Subexpression elimination never shares a leaf, such as a column or a literal skippable.forall(_.isInstanceOf[LeafExpression]) || sharedSubexpressions.value.exists { s => - // Avoid decimal promotion when Spark shares nothing. - s.common.isEmpty || !skippable.exists { expr => - DecimalPrecision.promote(expr).exists(e => s.common.exists(_.semanticEquals(e))) - } + !holdsSharedSubexpression(skippable, s) } } + /** + * Whether a node of `expr` can skip a child holding a subexpression the operator's expressions + * share. Spark evaluates that subexpression for every row, but the dispatcher, compiling `expr` + * alone, can skip it and so miss an ANSI error. While subexpression elimination is on, a + * subexpression `expr` itself repeats does not count, as the dispatcher eliminates it too. + * False outside `withSharedSubexpressions`. + */ + def skipsSharedSubexpression(expr: Expression): Boolean = + sharedSubexpressions.value.exists { s => + lazy val repeated = + if (SQLConf.get.subexpressionEliminationEnabled) commonSubexpressions(Seq(expr)) + else Nil + expr.exists { node => + val skippable = node match { + // the divisor is evaluated first, so only the dividend (`left`) can be skipped + case _: DivModLike | _: Pmod => node.children.take(1) + // `StringLocate` evaluates `start`, its last child, first and skips the others on NULL + case _: StringLocate => node.children.init + // `ArrayJoin` evaluates a nullable replacement first and skips the rest when it is NULL + case j: ArrayJoin if j.nullReplacement.exists(_.nullable) => node.children.init + // Spark 3.x codegen evaluates both children first; 4.0 rewrites it to `ArrayInsert` + case _: ArrayAppend => Nil + // Spark's codegen evaluates children in order and skips the rest after a NULL one + case _ => node.children.drop(1) + } + !skippable.forall(_.isInstanceOf[LeafExpression]) && + holdsSharedSubexpression(skippable, s, e => !repeated.exists(_.semanticEquals(e))) + } + } + + private def holdsSharedSubexpression( + exprs: Seq[Expression], + shared: SharedSubexpressions, + counts: Expression => Boolean = _ => true): Boolean = + // Avoid decimal promotion when Spark shares nothing. + shared.common.nonEmpty && exprs.exists { expr => + DecimalPrecision + .promote(expr) + .exists(e => shared.common.exists(_.semanticEquals(e)) && counts(e)) + } + /** * Has native execution evaluate the arguments of `scalarFunc`, the serialized `expr`, the way * Spark evaluates those of a null-intolerant `BinaryExpression` or `TernaryExpression`: each @@ -1424,7 +1471,15 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { inputs: Seq[Attribute], binding: Boolean): Option[(CodegenDispatchFallback, Expr)] = handler match { case h: CodegenDispatchFallback => - CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding).map(h -> _) + h.dispatchDeclineReason(expr) match { + case Some(reason) => + withFallbackReason( + expr, + s"${CometExplainInfo.exprDisplayName(expr)}: codegen dispatch: $reason") + None + case None => + CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding).map(h -> _) + } case _ => None } diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala b/spark/src/main/scala/org/apache/comet/serde/arrays.scala index 00018bae053..f20b6419fcd 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -60,7 +60,39 @@ object CometArrayRemove } } -object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBase { +/** + * The native NULL guard serializes a child twice, so a nondeterministic child is declined, + * nullable or not: a non-nullable one stays in step only through a DataFusion shortcut. + */ +private[serde] object NullGuardSupport { + + val nondeterministicReason: String = + "Comet has no native path for a nondeterministic operand such as `rand()` or " + + "`monotonically_increasing_id()`, because the native `NULL` guard would evaluate it twice." + + def nondeterministicChild(children: Seq[Expression]): Option[SupportLevel] = + children + .find(child => !child.deterministic) + .map(_ => Unsupported(Some(nondeterministicReason))) +} + +/** + * Whether evaluating `expr` out of Spark's order, or not at all, is unobservable. A literal or + * column read cannot throw, carry state or have a side effect. `foldable` is not enough: + * ConstantFolding leaves a throwing foldable expression in a conditional branch. + */ +private[serde] object EvaluationOrder { + + def isUnobservable(expr: Expression): Boolean = expr match { + case _: Literal | _: Attribute | _: BoundReference => true + case _ => false + } +} + +object CometArrayAppend + extends CometExpressionSerde[ArrayAppend] + with ArraysBase + with GatedCodegenDispatchFallback { /** * Spark's generated code for `ArrayAppend` evaluates the item for every row and guards only the @@ -74,20 +106,12 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBas * and evaluates each operand once over the whole batch, in order, whatever the operands are. */ private val nondeterministicReason: String = - "array_append with a nullable array and a nondeterministic operand such as `rand()`: the " + - "native NULL guard evaluates the operands more than once, on subsets of the rows" + "With a nullable array, Comet has no native path for a nondeterministic operand such as " + + "`rand()` or `monotonically_increasing_id()`, because the native `NULL` guard would " + + "evaluate it twice." override def getUnsupportedReasons(): Seq[String] = Seq(nondeterministicReason) - /** - * Whether evaluating `expr` for a row Spark would not guard it on is unobservable: a literal or - * column read cannot throw. `foldable` is not usable here, see CometArrayJoin.orderInsensitive. - */ - private def cannotThrow(expr: Expression): Boolean = expr match { - case _: Literal | _: Attribute | _: BoundReference => true - case _ => false - } - override def getSupportLevel(expr: ArrayAppend): SupportLevel = { if (expr.left.nullable && expr.children.exists(!_.deterministic)) { Unsupported(Some(nondeterministicReason)) @@ -146,7 +170,7 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBas // branch, so that it raises there as well, and still once per row. See // `nondeterministicReason` for why this is restricted to deterministic operands. val elseExprProto = - if (cannotThrow(expr.children(1))) { + if (EvaluationOrder.isUnobservable(expr.children(1))) { nullLiteralProto } else { for { @@ -211,13 +235,6 @@ object CometArrayContains case _ => false } - // Whether evaluating `expr` earlier than Spark would is unobservable: a literal or column read - // cannot throw. See CometArrayJoin.orderInsensitive. - private def orderInsensitive(expr: Expression): Boolean = expr match { - case _: Literal | _: Attribute | _: BoundReference => true - case _ => false - } - override def getSupportLevel(expr: ArrayContains): SupportLevel = { val elementType = expr.left.dataType.asInstanceOf[ArrayType].elementType if (hasNonDefaultStringCollation(elementType)) { @@ -226,7 +243,7 @@ object CometArrayContains } else if (isFlatFloat(elementType)) { // Spark skips the value when the array is null, so a value that can throw (an ANSI cast, // say) must not run for a null array. The native kernel evaluates both arguments first. - if (expr.left.nullable && !orderInsensitive(expr.right)) { + if (expr.left.nullable && !EvaluationOrder.isUnobservable(expr.right)) { Incompatible(Some(eagerEvalReason)) } else { Compatible() @@ -548,23 +565,10 @@ object CometArrayJoin "array_join evaluates its delimiter and null replacement eagerly, while Spark short-circuits " + "past them (https://github.com/apache/datafusion-comet/issues/3178)" - /** - * Whether evaluating `expr` earlier than Spark would is unobservable. - * - * Spark skips ArrayJoin's later arguments once an earlier one is null, and `eval` and - * `doGenCode` disagree on that order, while DataFusion evaluates every argument up front. A - * literal or column read cannot throw, carry state or have a side effect, so ordering cannot be - * observed for it; anything else goes to the codegen dispatcher. `foldable` is not usable here: - * ConstantFolding leaves a throwing foldable expression unfolded in a conditional branch. - */ - private def orderInsensitive(expr: Expression): Boolean = expr match { - case _: Literal | _: Attribute | _: BoundReference => true - case _ => false - } - override def getIncompatibleReasons(): Seq[String] = Seq(collationReason, eagerEvalReason) override def getSupportLevel(expr: ArrayJoin): SupportLevel = { + val laterArgs = expr.delimiter +: expr.nullReplacement.toSeq // Spark 4.0 widens ArrayJoin's input to StringTypeWithCollation. Concatenation itself is // collation-independent, so the joined value is always correct; only the output string's // collation metadata is dropped (Comet columns are UTF8_BINARY). Report Incompatible rather @@ -572,7 +576,7 @@ object CometArrayJoin // array_join native and matching Spark, consistent with CometReverse's #2190 handling. if (hasNonDefaultStringCollation(expr.array.dataType)) { Incompatible(Some(collationReason)) - } else if (!(expr.delimiter +: expr.nullReplacement.toSeq).forall(orderInsensitive)) { + } else if (!laterArgs.forall(EvaluationOrder.isUnobservable)) { Incompatible(Some(eagerEvalReason)) } else { Compatible() @@ -901,7 +905,7 @@ object CometArrayReverse extends CometExpressionSerde[Reverse] with ArraysBase { } -object CometElementAt extends CometExpressionSerde[ElementAt] { +object CometElementAt extends CometExpressionSerde[ElementAt] with GatedCodegenDispatchFallback { /** * Under ANSI, neither native shape reproduces Spark for a nullable nondeterministic operand. @@ -910,8 +914,9 @@ object CometElementAt extends CometExpressionSerde[ElementAt] { * whole batch first, so a throwing index fires on rows whose operand is NULL. `convert` * reproduces the short-circuit with a `CASE WHEN IS NOT NULL` guard, but that guard * serializes the operand twice, which a stateful operand cannot survive: the two copies advance - * its state independently and silently move values and NULLs. Declining leaves the lookup on - * Spark. Lifting this needs a native lookup that evaluates the operand once and masks the index + * its state independently and silently move values and NULLs. Declining routes the lookup + * through the JVM codegen dispatcher, where Spark's own `doGenCode` evaluates the operand once. + * Lifting this needs a native lookup that evaluates the operand once and masks the index * evaluation with the result, at which point the guard becomes unnecessary for every operand. */ private val eagerIndexReason: String = @@ -922,6 +927,8 @@ object CometElementAt extends CometExpressionSerde[ElementAt] { private def needsNullGuard(expr: ElementAt): Boolean = expr.failOnError && expr.left.nullable + override def getUnsupportedReasons(): Seq[String] = eagerIndexReason +: MapKeySupport.reasons + override def getSupportLevel(expr: ElementAt): SupportLevel = { if (needsNullGuard(expr) && !expr.left.deterministic) { Unsupported(Some(eagerIndexReason)) @@ -1050,14 +1057,19 @@ object CometArrayFilter extends CometExpressionSerde[ArrayFilter] { } } -object CometSize extends CometExpressionSerde[Size] { +object CometSize extends CometExpressionSerde[Size] with GatedCodegenDispatchFallback { + + override def getUnsupportedReasons(): Seq[String] = + Seq(NullGuardSupport.nondeterministicReason) override def getSupportLevel(expr: Size): SupportLevel = { - expr.child.dataType match { - case _: ArrayType => Compatible() - case _: MapType => Compatible() - case other => - Unsupported(Some(s"Unsupported child data type: $other")) + NullGuardSupport.nondeterministicChild(Seq(expr.child)).getOrElse { + expr.child.dataType match { + case _: ArrayType => Compatible() + case _: MapType => Compatible() + case other => + Unsupported(Some(s"Unsupported child data type: $other")) + } } } @@ -1128,10 +1140,13 @@ object CometArrayPosition extends CometExpressionSerde[ArrayPosition] with Array } } -object CometArraysZip extends CometExpressionSerde[ArraysZip] { +object CometArraysZip extends CometExpressionSerde[ArraysZip] with GatedCodegenDispatchFallback { override def getUnsupportedReasons(): Seq[String] = Seq( - "Not all input data types are supported; falls back to Spark for unsupported types") + "An array whose element type is a map, a day-time or year-month interval, a variant, a " + + "`TIME` value or a user-defined type has no native `arrays_zip` kernel, and neither does " + + "a struct or inner array that holds one of those.", + NullGuardSupport.nondeterministicReason) private def isTypeSupported(dt: DataType): Boolean = { import DataTypes._ @@ -1147,13 +1162,13 @@ object CometArraysZip extends CometExpressionSerde[ArraysZip] { } override def getSupportLevel(expr: ArraysZip): SupportLevel = { - val inputTypes = expr.children.map(_.dataType).toSet - for (dt <- inputTypes) { - if (!isTypeSupported(dt)) { - return Unsupported(Some(s"Unsupported child data type: $dt")) - } + NullGuardSupport.nondeterministicChild(expr.children).getOrElse { + expr.children + .map(_.dataType) + .collectFirst { case dt if !isTypeSupported(dt) => dt } + .map(dt => Unsupported(Some(s"Unsupported child data type: $dt"))) + .getOrElse(Compatible()) } - Compatible() } override def convert( diff --git a/spark/src/main/scala/org/apache/comet/serde/maps.scala b/spark/src/main/scala/org/apache/comet/serde/maps.scala index 86376ccfca8..58aea299a28 100644 --- a/spark/src/main/scala/org/apache/comet/serde/maps.scala +++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala @@ -46,6 +46,8 @@ private[serde] object MapKeySupport { "cannot reproduce Spark's equality for a complex key type (for example a `NULL` inside the " + "lookup key aborts the cast against a non-nullable nested component)." + val reasons: Seq[String] = Seq(floatingPointReason, collationReason, complexKeyReason) + /** * The `SupportLevel` for a map-consuming expression whose stored-key type is `keyType`. Spark * finds a key with `TypeUtils.getInterpretedOrdering` over the map's stored keys, while native @@ -115,7 +117,11 @@ object CometMapValues extends CometExpressionSerde[MapValues] { } } -object CometMapExtract extends CometExpressionSerde[GetMapValue] { +object CometMapExtract + extends CometExpressionSerde[GetMapValue] + with GatedCodegenDispatchFallback { + + override def getUnsupportedReasons(): Seq[String] = MapKeySupport.reasons override def getSupportLevel(expr: GetMapValue): SupportLevel = expr.child.dataType match { case MapType(keyType, _, _) => MapKeySupport.keySupport(keyType) @@ -171,21 +177,26 @@ private object MapKeyDedupPolicySupport { object CometMapFromArrays extends CometExpressionSerde[MapFromArrays] - with CodegenDispatchFallback { + with GatedCodegenDispatchFallback { override def getIncompatibleReasons(): Seq[String] = Seq(MapBuilderKeySupport.floatingPointReason, MapKeyDedupPolicySupport.incompatibleReason) + override def getUnsupportedReasons(): Seq[String] = + Seq(NullGuardSupport.nondeterministicReason) + override def getCompatibleNotes(): Seq[String] = Seq(MapKeyDedupPolicySupport.nullKeyReason) override def getSupportLevel(expr: MapFromArrays): SupportLevel = { - if (MapBuilderKeySupport.hasFloatingPointKey(expr.dataType)) { - Incompatible(Some(MapBuilderKeySupport.floatingPointReason)) - } else if (MapKeyDedupPolicySupport.isLastWin) { - Incompatible(Some(MapKeyDedupPolicySupport.incompatibleReason)) - } else { - Compatible(None) + NullGuardSupport.nondeterministicChild(expr.children).getOrElse { + if (MapBuilderKeySupport.hasFloatingPointKey(expr.dataType)) { + Incompatible(Some(MapBuilderKeySupport.floatingPointReason)) + } else if (MapKeyDedupPolicySupport.isLastWin) { + Incompatible(Some(MapKeyDedupPolicySupport.incompatibleReason)) + } else { + Compatible(None) + } } } diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala index af78e16e825..aa456ec42d9 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala @@ -46,12 +46,12 @@ object CometTakeOrderedAndProjectExec extends CometSink[TakeOrderedAndProjectExe if (!isCometShuffleEnabled(op.conf)) { return Unsupported(Some("TakeOrderedAndProject requires shuffle to be enabled")) } - op.projectList.foreach { p => - val o = exprToProto(p, op.child.output) - if (o.isEmpty) { - return Unsupported(Some(s"unsupported projection: $p")) - } - o + // The same scope as `finalNativePlan`, which serializes the projection at execution + val unsupportedProjection = withSharedSubexpressions(op.projectList) { + op.projectList.find(p => exprToProto(p, op.child.output).isEmpty) + } + if (unsupportedProjection.isDefined) { + return Unsupported(Some(s"unsupported projection: ${unsupportedProjection.get}")) } op.sortOrder.foreach { s => val o = exprToProto(s, op.child.output) diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala index 616a9716c8c..6318da35688 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala @@ -676,9 +676,12 @@ object CometShuffleExchangeExec reasons += s"${CometConf.COMET_SHUFFLE_NATIVE_HASH_PARTITIONING_ENABLED.key} is disabled" } - for (expr <- expressions) { - if (QueryPlanSerde.exprToProto(expr, inputs).isEmpty) { - reasons += s"unsupported hash partitioning expression: $expr" + // The same scope as CometNativeShuffleWriter, which serializes the keys as one projection + QueryPlanSerde.withSharedSubexpressions(expressions) { + for (expr <- expressions) { + if (QueryPlanSerde.exprToProto(expr, inputs).isEmpty) { + reasons += s"unsupported hash partitioning expression: $expr" + } } } for (dt <- expressions.map(_.dataType).distinct) { @@ -716,12 +719,15 @@ object CometShuffleExchangeExec s"${CometConf.COMET_SHUFFLE_NATIVE_RANGE_PARTITIONING_ENABLED.key} is disabled" return reasons.toSeq } - for (o <- orderings) { - val (result, expressionReasons) = - QueryPlanSerde.exprToProtoWithFallbackReasons(o, inputs) - if (result.isEmpty) { - reasons += s"unsupported range partitioning sort order: $o" - reasons ++= expressionReasons + // As for hash partitioning, the same scope as CometNativeShuffleWriter + QueryPlanSerde.withSharedSubexpressions(orderings.map(_.child)) { + for (o <- orderings) { + val (result, expressionReasons) = + QueryPlanSerde.exprToProtoWithFallbackReasons(o, inputs) + if (result.isEmpty) { + reasons += s"unsupported range partitioning sort order: $o" + reasons ++= expressionReasons + } } } for (dt <- orderings.map(_.dataType).distinct) { diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_append_calendar_interval_output.sql b/spark/src/test/resources/sql-tests/expressions/array/array_append_calendar_interval_output.sql new file mode 100644 index 00000000000..2b7850894a6 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_append_calendar_interval_output.sql @@ -0,0 +1,34 @@ +-- 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. + +-- The nondeterministic nullable array forces the dispatcher, whose interval conversion overflows +-- past about 292 years (#5279), as the 1e10-second rows do, so the result falls back to Spark. +-- Spark 4.0 rewrites `array_append` to `array_insert(-1)`, so this is Spark 3.x only. + +-- MaxSparkVersion: 3.5 + +statement +CREATE TABLE test_array_append_interval(s decimal(18, 6)) USING parquet + +statement +INSERT INTO test_array_append_interval VALUES (10000000000.000000), (-10000000000.000000), (1.000000), (NULL) + +query expect_fallback(holds a calendar interval) +SELECT array_append( + IF(monotonically_increasing_id() >= 0, array(make_interval(0, 0, 0, 0, 0, 0, s)), NULL), + make_interval(0, 0, 0, 0, 0, 0, s)) +FROM test_array_append_interval diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql index 93e9caa0b8c..6bbe2705717 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql @@ -51,18 +51,19 @@ FROM array_append_eager -- A column or literal item cannot raise, so it needs no extra evaluation query expect_native(array_append) SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), _1) AS a, - array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), 5) AS b + array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), 5) AS b, + array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), CAST(NULL AS INT)) AS c FROM array_append_eager -- A nondeterministic operand is declined when the array can be NULL: the native guard would -- serialize it more than once, and each copy would see only a subset of the rows. This covers the -- item and the array alike. -query expect_fallback(nondeterministic operand) +query expect_dispatch(array_append) SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), CAST(rand(7L) * 3 AS INT)) AS a FROM array_append_eager -query expect_fallback(nondeterministic operand) -SELECT _1, array_append(IF(rand(7L) < 2, CAST(NULL AS ARRAY), array(1)), _1) AS a +query expect_dispatch(array_append) +SELECT _1, array_append(IF(rand(7) * 0.0D = 0.0D, CAST(NULL AS ARRAY), array(1)), _1) AS a FROM array_append_eager -- A non-nullable array needs no guard, so `convert` emits none and the item is evaluated once over @@ -71,6 +72,10 @@ query expect_native(array_append) SELECT _1, array_append(array(1, 2), CAST(rand(7L) * 3 AS INT)) AS a FROM array_append_eager +query expect_native(array_append) +SELECT _1, array_append(array(1), _1) AS a +FROM array_append_eager + -- A non-nullable array still raises for a throwing item query expect_error_class(DIVIDE_BY_ZERO) SELECT _1, array_append(array(1, 2), 1 / (_1 - 1)) AS a @@ -102,6 +107,18 @@ SELECT _1, array_append(arr, 1 / (_1 - 1)) AS a FROM array_append_no_null query expect_native(array_append) SELECT _1, array_append(arr, 1 / (_1 + 1)) AS a FROM array_append_no_null +statement +CREATE TABLE array_append_mixed_null(_1 INT, arr ARRAY) USING parquet + +statement +INSERT INTO array_append_mixed_null SELECT /*+ COALESCE(1) */ * FROM VALUES (0, array(1)), (1, NULL), (2, array()), (3, array(4, 5)) + +query expect_native(array_append) +SELECT _1, array_append(arr, _1 + 1) AS a FROM array_append_mixed_null + +query expect_native(array_append) +SELECT _1, array_append(array(1), _1 + 1) AS a FROM array_append_mixed_null + -- Complex element types: the item is itself an array, and the ELSE branch's NULL has to be typed as -- the list of arrays that THEN returns. The item divides by zero only where the array is NULL. query expect_error_class(DIVIDE_BY_ZERO) @@ -130,3 +147,42 @@ FROM array_append_eager query expect_native(array_append) SELECT _1, array_append(arr, _1 + 1) AS a FROM array_append_all_null + +-- `element_at` raises on index 0 even without ANSI, so the item raises on the NULL-array row +query expect_error_class(INVALID_INDEX_OF_ZERO) +SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), + element_at(array(1), _1 - 1)) AS a +FROM array_append_eager + +query expect_error_class(INVALID_INDEX_OF_ZERO) +SELECT _1, array_append(array(1), element_at(array(1), _1 - 1)) AS a +FROM array_append_eager + +-- Control: the lookup raises on its own, so the queries above are not vacuous +query expect_error_class(INVALID_INDEX_OF_ZERO) +SELECT _1, element_at(array(1), _1 - 1) AS e +FROM array_append_eager + +-- Spark's projection evaluates the shared `element_at(array(1), _1 - 1)` on every row, so it +-- raises on the all-NULL row although each lookup skips its key on a NULL map +statement +CREATE TABLE array_append_shared_item( + _1 INT, a1 ARRAY, a2 ARRAY, m1 MAP, m2 MAP) +USING parquet + +statement +INSERT INTO array_append_shared_item VALUES + (1, CAST(NULL AS ARRAY), CAST(NULL AS ARRAY), + CAST(NULL AS MAP), CAST(NULL AS MAP)), + (2, array(2), array(3), map(1, 5), map(1, 6)) + +query expect_error_class(INVALID_INDEX_OF_ZERO) +SELECT array_append(a1, element_at(m1, element_at(array(1), _1 - 1))), + array_append(a2, element_at(m2, element_at(array(1), _1 - 1))) +FROM array_append_shared_item + +query expect_native(array_append) +SELECT array_append(a1, element_at(m1, element_at(array(1), _1 - 1))), + array_append(a2, element_at(m2, element_at(array(1), _1 - 1))) +FROM array_append_shared_item +WHERE _1 = 2 diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_append_nondeterministic_child.sql b/spark/src/test/resources/sql-tests/expressions/array/array_append_nondeterministic_child.sql new file mode 100644 index 00000000000..5fe9d484502 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_append_nondeterministic_child.sql @@ -0,0 +1,51 @@ +-- 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. + +-- Over a nullable array the native NULL guard serializes the array twice and evaluates the item in +-- both branches, so a nondeterministic operand is dispatched instead. Spark 4.0 rewrites +-- `array_append` to `array_insert(-1)`, so this is Spark 3.x only. + +-- MaxSparkVersion: 3.5 + +statement +CREATE TABLE test_array_append_nondet(_1 int) USING parquet + +statement +INSERT INTO test_array_append_nondet VALUES + (0), (1), (2), (3), (4), (5), (6), (7), (8), (9), (10), (11), (12), (13), (14), (15) + +query expect_dispatch(array_append) +SELECT _1, array_append(IF(monotonically_increasing_id() % 2 = 0, array(1), CAST(NULL AS ARRAY)), 2) AS a +FROM test_array_append_nondet + +-- A non-nullable array gets no guard, so each operand is evaluated once +query expect_native(array_append) +SELECT _1, array_append(array(monotonically_increasing_id()), 2) AS a +FROM test_array_append_nondet + +-- Spark advances the counter on all 16 rows, which the guard's two branches cannot reproduce +query expect_dispatch(array_append) +SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), monotonically_increasing_id()) AS a +FROM test_array_append_nondet + +query expect_native(array_append) +SELECT _1, array_append(array(1), monotonically_increasing_id()) AS a +FROM test_array_append_nondet + +query expect_native(array_append) +SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), 2) AS a +FROM test_array_append_nondet diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_append_shared_subexpression.sql b/spark/src/test/resources/sql-tests/expressions/array/array_append_shared_subexpression.sql new file mode 100644 index 00000000000..3a45dcb18b0 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_append_shared_subexpression.sql @@ -0,0 +1,73 @@ +-- 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. + +-- The nondeterministic nullable array forces the dispatcher, which compiles each call alone and so +-- can skip a subexpression Spark's projection shares. `OptimizeRand` would fold `rand(7) >= 0.0D`, +-- hence `rand(7) * 0.0D = 0.0D`. Spark 4.0 rewrites `array_append`, so this is Spark 3.x only. + +-- MaxSparkVersion: 3.5 + +-- Config: spark.sql.ansi.enabled=true +-- Config: spark.sql.subexpressionElimination.enabled=true + +statement +CREATE TABLE test_array_append_shared( + a1 array, a2 array, m1 map, m2 map, x double) +USING parquet + +statement +INSERT INTO test_array_append_shared VALUES + (CAST(NULL AS ARRAY), CAST(NULL AS ARRAY), + CAST(NULL AS MAP), CAST(NULL AS MAP), 0.0D), + (array(1.5D), array(2.5D), map(0.5D, 5.0D), map(0.5D, 6.0D), 2.0D) + +-- The lookup skips its key on the NULL-map row, where the shared `1.0D / x` divides by zero +query expect_error(DIVIDE_BY_ZERO) +SELECT array_append(IF(rand(7) * 0.0D = 0.0D, a1, NULL), element_at(m1, 1.0D / x)), + array_append(IF(rand(7) * 0.0D = 0.0D, a2, NULL), element_at(m2, 1.0D / x)) +FROM test_array_append_shared + +-- A single call shares nothing, so Spark skips the divide on the NULL map too. +query expect_dispatch(array_append) +SELECT array_append(IF(rand(7) * 0.0D = 0.0D, a1, NULL), element_at(m1, 1.0D / x)) +FROM test_array_append_shared + +-- `array_append` evaluates its item before its NULL check, so both paths raise here +query expect_error(DIVIDE_BY_ZERO) +SELECT array_append(IF(rand(7) * 0.0D = 0.0D, a1, NULL), 1.0D / x), + array_append(IF(rand(7) * 0.0D = 0.0D, a2, NULL), 1.0D / x) +FROM test_array_append_shared + +-- A divisor that is never zero: dispatched, unless a map lookup could skip it +query expect_dispatch(array_append) +SELECT array_append(IF(rand(7) * 0.0D = 0.0D, a1, NULL), 1.0D / (x + 1.0D)), + array_append(IF(rand(7) * 0.0D = 0.0D, a2, NULL), 1.0D / (x + 1.0D)) +FROM test_array_append_shared + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT array_append(IF(rand(7) * 0.0D = 0.0D, a1, NULL), element_at(m1, 1.0D / (x + 1.0D))), + array_append(IF(rand(7) * 0.0D = 0.0D, a2, NULL), element_at(m2, 1.0D / (x + 1.0D))) +FROM test_array_append_shared + +-- With elimination off nothing is shared, so Spark skips the divide on the NULL-map row too +statement +SET spark.sql.subexpressionElimination.enabled=false + +query expect_dispatch(array_append) +SELECT array_append(IF(rand(7) * 0.0D = 0.0D, a1, NULL), element_at(m1, 1.0D / x)), + array_append(IF(rand(7) * 0.0D = 0.0D, a2, NULL), element_at(m2, 1.0D / x)) +FROM test_array_append_shared diff --git a/spark/src/test/resources/sql-tests/expressions/array/arrays_zip.sql b/spark/src/test/resources/sql-tests/expressions/array/arrays_zip.sql index e62e3842e8f..8537796431d 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/arrays_zip.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/arrays_zip.sql @@ -202,8 +202,7 @@ SELECT arrays_zip(array(DATE '1997', DATE '1998', NULL), array(TIMESTAMP '1997-0 query SELECT arrays_zip(array(X'123456', X'123', null), array(array(X'789', X'1', null, null))) --- Arrays of Time (supported bySpark 4.1.0: https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/types/TimeType.html) --- SELECT arrays_zip(array(TIME '23:59:59.999999', TIME '2:0:3')); +-- Arrays of TIME need Spark 4.1 and spark.sql.timeType.enabled: see arrays_zip_time.sql -- Arrays of structs query @@ -218,12 +217,49 @@ SELECT arrays_zip(array(struct(1, 2, 3), struct(2, 3, 4))); -- query -- SELECT arrays_zip(array(struct(1, 2, 3), struct(2, 3, 4), struct(null, null, null))); --- Arrays of maps --- FIXME: COMET: map is not supported, unsupported arguments for CreateArray, unsupported arguments for ArraysZip --- +------------------------------------------------------------------+ --- |arrays_zip(array(map(1.0, 2, 3.0, 4)), array(map(1.0, 2, 3.0, 4)))| --- +------------------------------------------------------------------+ --- |[{{1.0 -> 2, 3.0 -> 4}, {1.0 -> 2, 3.0 -> 4}}] | --- +------------------------------------------------------------------+ --- query --- SELECT arrays_zip(array(map(1.0, '2', 3.0, '4')), array(map(1.0, '2', 3.0, '4'))); +-- Arrays of maps: no native kernel takes a map element, so these dispatch +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(map(1.0, '2', 3.0, '4')), array(map(1.0, '2', 3.0, '4'))); + +statement +CREATE TABLE test_arrays_zip_map(m array>, n array>) USING parquet + +statement +INSERT INTO test_arrays_zip_map VALUES + (array(map('a', 1, 'b', 2), map('c', NULL)), array(map(1, 'x'))), + (array(CAST(map() AS map), NULL), array(map(2, 'y'), map(3, NULL), NULL)), + (NULL, array(map(4, 'z'))), + (array(map('d', 4)), NULL), + (array(), array(map(5, 'w'))) + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(m, n) FROM test_arrays_zip_map + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(m, array(1, 2, 3)) FROM test_arrays_zip_map + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(named_struct('m', m)), array(n)) FROM test_arrays_zip_map + +-- Interval elements have no native kernel either, so these dispatch. They come from int columns, +-- as a Parquet interval column falls back at the scan. +statement +CREATE TABLE test_arrays_zip_interval(d int, h int, y int, mo int) USING parquet + +statement +INSERT INTO test_arrays_zip_interval VALUES (1, 2, 1, 2), (-3, 0, -3, -4), (0, 0, 0, 0), (NULL, 5, NULL, 6) + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(make_dt_interval(d, h), NULL), array(1, 2, 3)) FROM test_arrays_zip_interval + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(make_ym_interval(y, mo)), array(make_ym_interval(mo, y), NULL)) FROM test_arrays_zip_interval + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(make_dt_interval(d)), array(named_struct('i', make_ym_interval(y)))) FROM test_arrays_zip_interval + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(array(make_ym_interval(y), NULL)), array(d)) FROM test_arrays_zip_interval + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(INTERVAL '1 02:03:04.5' DAY TO SECOND, NULL), array(INTERVAL '1-2' YEAR TO MONTH)) diff --git a/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_nondeterministic_child.sql b/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_nondeterministic_child.sql new file mode 100644 index 00000000000..1af6b91e34e --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_nondeterministic_child.sql @@ -0,0 +1,62 @@ +-- 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. + +-- The native NULL guard serializes each child twice, so a nondeterministic child would drift +-- between the copies and is dispatched instead. + +statement +CREATE TABLE test_arrays_zip_nondet(_1 int, arr array) USING parquet + +statement +INSERT INTO test_arrays_zip_nondet +SELECT id, IF(id % 4 = 3, NULL, array(id, id + 1)) FROM range(0, 16) + +query expect_dispatch(arrays_zip) +SELECT _1, arrays_zip(IF(monotonically_increasing_id() % 2 = 0, array(1), CAST(NULL AS ARRAY)), array(2)) AS z +FROM test_arrays_zip_nondet + +query expect_dispatch(arrays_zip) +SELECT _1, arrays_zip(array(2), IF(monotonically_increasing_id() % 2 = 0, array(1), CAST(NULL AS ARRAY))) AS z +FROM test_arrays_zip_nondet + +-- Declined even when non-nullable, rather than relying on the guard matching every row +query expect_dispatch(arrays_zip) +SELECT _1, arrays_zip(array(monotonically_increasing_id()), array(2)) AS z +FROM test_arrays_zip_nondet + +-- Column arrays make the kernel copy ListVectors, and every fourth row brings a NULL array from +-- the input rather than from the IF +query expect_dispatch(arrays_zip) +SELECT _1, arrays_zip(IF(monotonically_increasing_id() % 2 = 0, arr, CAST(NULL AS ARRAY)), arr) AS z +FROM test_arrays_zip_nondet + +query expect_native(arrays_zip) +SELECT _1, arrays_zip(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), array(2)) AS z +FROM test_arrays_zip_nondet + +-- The two subtrees bind to the same ordinals, yet each keeps its own counter as in Spark (0 to 3 +-- in both). One partition writes one file, so all four rows come from one task. +statement +CREATE TABLE test_arrays_zip_nondet_pair(x int, y int) USING parquet + +statement +INSERT INTO test_arrays_zip_nondet_pair SELECT id, id + 100 FROM range(0, 4, 1, 1) + +query expect_dispatch(arrays_zip, monotonically_increasing_id) +SELECT arrays_zip(array(map(x, monotonically_increasing_id()))) AS a, + arrays_zip(array(map(y, monotonically_increasing_id()))) AS b +FROM test_arrays_zip_nondet_pair diff --git a/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_shared_subexpression.sql b/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_shared_subexpression.sql new file mode 100644 index 00000000000..44f8aa259d3 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_shared_subexpression.sql @@ -0,0 +1,72 @@ +-- 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. + +-- A nondeterministic argument forces the dispatcher, which compiles each call alone and so can +-- skip a subexpression that Spark's projection shares. `rand(7) * 0.0D` changes no value. + +-- Config: spark.sql.ansi.enabled=true +-- Config: spark.sql.subexpressionElimination.enabled=true + +-- On the last row both `1.0D / x` and `1.0D / (x + 1.0D)` find a key. +statement +CREATE TABLE test_arrays_zip_shared( + a1 array, a2 array, m1 map, m2 map, x double) +USING parquet + +statement +INSERT INTO test_arrays_zip_shared VALUES + (array(1), array(2), CAST(NULL AS MAP), CAST(NULL AS MAP), 0.0D), + (array(3), array(4), map(0.5D, 5), map(0.5D, 6), 2.0D), + (array(5), array(6), map(1.0D, 7, 0.5D, 8), map(1.0D, 9, 0.5D, 10), 1.0D) + +query expect_error(DIVIDE_BY_ZERO) +SELECT arrays_zip(a1, array(element_at(m1, 1.0D / x + rand(7) * 0.0D))), + arrays_zip(a2, array(element_at(m2, 1.0D / x + rand(7) * 0.0D))) +FROM test_arrays_zip_shared + +-- A single call shares nothing, so Spark skips the divide on the NULL map too. +query expect_dispatch(arrays_zip) +SELECT arrays_zip(a1, array(element_at(m1, 1.0D / x + rand(7) * 0.0D))) FROM test_arrays_zip_shared + +-- `x + 1.0D` is never zero, yet the shared key still declines the dispatcher +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT arrays_zip(a1, array(element_at(m1, 1.0D / (x + 1.0D) + rand(7) * 0.0D))), + arrays_zip(a2, array(element_at(m2, 1.0D / (x + 1.0D) + rand(7) * 0.0D))) +FROM test_arrays_zip_shared + +-- `arrays_zip` skips its second argument, the shared `1.0D / x`, when the first is NULL +statement +CREATE TABLE test_arrays_zip_shared_null_array(a1 array, a2 array, x double) +USING parquet + +statement +INSERT INTO test_arrays_zip_shared_null_array VALUES + (CAST(NULL AS ARRAY), CAST(NULL AS ARRAY), 0.0D), + (array(3), array(4), 2.0D) + +query expect_error(DIVIDE_BY_ZERO) +SELECT arrays_zip(a1, array(1.0D / x + rand(7) * 0.0D)), + arrays_zip(a2, array(1.0D / x + rand(7) * 0.0D)) +FROM test_arrays_zip_shared_null_array + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(a1, array(1.0D / x + rand(7) * 0.0D)) FROM test_arrays_zip_shared_null_array + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT arrays_zip(a1, array(1.0D / (x + 1.0D) + rand(7) * 0.0D)), + arrays_zip(a2, array(1.0D / (x + 1.0D) + rand(7) * 0.0D)) +FROM test_arrays_zip_shared_null_array diff --git a/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_time.sql b/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_time.sql new file mode 100644 index 00000000000..012d3e49054 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/arrays_zip_time.sql @@ -0,0 +1,32 @@ +-- 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. + +-- MinSparkVersion: 4.1 +-- Config: spark.sql.timeType.enabled=true + +-- TIME elements have no native arrays_zip kernel, so these dispatch +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(TIME '23:59:59.999999', TIME '2:0:3')) + +statement +CREATE TABLE test_arrays_zip_time(hours int, minutes int, secs decimal(16, 6)) USING parquet + +statement +INSERT INTO test_arrays_zip_time VALUES (0, 0, 0.000000), (23, 59, 59.999999), (12, NULL, 30.000000) + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(make_time(hours, minutes, secs), NULL), array(hours)) FROM test_arrays_zip_time diff --git a/spark/src/test/resources/sql-tests/expressions/array/element_at_ansi.sql b/spark/src/test/resources/sql-tests/expressions/array/element_at_ansi.sql index 89293ccefd9..408661236c8 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/element_at_ansi.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/element_at_ansi.sql @@ -119,8 +119,14 @@ FROM ansi_element_at_null -- both operands are NULL on every row and Spark returns NULL without evaluating either index. -- The non-ANSI spelling stays native and is covered in element_at.sql. -- https://github.com/apache/datafusion-comet/issues/5544 -query expect_fallback(nullable nondeterministic array or map operand) +query expect_dispatch(element_at) SELECT id, element_at(IF(monotonically_increasing_id() % 2 = 0, CAST(NULL AS ARRAY), array(1)), 1) AS v1, element_at(IF(rand(7L) < 2, CAST(NULL AS ARRAY), array(1)), 1 + (id % (id - 2))) AS v2 FROM ansi_element_at_null + +-- A calendar interval result falls back instead (#5279) +query expect_fallback(holds a calendar interval) +SELECT element_at(IF(monotonically_increasing_id() >= 0, + array(make_interval(0, 0, 0, 0, 0, 0, 10000000000.000000)), NULL), 1) +FROM ansi_element_at_null diff --git a/spark/src/test/resources/sql-tests/expressions/array/size_nondeterministic_child.sql b/spark/src/test/resources/sql-tests/expressions/array/size_nondeterministic_child.sql new file mode 100644 index 00000000000..e6cdb294adb --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/size_nondeterministic_child.sql @@ -0,0 +1,46 @@ +-- 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. + +-- The native NULL guard serializes the child twice, so a nondeterministic child would drift +-- between the copies and is dispatched instead. + +-- ConfigMatrix: spark.sql.legacy.sizeOfNull=true,false + +statement +CREATE TABLE test_size_nondet(_1 int, arr array) USING parquet + +statement +INSERT INTO test_size_nondet +SELECT id, IF(id % 4 = 3, NULL, array(id, id + 1)) FROM range(0, 16) + +query expect_dispatch(size) +SELECT _1, size(IF(monotonically_increasing_id() % 2 = 0, array(1), CAST(NULL AS ARRAY))) AS s +FROM test_size_nondet + +-- Declined even when non-nullable, rather than relying on the guard matching every row +query expect_dispatch(size) +SELECT _1, size(array(monotonically_increasing_id())) AS s +FROM test_size_nondet + +query expect_native(size) +SELECT _1, size(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY))) AS s +FROM test_size_nondet + +-- A column array makes the dispatched kernel copy a ListVector +query expect_dispatch(size) +SELECT _1, size(IF(monotonically_increasing_id() % 2 = 0, arr, CAST(NULL AS ARRAY))) AS s +FROM test_size_nondet diff --git a/spark/src/test/resources/sql-tests/expressions/array/size_shared_subexpression.sql b/spark/src/test/resources/sql-tests/expressions/array/size_shared_subexpression.sql new file mode 100644 index 00000000000..6ef2fad6b09 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/size_shared_subexpression.sql @@ -0,0 +1,52 @@ +-- 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. + +-- A `size` over a lookup whose key holds `rand(7)` is dispatched, and the decline check finds the +-- lookup skipping the shared `1.0D / x` on the NULL-map row. `rand(7) * 0.0D` changes no value. + +-- Config: spark.sql.ansi.enabled=true +-- Config: spark.sql.subexpressionElimination.enabled=true + +-- On the last row both `1.0D / x` and `1.0D / (x + 1.0D)` find a key. +statement +CREATE TABLE test_size_shared(m1 map>, m2 map>, x double) +USING parquet + +statement +INSERT INTO test_size_shared VALUES + (CAST(NULL AS MAP>), CAST(NULL AS MAP>), 0.0D), + (map(0.5D, array(1, 2)), map(0.5D, array(3)), 2.0D), + (map(1.0D, array(1), 0.5D, array(1, 2, 3)), map(1.0D, array(4, 5), 0.5D, array(6)), 1.0D) + +-- Without `rand(7)` `size` is native, so this exercises the dispatched `element_at` +query expect_error(DIVIDE_BY_ZERO) +SELECT size(element_at(m1, 1.0D / x)), size(element_at(m2, 1.0D / x)) FROM test_size_shared + +query expect_error(DIVIDE_BY_ZERO) +SELECT size(element_at(m1, 1.0D / x + rand(7) * 0.0D)), + size(element_at(m2, 1.0D / x + rand(7) * 0.0D)) +FROM test_size_shared + +-- A single call shares nothing, so Spark skips the divide on the NULL map too. +query expect_dispatch(size) +SELECT size(element_at(m1, 1.0D / x + rand(7) * 0.0D)) FROM test_size_shared + +-- `x + 1.0D` is never zero, yet the shared key still declines the dispatcher +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT size(element_at(m1, 1.0D / (x + 1.0D) + rand(7) * 0.0D)), + size(element_at(m2, 1.0D / (x + 1.0D) + rand(7) * 0.0D)) +FROM test_size_shared diff --git a/spark/src/test/resources/sql-tests/expressions/map/element_at_map.sql b/spark/src/test/resources/sql-tests/expressions/map/element_at_map.sql index b98854320f8..c7ace7d869e 100644 --- a/spark/src/test/resources/sql-tests/expressions/map/element_at_map.sql +++ b/spark/src/test/resources/sql-tests/expressions/map/element_at_map.sql @@ -52,31 +52,50 @@ SELECT element_at(mi, CAST(1 AS BIGINT)), element_at(mi, CAST(2 AS SMALLINT)) FR query SELECT element_at(map('a', 1, 'b', 2), 'a'), element_at(map('a', 1, 'b', 2), 'missing'), element_at(map('a', 1, 'b', 2), NULL) --- Map key types whose Spark equality Comet's native `map_extract` cannot reproduce fall back to --- Spark. These stay on the constructor path here because the SQL harness excludes --- `ConstantFolding`; `CometMapExpressionSuite` covers the folded-literal form of each. +-- Map key types whose Spark equality Comet's native `map_extract` cannot reproduce are dispatched. +-- The folded-literal forms, which need `ConstantFolding`, are in `map_lookup_folded_literal.sql`. -- Spark compares with `nanSafeCompareDoubles`, so a `-0.0` lookup finds the `+0.0` key. Native --- lookup compares the raw Arrow values. -query expect_fallback(Spark looks a key up in its SQL ordering) -SELECT element_at(map(CAST(0 AS DOUBLE), 7), CAST(-0.0 AS DOUBLE)) +-- lookup compares the raw Arrow values. `-0.0D` because a decimal `-0.0` has no signed zero. +query expect_dispatch(element_at) +SELECT element_at(map(CAST(0 AS DOUBLE), 7), -0.0D) -query expect_fallback(Spark looks a key up in its SQL ordering) -SELECT element_at(map(CAST(0 AS FLOAT), 7), CAST(-0.0 AS FLOAT)) +query expect_dispatch(element_at) +SELECT element_at(map(CAST(0 AS FLOAT), 7), CAST(-0.0D AS FLOAT)) + +-- Spark's `nanSafeCompareDoubles` treats NaN as equal to itself, so a NaN lookup finds a NaN key. +query expect_dispatch(element_at) +SELECT element_at(map(CAST('NaN' AS DOUBLE), 7), CAST('NaN' AS DOUBLE)) -- The floating-point decline walks every nesting level of the key type, so an array-of-double key --- falls back for the same reason. -query expect_fallback(Spark looks a key up in its SQL ordering) -SELECT element_at(map(array(CAST(0 AS DOUBLE)), 7), array(CAST(-0.0 AS DOUBLE))) +-- is dispatched for the same reason. +query expect_dispatch(element_at) +SELECT element_at(map(array(CAST(0 AS DOUBLE)), 7), array(-0.0D)) -- A complex key type: `map_extract` casts the lookup key to the map's exact Arrow key type, so a -- NULL inside the lookup key would abort the cast instead of missing the lookup. -query expect_fallback(casts the lookup key to the map's exact Arrow key type) +query expect_dispatch(element_at) SELECT element_at(map(array(1), 7), array(CAST(NULL AS INT))) -query expect_fallback(casts the lookup key to the map's exact Arrow key type) +query expect_dispatch(element_at) SELECT element_at(map(named_struct('a', 1), 7), named_struct('a', 1)) +statement +CREATE TABLE test_element_at_struct_key(m map, int>, k int) USING parquet + +statement +INSERT INTO test_element_at_struct_key VALUES + (map(named_struct('a', 1, 'b', 'x'), 7), 1), + (map(named_struct('a', 2, 'b', 'y'), 8), 3), + (map(named_struct('a', 3, 'b', 'z'), 9), NULL), + (NULL, 1) + +query expect_dispatch(element_at) +SELECT element_at(m, named_struct('a', k, 'b', 'x')), + element_at(m, named_struct('a', k, 'b', CAST(NULL AS STRING))), + element_at(m, CAST(NULL AS STRUCT)) +FROM test_element_at_struct_key + -- `BinaryType` keys need no decline: Arrow compares them by content, as Spark's ordering does. query SELECT element_at(map(CAST('a' AS BINARY), 1, CAST('b' AS BINARY), 2), CAST('b' AS BINARY)) diff --git a/spark/src/test/resources/sql-tests/expressions/map/element_at_map_collation.sql b/spark/src/test/resources/sql-tests/expressions/map/element_at_map_collation.sql index 763e1b31f12..0dbb5c04f99 100644 --- a/spark/src/test/resources/sql-tests/expressions/map/element_at_map_collation.sql +++ b/spark/src/test/resources/sql-tests/expressions/map/element_at_map_collation.sql @@ -20,7 +20,7 @@ -- Spark 4.0+ supports string collations. Spark compares a map key under its declared collation, -- so a `UTF8_LCASE` map matches `A1` against a dynamic `a1` lookup and returns `7`. Comet's native -- `map_extract` compares string keys as `UTF8_BINARY`, so `CometElementAt` declines the lookup and --- Spark evaluates the projection. The `element_at` form is used rather than `m[key]` because +-- runs it through the dispatcher. The `element_at` form is used rather than `m[key]` because -- Spark's `SimplifyExtractValueOps` rewrites `map(...)[key]` over a literal map into a `CASE` -- before it can reach the native map lookup. @@ -30,6 +30,6 @@ CREATE TABLE test_element_at_collation(k string) USING parquet statement INSERT INTO test_element_at_collation VALUES ('a1'), ('A1'), ('zz'), (NULL) -query expect_fallback(cannot honour a non-default collation) +query expect_dispatch(element_at) SELECT element_at(map(CAST('A1' AS STRING COLLATE UTF8_LCASE), 7), CAST(k AS STRING COLLATE UTF8_LCASE)) FROM test_element_at_collation diff --git a/spark/src/test/resources/sql-tests/expressions/map/get_map_value.sql b/spark/src/test/resources/sql-tests/expressions/map/get_map_value.sql index b693cf5ae15..65155dd4a30 100644 --- a/spark/src/test/resources/sql-tests/expressions/map/get_map_value.sql +++ b/spark/src/test/resources/sql-tests/expressions/map/get_map_value.sql @@ -31,18 +31,42 @@ SELECT m['x'], m['missing'] FROM test_map query spark_answer_only SELECT map('a', 1, 'b', 2)['a'], map('a', 1, 'b', 2)['missing'], map('a', 1, 'b', 2)[NULL] --- Map key types whose Spark equality Comet's native `map_extract` cannot reproduce fall back to --- Spark. `x[key]` routes through `GetMapValue` / `CometMapExtract` (the `element_at` form in +-- Map key types whose Spark equality native `map_extract` cannot reproduce are dispatched. +-- `x[key]` routes through `GetMapValue` / `CometMapExtract` (the `element_at` form in -- `element_at_map.sql` exercises the same guard on `CometElementAt`). A `map` column -- is used rather than a `map(...)` literal because Spark's `SimplifyExtractValueOps` rewrites -- `map(...)[key]` over a literal map into a `CASE` before it can reach the native lookup. Spark -- compares with `nanSafeCompareDoubles`, so a `-0.0` lookup finds the `+0.0` key and returns `7`; -- native lookup compares the raw Arrow values. +-- `-0.0D` because a decimal `-0.0` has no signed zero. statement CREATE TABLE test_map_double(m map) USING parquet statement INSERT INTO test_map_double VALUES (map(CAST(0 AS DOUBLE), 7)), (map(CAST(1 AS DOUBLE), 8)), (NULL) -query expect_fallback(Spark looks a key up in its SQL ordering) -SELECT m[CAST(-0.0 AS DOUBLE)] FROM test_map_double +query expect_dispatch(getmapvalue) +SELECT m[-0.0D] FROM test_map_double + +-- A NaN key is found by a NaN lookup, as `nanSafeCompareDoubles` treats NaN as equal to itself. +statement +INSERT INTO test_map_double VALUES (map(CAST('NaN' AS DOUBLE), 9)) + +query expect_dispatch(getmapvalue) +SELECT m[CAST('NaN' AS DOUBLE)] FROM test_map_double + +statement +CREATE TABLE test_map_struct_key(m map, int>, k int) USING parquet + +statement +INSERT INTO test_map_struct_key VALUES + (map(named_struct('a', 1, 'b', 'x'), 7), 1), + (map(named_struct('a', 2, 'b', 'y'), 8), 3), + (map(named_struct('a', 3, 'b', 'z'), 9), NULL), + (NULL, 1) + +query expect_dispatch(getmapvalue) +SELECT m[named_struct('a', k, 'b', 'x')], + m[named_struct('a', k, 'b', CAST(NULL AS STRING))], + m[CAST(NULL AS STRUCT)] +FROM test_map_struct_key diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_gate.sql b/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_gate.sql index 6708e62e3c7..9eadf08adef 100644 --- a/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_gate.sql +++ b/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_gate.sql @@ -37,10 +37,11 @@ SELECT id, map_keys(map_from_entries(array(named_struct('k', -f, 'v', 1), named_struct('k', 2.5F, 'v', 2)))) FROM float_map_keys +-- Distinct values arrays: with folding off a repeated `array(1, 2)` is shared and blocks dispatch query expect_dispatch(map_from_arrays) SELECT id, map_keys(map_from_arrays(array(-d, 2.5D), array(1, 2))), - map_keys(map_from_arrays(array(-f, 2.5F), array(1, 2))) + map_keys(map_from_arrays(array(-f, 2.5F), array(3, 4))) FROM float_map_keys -- NaNs that differ only in the sign bit are one key on every version. diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_gate_legacy.sql b/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_gate_legacy.sql index c1cbfab3358..c11320cc74a 100644 --- a/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_gate_legacy.sql +++ b/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_gate_legacy.sql @@ -34,8 +34,9 @@ SELECT map_keys(map_from_entries(array(named_struct('k', -f, 'v', 1), named_struct('k', 2.5F, 'v', 2)))) FROM float_map_keys_legacy +-- Distinct values arrays: with folding off a repeated `array(1, 2)` is shared and blocks dispatch query expect_dispatch(map_from_arrays) SELECT map_keys(map_from_arrays(array(d, -d), array(1, 2))), - map_keys(map_from_arrays(array(f, -f), array(1, 2))) + map_keys(map_from_arrays(array(f, -f), array(3, 4))) FROM float_map_keys_legacy diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_last_win.sql b/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_last_win.sql index b3c210dab1c..bd156e23890 100644 --- a/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_last_win.sql +++ b/spark/src/test/resources/sql-tests/expressions/map/map_builders_float_key_last_win.sql @@ -52,11 +52,12 @@ SELECT id, FROM float_map_keys_last_win -- NaNs with different sign bits are one DOUBLE or FLOAT key: one entry, holding the last value. +-- `array(3, 4)`: with folding off a repeated `array(1, 2)` is shared and blocks dispatch query expect_dispatch(map_from_arrays) SELECT map_keys(map_from_arrays(array(d, -d), array(1, 2))), map_values(map_from_arrays(array(d, -d), array(1, 2))), - map_keys(map_from_arrays(array(f, -f), array(1, 2))), - map_values(map_from_arrays(array(f, -f), array(1, 2))) + map_keys(map_from_arrays(array(f, -f), array(3, 4))), + map_values(map_from_arrays(array(f, -f), array(3, 4))) FROM float_map_keys_last_win WHERE id = 2 query expect_dispatch(map_from_entries) @@ -70,8 +71,8 @@ FROM float_map_keys_last_win WHERE id = 2 query expect_dispatch(map_from_arrays) SELECT map_keys(map_from_arrays(array(d, -d), array(1, 2))), map_values(map_from_arrays(array(d, -d), array(1, 2))), - map_keys(map_from_arrays(array(f, -f), array(1, 2))), - map_values(map_from_arrays(array(f, -f), array(1, 2))) + map_keys(map_from_arrays(array(f, -f), array(3, 4))), + map_values(map_from_arrays(array(f, -f), array(3, 4))) FROM float_map_keys_last_win WHERE id = 1 query expect_dispatch(map_from_entries) diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_nondeterministic_child.sql b/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_nondeterministic_child.sql new file mode 100644 index 00000000000..249ffb5d6e4 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_nondeterministic_child.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. + +-- The native NULL guard serializes both children twice, so a nondeterministic child would drift +-- between the copies and is dispatched instead. + +statement +CREATE TABLE test_map_from_arrays_nondet(_1 int, k array, v array) USING parquet + +statement +INSERT INTO test_map_from_arrays_nondet +SELECT id, IF(id % 4 = 3, NULL, array(id, id + 100)), array(id * 2, id * 2 + 1) FROM range(0, 16) + +query expect_dispatch(map_from_arrays) +SELECT _1, map_from_arrays(IF(monotonically_increasing_id() % 2 = 0, array(1), CAST(NULL AS ARRAY)), array(2)) AS m +FROM test_map_from_arrays_nondet + +query expect_dispatch(map_from_arrays) +SELECT _1, map_from_arrays(array(1), IF(monotonically_increasing_id() % 2 = 0, array(2), CAST(NULL AS ARRAY))) AS m +FROM test_map_from_arrays_nondet + +-- Declined even when non-nullable, rather than relying on the guard matching every row +query expect_dispatch(map_from_arrays) +SELECT _1, map_from_arrays(array(monotonically_increasing_id()), array(2)) AS m +FROM test_map_from_arrays_nondet + +-- Column arrays make the dispatched kernel copy ListVectors +query expect_dispatch(map_from_arrays) +SELECT _1, map_from_arrays(IF(monotonically_increasing_id() % 2 = 0, k, CAST(NULL AS ARRAY)), v) AS m +FROM test_map_from_arrays_nondet + +query expect_native(map_from_arrays) +SELECT _1, map_from_arrays(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), array(2)) AS m +FROM test_map_from_arrays_nondet diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_shared_subexpression.sql b/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_shared_subexpression.sql new file mode 100644 index 00000000000..2644dca7b35 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_shared_subexpression.sql @@ -0,0 +1,55 @@ +-- 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. + +-- Under `LAST_WIN` the call is dispatched, where `nullSafeCodeGen` skips the values on a NULL-keys +-- row, while Spark's projection evaluates a shared subexpression of them on every row, so such a +-- call falls back. + +-- Config: spark.sql.ansi.enabled=true +-- Config: spark.sql.subexpressionElimination.enabled=true +-- Config: spark.sql.mapKeyDedupPolicy=LAST_WIN + +statement +CREATE TABLE test_map_from_arrays_shared(k1 array, k2 array, x double) USING parquet + +statement +INSERT INTO test_map_from_arrays_shared VALUES + (CAST(NULL AS ARRAY), CAST(NULL AS ARRAY), 0.0D), + (array(1, 1), array(2), 2.0D) + +-- The `1.0D / x` that both values arrays hold divides by zero on the row whose keys are NULL. +query expect_error(DIVIDE_BY_ZERO) +SELECT map_from_arrays(k1, array(1.0D / x, 1.0D)), map_from_arrays(k2, array(1.0D / x)) +FROM test_map_from_arrays_shared + +-- A single call shares nothing, so Spark skips the values on the NULL keys too. +query expect_dispatch(map_from_arrays) +SELECT map_from_arrays(k1, array(1.0D / x, 1.0D)) FROM test_map_from_arrays_shared + +-- `x + 1.0D` is never zero, yet the shared values still decline the dispatcher +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT map_from_arrays(k1, array(1.0D / (x + 1.0D), 1.0D)), + map_from_arrays(k2, array(1.0D / (x + 1.0D))) +FROM test_map_from_arrays_shared + +-- With elimination off nothing is shared, so Spark skips the values on the NULL-keys row too +statement +SET spark.sql.subexpressionElimination.enabled=false + +query expect_dispatch(map_from_arrays) +SELECT map_from_arrays(k1, array(1.0D / x, 1.0D)), map_from_arrays(k2, array(1.0D / x)) +FROM test_map_from_arrays_shared diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_lookup_folded_literal.sql b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_folded_literal.sql new file mode 100644 index 00000000000..b90bf15800b --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_folded_literal.sql @@ -0,0 +1,45 @@ +-- 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. + +-- Folding turns the literal maps into literals that the dispatched lookup receives. The +-- constructor forms are in `element_at_map.sql` and `get_map_value.sql`. + +-- ConstantFolding: enabled + +statement +CREATE TABLE test_map_lookup_folded(_1 int, _2 bigint) USING parquet + +statement +INSERT INTO test_map_lookup_folded VALUES (1, 1), (2, 2), (3, 3) + +-- A per-row key keeps the lookup itself from folding; `-0.0` must find the `+0.0` key +query expect_dispatch(element_at) +SELECT _1 AS id, element_at(map(CAST(0 AS DOUBLE), 7), + CAST(concat('-', CAST(_1 - 1 AS STRING), '.0') AS DOUBLE)) AS v +FROM test_map_lookup_folded + +-- The DOUBLE-keyed map is nested in a map value, so its keys never reach `CometLiteral`; the +-- decline must come from the lookup whose map argument is `MAP` +query expect_dispatch(element_at) +SELECT _1 AS id, element_at(element_at(map(1, map(CAST(0 AS DOUBLE), 7)), _1), + CAST(concat('-', CAST(_1 - 1 AS STRING), '.0') AS DOUBLE)) AS v +FROM test_map_lookup_folded + +-- Spark permits a lookup array holding a NULL element +query expect_dispatch(element_at) +SELECT _1 AS id, element_at(map(array(1), 7), array(IF(_1 = 2, CAST(NULL AS INT), _1))) AS v +FROM test_map_lookup_folded diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_lookup_folded_literal_collation.sql b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_folded_literal_collation.sql new file mode 100644 index 00000000000..6bb1c566d93 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_folded_literal_collation.sql @@ -0,0 +1,40 @@ +-- 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. + +-- MinSparkVersion: 4.0 + +-- The collated counterpart of `map_lookup_folded_literal.sql` + +-- ConstantFolding: enabled + +statement +CREATE TABLE test_map_lookup_folded_collation(_1 string, _2 int) USING parquet + +statement +INSERT INTO test_map_lookup_folded_collation VALUES ('a1', 1), ('zz', 1), ('A1', 2) + +query expect_dispatch(element_at) +SELECT element_at(map(CAST('A1' AS STRING COLLATE UTF8_LCASE), 7), + CAST(_1 AS STRING COLLATE UTF8_LCASE)) AS v +FROM test_map_lookup_folded_collation + +-- The column key `_2` keeps the INT-keyed lookup on `map(1, ...)` from folding, so the decline +-- must come from the lookup whose map argument is `MAP` +query expect_dispatch(element_at) +SELECT element_at(element_at(map(1, map(CAST('A1' AS STRING COLLATE UTF8_LCASE), 7)), _2), + CAST(_1 AS STRING COLLATE UTF8_LCASE)) AS v +FROM test_map_lookup_folded_collation diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression.sql b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression.sql new file mode 100644 index 00000000000..6f9f79b4d43 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression.sql @@ -0,0 +1,285 @@ +-- 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's projection evaluates shared subexpressions on every row, but the dispatcher compiles +-- each DOUBLE-keyed lookup alone and skips its key on a NULL map, so such a lookup falls back. + +-- Config: spark.sql.ansi.enabled=true +-- Config: spark.sql.subexpressionElimination.enabled=true + +-- On the last row both `1.0D / x` and `1.0D / (x + 1.0D)` find a key. +statement +CREATE TABLE test_map_lookup_shared( + m1 map, m2 map, x double, y double, k int) +USING parquet + +statement +INSERT INTO test_map_lookup_shared VALUES + (CAST(NULL AS MAP), CAST(NULL AS MAP), 0.0D, CAST(NULL AS DOUBLE), + 1), + (map(0.5D, 5), map(0.5D, 6), 2.0D, 4.0D, 2), + (map(1.0D, 7, 0.5D, 8), map(1.0D, 9, 0.5D, 10), 1.0D, 3.0D, 3) + +-- The shared `1.0D / x` divides by zero on the row whose maps are NULL. +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(m1, 1.0D / x), element_at(m2, 1.0D / x) FROM test_map_lookup_shared + +query expect_error(DIVIDE_BY_ZERO) +SELECT m1[1.0D / x], m2[1.0D / x] FROM test_map_lookup_shared + +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(m1, 1.0D / x + 1.0D), element_at(m2, 1.0D / x + 2.0D) +FROM test_map_lookup_shared + +-- A divide skips its dividend when the divisor `y` is NULL, though it sits in the map argument +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if((1.0D / x) / y > 0.0D, m1, m2), 0.5D), + element_at(if((1.0D / x) / (y + 1.0D) > 0.0D, m2, m1), 0.5D) +FROM test_map_lookup_shared + +-- An aggregate's update evaluates a subexpression its functions share for every row too. +query expect_error(DIVIDE_BY_ZERO) +SELECT max(element_at(m1, 1.0D / x)), max(element_at(m2, 1.0D / x)) FROM test_map_lookup_shared + +-- `x + 1.0D` is never zero, yet the shared key still declines the dispatcher +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT element_at(m1, 1.0D / (x + 1.0D)), element_at(m2, 1.0D / (x + 1.0D)) +FROM test_map_lookup_shared + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT element_at(if((1.0D / (x + 1.0D)) / y > 0.0D, m1, m2), 0.5D), + element_at(if((1.0D / (x + 1.0D)) / (y + 1.0D) > 0.0D, m2, m1), 0.5D) +FROM test_map_lookup_shared + +query expect_dispatch(element_at) +SELECT element_at(if((1.0D / (x + 1.0D)) / y > 0.0D, m1, m2), 0.5D) FROM test_map_lookup_shared + +-- `pmod` skips its dividend the same way. +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(pmod(1.0D / x, y) > 0.0D, m1, m2), 0.5D), + element_at(if(pmod(1.0D / x, y + 1.0D) > 0.0D, m2, m1), 0.5D) +FROM test_map_lookup_shared + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT element_at(if(pmod(1.0D / (x + 1.0D), y) > 0.0D, m1, m2), 0.5D), + element_at(if(pmod(1.0D / (x + 1.0D), y + 1.0D) > 0.0D, m2, m1), 0.5D) +FROM test_map_lookup_shared + +-- A shared divisor is evaluated first, so it is never skipped: both raise here, and the +-- never-zero forms stay dispatched +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(y / (1.0D / x) > 0.0D, m1, m2), 0.5D), + element_at(if((y + 1.0D) / (1.0D / x) > 0.0D, m2, m1), 0.5D) +FROM test_map_lookup_shared + +query expect_dispatch(element_at) +SELECT element_at(if(y / (1.0D / (x + 1.0D)) > 0.0D, m1, m2), 0.5D), + element_at(if((y + 1.0D) / (1.0D / (x + 1.0D)) > 0.0D, m2, m1), 0.5D) +FROM test_map_lookup_shared + +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(pmod(y, 1.0D / x) > 0.0D, m1, m2), 0.5D), + element_at(if(pmod(y + 1.0D, 1.0D / x) > 0.0D, m2, m1), 0.5D) +FROM test_map_lookup_shared + +query expect_dispatch(element_at) +SELECT element_at(if(pmod(y, 1.0D / (x + 1.0D)) > 0.0D, m1, m2), 0.5D), + element_at(if(pmod(y + 1.0D, 1.0D / (x + 1.0D)) > 0.0D, m2, m1), 0.5D) +FROM test_map_lookup_shared + +-- The dispatcher eliminates a subexpression the lookup itself repeats, as Spark does +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(m1, 1.0D / x + 1.0D / x) FROM test_map_lookup_shared + +query expect_dispatch(element_at) +SELECT element_at(m1, 1.0D / (x + 1.0D) + 1.0D / (x + 1.0D)) FROM test_map_lookup_shared + +-- An `if` shares only what it always evaluates, so its else-branch `1.0D / x` is not shared +query expect_dispatch(element_at) +SELECT element_at(m1, 1.0D / x), if(x = 0.0D, 0.0D, 1.0D / x) FROM test_map_lookup_shared + +-- A single lookup shares nothing, so Spark skips the divide on the NULL map too. +query expect_dispatch(element_at) +SELECT element_at(m1, 1.0D / x) FROM test_map_lookup_shared + +query expect_dispatch(getmapvalue) +SELECT m1[1.0D / x] FROM test_map_lookup_shared + +-- Subexpression elimination never shares a column. +query expect_dispatch(element_at) +SELECT element_at(m1, x), element_at(m2, x) FROM test_map_lookup_shared + +-- The shared `coalesce(m1, m2)` is the first argument, which a lookup never skips +query expect_dispatch(element_at) +SELECT element_at(coalesce(m1, m2), x * 0.25D), element_at(coalesce(m1, m2), x * 0.5D) +FROM test_map_lookup_shared + +-- The planner checks the final aggregate outside any operator; a decline there would send the +-- native partial `avg` back to Spark +query expect_dispatch(element_at) +SELECT k, element_at(map(CAST(k AS DOUBLE), 7), avg(x) + 1.0D) +FROM test_map_lookup_shared GROUP BY k + +-- `locate` evaluates its start first, so a NULL `n` skips the shared divide in the substring +statement +CREATE TABLE test_map_lookup_shared_locate( + m1 map, m2 map, x double, s string, n int) +USING parquet + +statement +INSERT INTO test_map_lookup_shared_locate VALUES + (map(0.5D, 5), map(0.5D, 6), 0.0D, 'abc', CAST(NULL AS INT)), + (map(0.5D, 5), map(0.5D, 6), 2.0D, 'x0.5', 1) + +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(locate(CAST(1.0D / x AS STRING), s, n) > 0, m1, m2), 0.5D), + element_at(if(locate(CAST(1.0D / x + 1.0D AS STRING), s, n) > 0, m2, m1), 0.5D) +FROM test_map_lookup_shared_locate + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT element_at(if(locate(CAST(1.0D / (x + 1.0D) AS STRING), s, n) > 0, m1, m2), 0.5D), + element_at(if(locate(CAST(1.0D / (x + 1.0D) + 1.0D AS STRING), s, n) > 0, m2, m1), 0.5D) +FROM test_map_lookup_shared_locate + +-- `array_join` evaluates a nullable replacement first, so a NULL `r` skips the shared array. +-- Nullable delimiters dodge SPARK-57200: before 4.2 a join whose array and delimiter cannot be +-- NULL but whose replacement can returns NULL on every row. +statement +CREATE TABLE test_map_lookup_shared_array_join( + m1 map, m2 map, x double, d1 string, d2 string, r string) +USING parquet + +statement +INSERT INTO test_map_lookup_shared_array_join VALUES + (map(0.5D, 5), map(0.5D, 6), 0.0D, ',', ';', CAST(NULL AS STRING)), + (map(0.5D, 5), map(0.5D, 6), 2.0D, ',', ';', 'z') + +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d1, r) IS NULL, m1, m2), 0.5D), + element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d2, r) IS NULL, m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +-- A NULL literal replacement can be NULL too, so it still declines. +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d1, CAST(NULL AS STRING)) IS NULL, + m1, m2), 0.5D), + element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d2, CAST(NULL AS STRING)) IS NULL, + m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(array_join(array('a', 'b'), CAST(1.0D / x AS STRING), r) IS NULL, m1, m2), + 0.5D), + element_at(if(array_join(array('a', 'c'), CAST(1.0D / x AS STRING), r) IS NULL, m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +-- Without a replacement the array is evaluated first, so Spark and the dispatcher both raise +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d1) IS NULL, m1, m2), 0.5D), + element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d2) IS NULL, m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +-- So does a divide in the replacement, or in the array when the replacement cannot be NULL +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(array_join(array('a'), d1, CAST(1.0D / x AS STRING)) IS NULL, m1, m2), 0.5D), + element_at(if(array_join(array('b'), d2, CAST(1.0D / x AS STRING)) IS NULL, m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d1, 'z') IS NULL, m1, m2), 0.5D), + element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d2, 'z') IS NULL, m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +-- A single lookup shares nothing, so Spark skips the divide on the NULL replacement too. +query expect_dispatch(element_at) +SELECT element_at(if(array_join(array(CAST(1.0D / x AS STRING)), d1, r) IS NULL, m1, m2), 0.5D) +FROM test_map_lookup_shared_array_join + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT element_at(if(array_join(array(CAST(1.0D / (x + 1.0D) AS STRING)), d1, r) IS NULL, + m1, m2), 0.5D), + element_at(if(array_join(array(CAST(1.0D / (x + 1.0D) AS STRING)), d2, r) IS NULL, + m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT element_at(if(array_join(array('a', 'b'), CAST(1.0D / (x + 1.0D) AS STRING), r) IS NULL, + m1, m2), 0.5D), + element_at(if(array_join(array('a', 'c'), CAST(1.0D / (x + 1.0D) AS STRING), r) IS NULL, + m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +query expect_dispatch(element_at) +SELECT element_at(if(array_join(array(CAST(1.0D / (x + 1.0D) AS STRING)), d1) IS NULL, + m1, m2), 0.5D), + element_at(if(array_join(array(CAST(1.0D / (x + 1.0D) AS STRING)), d2) IS NULL, + m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +query expect_dispatch(element_at) +SELECT element_at(if(array_join(array('a'), d1, CAST(1.0D / (x + 1.0D) AS STRING)) IS NULL, + m1, m2), 0.5D), + element_at(if(array_join(array('b'), d2, CAST(1.0D / (x + 1.0D) AS STRING)) IS NULL, + m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +query expect_dispatch(element_at) +SELECT element_at(if(array_join(array(CAST(1.0D / (x + 1.0D) AS STRING)), d1, 'z') IS NULL, + m1, m2), 0.5D), + element_at(if(array_join(array(CAST(1.0D / (x + 1.0D) AS STRING)), d2, 'z') IS NULL, + m2, m1), 0.5D) +FROM test_map_lookup_shared_array_join + +-- With elimination off only a whole-stage codegen aggregate shares subexpressions +statement +SET spark.sql.subexpressionElimination.enabled=false + +query expect_error(DIVIDE_BY_ZERO) +SELECT max(element_at(m1, 1.0D / x)), max(element_at(m2, 1.0D / x)) FROM test_map_lookup_shared + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT max(element_at(m1, 1.0D / (x + 1.0D))), max(element_at(m2, 1.0D / (x + 1.0D))) +FROM test_map_lookup_shared + +query expect_dispatch(element_at) +SELECT element_at(m1, 1.0D / x), element_at(m2, 1.0D / x) FROM test_map_lookup_shared + +query expect_dispatch(getmapvalue) +SELECT m1[1.0D / x], m2[1.0D / x] FROM test_map_lookup_shared + +-- With the setting off the dispatcher keeps a repeat, so an aggregate counts it as shared +query expect_error(DIVIDE_BY_ZERO) +SELECT max(element_at(m1, 1.0D / x + 1.0D / x)) FROM test_map_lookup_shared + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT max(element_at(m1, 1.0D / (x + 1.0D) + 1.0D / (x + 1.0D))) FROM test_map_lookup_shared + +query expect_dispatch(element_at) +SELECT element_at(m1, 1.0D / x + 1.0D / x) FROM test_map_lookup_shared + +query expect_dispatch(element_at) +SELECT element_at(m1, 1.0D / x) FROM test_map_lookup_shared + +-- The decline does not depend on ANSI mode +statement +SET spark.sql.subexpressionElimination.enabled=true + +statement +SET spark.sql.ansi.enabled=false + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT element_at(m1, 1.0D / x), element_at(m2, 1.0D / x) FROM test_map_lookup_shared diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression_decimal.sql b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression_decimal.sql new file mode 100644 index 00000000000..00ef7271720 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression_decimal.sql @@ -0,0 +1,56 @@ +-- 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. + +-- As map_lookup_shared_subexpression.sql, but the shared key holds decimal arithmetic, which the +-- serde wraps in CheckOverflow before the dispatcher sees it. + +-- Config: spark.sql.ansi.enabled=true +-- Config: spark.sql.subexpressionElimination.enabled=true + +-- On the last row both `d / x` and `d / (x + 1)` find a key. +statement +CREATE TABLE test_map_lookup_shared_decimal( + m1 map, m2 map, d decimal(10, 0), x decimal(10, 0)) +USING parquet + +statement +INSERT INTO test_map_lookup_shared_decimal VALUES + (CAST(NULL AS MAP), CAST(NULL AS MAP), 1, 0), + (map(0.5D, 5), map(0.5D, 6), 1, 3), + (map(1.0D, 7, 0.5D, 8), map(1.0D, 9, 0.5D, 10), 1, 1) + +-- The shared `d / x` divides by zero on the row whose maps are NULL. +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(m1, CAST(d / x AS DOUBLE)), element_at(m2, CAST(d / x AS DOUBLE)) +FROM test_map_lookup_shared_decimal + +query expect_error(DIVIDE_BY_ZERO) +SELECT m1[CAST(d / x AS DOUBLE)], m2[CAST(d / x AS DOUBLE)] FROM test_map_lookup_shared_decimal + +-- `x + 1` is never zero, yet the shared key still declines the dispatcher +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT element_at(m1, CAST(d / (x + 1) AS DOUBLE)), element_at(m2, CAST(d / (x + 1) AS DOUBLE)) +FROM test_map_lookup_shared_decimal + +-- The dispatcher eliminates a subexpression the lookup itself repeats, as Spark does +query expect_dispatch(element_at) +SELECT element_at(m1, CAST(d / (x + 1) AS DOUBLE) + CAST(d / (x + 1) AS DOUBLE)) +FROM test_map_lookup_shared_decimal + +-- A single lookup shares nothing, so Spark skips the divide on the NULL map too. +query expect_dispatch(element_at) +SELECT element_at(m1, CAST(d / x AS DOUBLE)) FROM test_map_lookup_shared_decimal diff --git a/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression_operators.sql b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression_operators.sql new file mode 100644 index 00000000000..50bfb3e9979 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/map/map_lookup_shared_subexpression_operators.sql @@ -0,0 +1,88 @@ +-- 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. + +-- Shuffle keys and a TakeOrderedAndProject's projection share subexpressions like a project list, +-- so the planner must decline in the scope execution serializes them in. + +-- Config: spark.sql.ansi.enabled=true +-- Config: spark.sql.subexpressionElimination.enabled=true +-- Config: spark.sql.shuffle.partitions=10 + +statement +CREATE TABLE test_map_lookup_shared_ops(m1 map, m2 map, x double) +USING parquet + +-- `1.0D / (x + 1.0D)`, whose divisor is never zero, finds a key in the last three rows. +statement +INSERT INTO test_map_lookup_shared_ops VALUES + (CAST(NULL AS MAP), CAST(NULL AS MAP), 0.0D), + (map(0.5D, 5), map(0.5D, 6), 2.0D), + (map(0.5D, 7), map(0.5D, 8), 1.0D), + (map(0.25D, 1), map(0.25D, 2), 3.0D), + (map(0.125D, 3), map(0.125D, 4), 7.0D) + +-- Native shuffle declines; JVM columnar shuffle evaluates the keys with Spark's projection +query expect_error(DIVIDE_BY_ZERO) +SELECT x +FROM test_map_lookup_shared_ops +DISTRIBUTE BY element_at(m1, 1.0D / x), element_at(m2, 1.0D / x) + +query +SELECT x +FROM test_map_lookup_shared_ops +DISTRIBUTE BY element_at(m1, 1.0D / (x + 1.0D)), element_at(m2, 1.0D / (x + 1.0D)) + +query expect_error(DIVIDE_BY_ZERO) +SELECT x +FROM test_map_lookup_shared_ops +ORDER BY element_at(m1, 1.0D / x), element_at(m2, 1.0D / x) + +-- Two rows miss both lookups, so `x` breaks the tie. +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT x +FROM test_map_lookup_shared_ops +ORDER BY element_at(m1, 1.0D / (x + 1.0D)), element_at(m2, 1.0D / (x + 1.0D)), x + +-- The subquery keeps the projection above the sort, as a TakeOrderedAndProject +query expect_error(DIVIDE_BY_ZERO) +SELECT element_at(m1, 1.0D / x), element_at(m2, 1.0D / x) +FROM (SELECT m1, m2, x FROM test_map_lookup_shared_ops ORDER BY x) +LIMIT 10 + +query expect_fallback(unsupported projection) +SELECT element_at(m1, 1.0D / (x + 1.0D)), element_at(m2, 1.0D / (x + 1.0D)) +FROM (SELECT m1, m2, x FROM test_map_lookup_shared_ops ORDER BY x) +LIMIT 10 + +-- The decline does not depend on ANSI mode +statement +SET spark.sql.ansi.enabled=false + +query +SELECT x +FROM test_map_lookup_shared_ops +DISTRIBUTE BY element_at(m1, 1.0D / x), element_at(m2, 1.0D / x) + +query expect_fallback(holds a subexpression that the operator's other expressions share) +SELECT x +FROM test_map_lookup_shared_ops +ORDER BY element_at(m1, 1.0D / x), element_at(m2, 1.0D / x), x + +query expect_fallback(unsupported projection) +SELECT element_at(m1, 1.0D / x), element_at(m2, 1.0D / x) +FROM (SELECT m1, m2, x FROM test_map_lookup_shared_ops ORDER BY x) +LIMIT 10 diff --git a/spark/src/test/resources/sql-tests/expressions/misc/codegen_dispatch_calendar_interval_output.sql b/spark/src/test/resources/sql-tests/expressions/misc/codegen_dispatch_calendar_interval_output.sql new file mode 100644 index 00000000000..1bcad80a7ef --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/misc/codegen_dispatch_calendar_interval_output.sql @@ -0,0 +1,64 @@ +-- 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. + +-- The dispatcher's interval conversion overflows past about 292 years (#5279), so a would-be +-- dispatched expression whose result holds a calendar interval falls back to Spark. + +statement +CREATE TABLE test_interval_output_big(s decimal(18, 6)) USING parquet + +statement +INSERT INTO test_interval_output_big VALUES (10000000000.000000), (-10000000000.000000), (NULL) + +query expect_fallback(holds a calendar interval) +SELECT arrays_zip(array(make_interval(0, 0, 0, 0, 0, 0, s))) FROM test_interval_output_big + +query expect_fallback(holds a calendar interval) +SELECT arrays_zip(array(named_struct('i', make_interval(0, 0, 0, 0, 0, 0, s)))) FROM test_interval_output_big + +query expect_fallback(holds a calendar interval) +SELECT arrays_zip(array(map(1, make_interval(0, 0, 0, 0, 0, 0, s)))) FROM test_interval_output_big + +-- A floating-point map key has no native lookup, so these would dispatch. +query expect_fallback(holds a calendar interval) +SELECT element_at(map_from_arrays(array(1.0D), array(make_interval(0, 0, 0, 0, 0, 0, s))), 1.0D) +FROM test_interval_output_big + +query expect_fallback(holds a calendar interval) +SELECT map_from_arrays(array(1.0D), array(make_interval(0, 0, 0, 0, 0, 0, s)))[1.0D] +FROM test_interval_output_big + +-- Native map construction compares floating-point keys by bits, so this would dispatch. +query expect_fallback(holds a calendar interval) +SELECT map_from_arrays(array(1.0D), array(make_interval(0, 0, 0, 0, 0, 0, s))) +FROM test_interval_output_big + +-- A nondeterministic child has no native path, so this would dispatch. +query expect_fallback(holds a calendar interval) +SELECT map_from_arrays(array(monotonically_increasing_id()), array(make_interval(0, 0, 0, 0, 0, 0, s))) +FROM test_interval_output_big + +query expect_dispatch(arrays_zip) +SELECT arrays_zip(array(map(1, s))) FROM test_interval_output_big + +-- Under LAST_WIN `map_from_arrays` has no native path for any key type, so this would dispatch. +statement +SET spark.sql.mapKeyDedupPolicy=LAST_WIN + +query expect_fallback(holds a calendar interval) +SELECT map_from_arrays(array('a'), array(make_interval(0, 0, 0, 0, 0, 0, s))) +FROM test_interval_output_big diff --git a/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala index 6b8202d240c..581505dcd51 100644 --- a/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala @@ -386,35 +386,6 @@ class CometMapExpressionSuite extends CometTestBase { } } - // Finding E: a map nested inside a map value is handed to the JVM dispatcher whole, so its double - // keys never revisit `CometLiteral`. The guard has to live on the lookup instead. The outer key - // type is INT, so inspecting only the outermost map would admit the query; the fallback comes - // from the inner `element_at`, whose child is `MapType(DoubleType, IntegerType)`. Constant folding - // is on here, which is the only configuration where the inner map becomes such a literal, so this - // regression cannot be expressed in a SQL fixture (the harness disables folding). The direct - // single-level double-key lookups live in `element_at_map.sql` / `get_map_value.sql`. - test("nested map lookup with floating-point keys falls back") { - withParquetTable((1 until 4).map(i => (i, i.toLong)), "tbl") { - val negZero = "CAST(concat('-', CAST(_1 - 1 AS STRING), '.0') AS DOUBLE)" - checkSparkAnswerAndFallbackReason( - "SELECT _1 AS id, element_at(element_at(map(1, map(CAST(0 AS DOUBLE), 7)), _1), " + - s"$negZero) AS v FROM tbl", - "Spark looks a key up in its SQL ordering") - } - } - - // Finding E for a collated inner key. Same folding-on-only bypass as the double-key case above; - // the direct single-level collated lookup lives in `element_at_map_collation.sql`. - test("nested map lookup with collated string keys falls back") { - assume(isSpark40Plus) - withParquetTable(Seq(("a1", 0)), "tbl") { - checkSparkAnswerAndFallbackReason( - "SELECT element_at(element_at(map(1, map(CAST('A1' AS STRING COLLATE UTF8_LCASE), 7)), 1), " + - "CAST(_1 AS STRING COLLATE UTF8_LCASE)) AS v FROM tbl", - "cannot honour a non-default collation") - } - } - // A folded map with `CalendarIntervalType` keys reaches `CometLiteral`. `ArrayBasedMapBuilder` // dedups such keys by hash equality, but the folded-literal duplicate-key check needs an // interpreted ordering and `PhysicalCalendarIntervalType` has none ("does not support ordered @@ -488,19 +459,6 @@ class CometMapExpressionSuite extends CometTestBase { } } - // Direct single-level folded map with floating-point keys: `map(CAST(0 AS DOUBLE), 7)` folds to a - // literal and `element_at` with a dynamic `-0.0` lookup must match Spark's `+0.0` key. - // Native `map_extract` compares raw Arrow values, so `MapKeySupport` declines it at `element_at`. - // Spark returns 7, NULL, NULL. (`element_at_map.sql` covers the folding-off constructor form.) - test("folded map literal with floating-point keys in element_at falls back (multirow)") { - withParquetTable((1 until 4).map(i => (i, i.toLong)), "tbl") { - val lookup = "CAST(concat('-', CAST(_1 - 1 AS STRING), '.0') AS DOUBLE)" - checkSparkAnswerAndFallbackReason( - s"SELECT _1 AS id, element_at(map(CAST(0 AS DOUBLE), 7), $lookup) AS v FROM tbl", - "Spark looks a key up in its SQL ordering") - } - } - // A TRY cast over a foldable map folds to a literal whose failing key is null. With a single // entry it has to stay on Spark too, rather than being rebuilt as a `CreateMap` that rejects the // null key. Spark returns NULL for both rows. @@ -514,31 +472,6 @@ class CometMapExpressionSuite extends CometTestBase { } } - // Direct single-level folded map with collated string keys. Native lookup is bytewise, so a - // case-insensitive `a1` lookup against a stored `A1` cannot match; `MapKeySupport` declines it. - test("folded map literal with collated string keys in element_at falls back (multirow)") { - assume(isSpark40Plus) - withParquetTable(Seq(("a1", 0)), "tbl") { - checkSparkAnswerAndFallbackReason( - "SELECT element_at(map(CAST('A1' AS STRING COLLATE UTF8_LCASE), 7), " + - "CAST(_1 AS STRING COLLATE UTF8_LCASE)) AS v FROM tbl", - "cannot honour a non-default collation") - } - } - - // Folded map with a complex (array) key. Spark permits a dynamic lookup array containing a NULL - // element; native `map_extract` casts the lookup to the key's exact Arrow type and cannot - // reproduce Spark's equality, so `MapKeySupport` declines every complex key type. Spark returns - // 7, NULL, NULL. (`element_at_map.sql` covers the constructor form with a non-null lookup.) - test("folded map literal with complex array key in element_at falls back (multirow)") { - withParquetTable((1 until 4).map(i => (i, i.toLong)), "tbl") { - checkSparkAnswerAndFallbackReason( - "SELECT _1 AS id, element_at(map(array(1), 7), " + - "array(IF(_1 = 2, CAST(NULL AS INT), _1))) AS v FROM tbl", - "casts the lookup key to the map's exact Arrow key type") - } - } - // A folded nested INT-keyed map is admitted natively (all key-type guards pass), so the outer // `element_at` runs as native `map_extract`. With ANSI disabled, a remainder-by-zero lookup key // evaluates to NULL instead of throwing, so `map_extract(NULL_map, NULL)` returns NULL and the