feat: route wide-decimal hash and non-literal sha2 through the codege…

docs/source/contributor-guide/expression-audits/hash_funcs.md

@@ -34,7 +34,7 @@
 - Spark 3.5.8 (audited 2026-05-27): baseline. `Murmur3Hash(children, seed) extends HashExpression[Int]`; produces a Murmur3 hash with a configurable Int seed and `IntegerType` result. Comet routes via `CometMurmur3Hash` to the native `murmur3_hash` UDF.
 - Spark 4.0.1 (audited 2026-05-27): semantics unchanged; some inner helper refactors only.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
-- Known limitation: `DecimalType` children with precision > 18 fall back because Spark hashes them through Java `BigDecimal`; `TimeType` (Spark 4.0+) is also unsupported. The same limitations apply to `xxhash64`, `sha1`, `sha2` through the shared `HashUtils`.
+- Known limitation: the native kernel does not hash `DecimalType` children with precision > 18 (Spark hashes them through Java `BigDecimal`), including when nested in array, struct, or map. With the JVM codegen dispatcher enabled (the default), `CodegenDispatchFallback` runs Spark's `HashExpression.doGenCode` inside the Comet pipeline so the enclosing operator stays native. The projection falls back to Spark only when the dispatcher is disabled or refuses the tree. `TimeType` is out of scope for that dispatcher enrollment: `getSupportLevel` reports `Compatible` so the mixin does not intercept it, and `convert` declines the native path so the projection falls back to Spark. The same wide-decimal routing applies to `xxhash64`.
 
 ## md5
 
@@ -53,21 +53,21 @@
 ## sha1
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `Sha1(child) extends UnaryExpression with NullIntolerant`; `inputTypes = Seq(BinaryType) -> StringType`. Comet routes via `CometSha1` to the native `sha1` UDF.
+- Spark 3.5.8 (audited 2026-05-27): baseline. `Sha1(child) extends UnaryExpression with NullIntolerant`; `inputTypes = Seq(BinaryType) -> StringType`. The Analyzer casts accepted inputs to `BinaryType` before Comet serde, and `CometSha1` routes the resulting binary child to the native `sha1` UDF.
 - Spark 4.0.1 (audited 2026-05-27): trait set gains `DefaultStringProducingExpression` and `NullIntolerant` is replaced by `nullIntolerant: Boolean`. Runtime unchanged.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
 
 ## sha2
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `Sha2(left, right) extends BinaryExpression`; `inputTypes = Seq(BinaryType, IntegerType) -> StringType`. The `numBits` argument selects SHA-224/256/384/512 (0 is treated as 256); other values return NULL. Comet routes via `CometSha2` to the native `sha2` UDF; non-foldable `numBits` falls back to Spark.
+- Spark 3.5.8 (audited 2026-05-27): baseline. `Sha2(left, right) extends BinaryExpression`; `inputTypes = Seq(BinaryType, IntegerType) -> StringType`. The `numBits` argument selects SHA-224/256/384/512 (0 is treated as 256); other values return NULL. Comet routes via `CometSha2` to the native `sha2` UDF when `numBits` is foldable. A non-foldable `numBits` has no native path; with the JVM codegen dispatcher enabled (the default), `CodegenDispatchFallback` runs Spark's `Sha2.doGenCode` inside the Comet pipeline. The projection falls back to Spark only when the dispatcher is disabled or refuses the tree.
 - Spark 4.0.1 (audited 2026-05-27): trait set gains `DefaultStringProducingExpression` and the `nullIntolerant: Boolean` refactor. Runtime unchanged.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
 
 ## xxhash64
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `XxHash64(children, seed) extends HashExpression[Long]`; produces an xxHash64 hash with a configurable Long seed and `LongType` result. Comet routes via `CometXxHash64` to the native `xxhash64` UDF.
+- Spark 3.5.8 (audited 2026-05-27): baseline. `XxHash64(children, seed) extends HashExpression[Long]`; produces an xxHash64 hash with a configurable Long seed and `LongType` result. Comet routes via `CometXxHash64` to the native `xxhash64` UDF. Wide-decimal routing matches `hash` via the shared `HashUtils` and `CodegenDispatchFallback`; `TimeType` is likewise excluded from dispatcher enrollment.
 - Spark 4.0.1 (audited 2026-05-27): semantics unchanged.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
 

docs/source/user-guide/latest/expressions.md

@@ -349,12 +349,12 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | Function | Status | Implementation | Notes |
 | --- | --- | --- | --- |
 | `crc32` | ✅ | Native |  |
-| `hash` | ✅ | Native |  |
+| `hash` | ✅ | Hybrid | Decimal precision > 18 (including nested) routes through the JVM codegen dispatcher ([audit](../../contributor-guide/expression-audits/hash_funcs.md#hash)) |
 | `md5` | ✅ | Native |  |
-| `sha` | ✅ | Native |  |
+| `sha` | ✅ | Native | Alias of `sha1` |
 | `sha1` | ✅ | Native |  |
-| `sha2` | ✅ | Native |  |
-| `xxhash64` | ✅ | Native |  |
+| `sha2` | ✅ | Hybrid | Non-foldable `numBits` routes through the JVM codegen dispatcher ([audit](../../contributor-guide/expression-audits/hash_funcs.md#sha2)) |
+| `xxhash64` | ✅ | Hybrid | Decimal precision > 18 (including nested) routes through the JVM codegen dispatcher ([audit](../../contributor-guide/expression-audits/hash_funcs.md#hash)) |
 
 ---
 

spark/src/main/scala/org/apache/comet/serde/hash.scala

@@ -22,9 +22,15 @@ package org.apache.comet.serde
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, Murmur3Hash, Sha1, Sha2, XxHash64}
 import org.apache.spark.sql.types.{ArrayType, DataType, DecimalType, IntegerType, LongType, MapType, StringType, StructType}
 
+import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, isTimeType, scalarFunctionExprToProtoWithReturnType, serializeDataType, supportedDataType}
 
-object CometXxHash64 extends CometExpressionSerde[XxHash64] {
+// Native-unsupported, Spark-codegen-compatible cases (`DecimalType` precision > 18, including
+// nested, and `sha2` with a non-foldable `numBits`) stay in the Comet pipeline via
+// `CodegenDispatchFallback` on the applicable hash serdes below. `TimeType` is out of scope for
+// that dispatcher enrollment: `getSupportLevel` reports `Compatible` so the mixin does not
+// intercept, and `convert` declines the native path so the projection falls back to Spark.
+object CometXxHash64 extends CometExpressionSerde[XxHash64] with CodegenDispatchFallback {
 
   override def getUnsupportedReasons(): Seq[String] = HashUtils.unsupportedReasons
 
@@ -35,18 +41,20 @@ object CometXxHash64 extends CometExpressionSerde[XxHash64] {
       expr: XxHash64,
       inputs: Seq[Attribute],
       binding: Boolean): Option[ExprOuterClass.Expr] = {
-    val exprs = expr.children.map(exprToProtoInternal(_, inputs, binding))
-    val seedBuilder = LiteralOuterClass.Literal
-      .newBuilder()
-      .setDatatype(serializeDataType(LongType).get)
-      .setLongVal(expr.seed)
-    val seedExpr = Some(ExprOuterClass.Expr.newBuilder().setLiteral(seedBuilder).build())
-    // the seed is put at the end of the arguments
-    scalarFunctionExprToProtoWithReturnType("xxhash64", LongType, false, exprs :+ seedExpr: _*)
+    HashUtils.convertNativeOrSparkFallback(expr) {
+      val exprs = expr.children.map(exprToProtoInternal(_, inputs, binding))
+      val seedBuilder = LiteralOuterClass.Literal
+        .newBuilder()
+        .setDatatype(serializeDataType(LongType).get)
+        .setLongVal(expr.seed)
+      val seedExpr = Some(ExprOuterClass.Expr.newBuilder().setLiteral(seedBuilder).build())
+      // the seed is put at the end of the arguments
+      scalarFunctionExprToProtoWithReturnType("xxhash64", LongType, false, exprs :+ seedExpr: _*)
+    }
   }
 }
 
-object CometMurmur3Hash extends CometExpressionSerde[Murmur3Hash] {
+object CometMurmur3Hash extends CometExpressionSerde[Murmur3Hash] with CodegenDispatchFallback {
 
   override def getUnsupportedReasons(): Seq[String] = HashUtils.unsupportedReasons
 
@@ -57,22 +65,24 @@ object CometMurmur3Hash extends CometExpressionSerde[Murmur3Hash] {
       expr: Murmur3Hash,
       inputs: Seq[Attribute],
       binding: Boolean): Option[ExprOuterClass.Expr] = {
-    val exprs = expr.children.map(exprToProtoInternal(_, inputs, binding))
-    val seedBuilder = LiteralOuterClass.Literal
-      .newBuilder()
-      .setDatatype(serializeDataType(IntegerType).get)
-      .setIntVal(expr.seed)
-    val seedExpr = Some(ExprOuterClass.Expr.newBuilder().setLiteral(seedBuilder).build())
-    // the seed is put at the end of the arguments
-    scalarFunctionExprToProtoWithReturnType(
-      "murmur3_hash",
-      IntegerType,
-      false,
-      exprs :+ seedExpr: _*)
+    HashUtils.convertNativeOrSparkFallback(expr) {
+      val exprs = expr.children.map(exprToProtoInternal(_, inputs, binding))
+      val seedBuilder = LiteralOuterClass.Literal
+        .newBuilder()
+        .setDatatype(serializeDataType(IntegerType).get)
+        .setIntVal(expr.seed)
+      val seedExpr = Some(ExprOuterClass.Expr.newBuilder().setLiteral(seedBuilder).build())
+      // the seed is put at the end of the arguments
+      scalarFunctionExprToProtoWithReturnType(
+        "murmur3_hash",
+        IntegerType,
+        false,
+        exprs :+ seedExpr: _*)
+    }
   }
 }
 
-object CometSha2 extends CometExpressionSerde[Sha2] {
+object CometSha2 extends CometExpressionSerde[Sha2] with CodegenDispatchFallback {
 
   private val nonFoldableNumBitsReason =
     "The `numBits` argument must be a foldable literal value"
@@ -81,7 +91,11 @@ object CometSha2 extends CometExpressionSerde[Sha2] {
     HashUtils.unsupportedReasons :+ nonFoldableNumBitsReason
 
   override def getSupportLevel(expr: Sha2): SupportLevel = {
-    if (!expr.right.foldable) {
+    // TimeType is not enrolled in the dispatcher; check it before the non-foldable `numBits`
+    // `Unsupported` so a mixed tree does not take the mixin path.
+    if (HashUtils.containsTimeTypeInChildren(expr)) {
+      Compatible()
+    } else if (!expr.right.foldable) {
       Unsupported(Some(nonFoldableNumBitsReason))
     } else {
       HashUtils.supportLevelForChildren(expr)
@@ -92,16 +106,16 @@ object CometSha2 extends CometExpressionSerde[Sha2] {
       expr: Sha2,
       inputs: Seq[Attribute],
       binding: Boolean): Option[ExprOuterClass.Expr] = {
-    val leftExpr = exprToProtoInternal(expr.left, inputs, binding)
-    val numBitsExpr = exprToProtoInternal(expr.right, inputs, binding)
-    scalarFunctionExprToProtoWithReturnType("sha2", StringType, false, leftExpr, numBitsExpr)
+    HashUtils.convertNativeOrSparkFallback(expr) {
+      val leftExpr = exprToProtoInternal(expr.left, inputs, binding)
+      val numBitsExpr = exprToProtoInternal(expr.right, inputs, binding)
+      scalarFunctionExprToProtoWithReturnType("sha2", StringType, false, leftExpr, numBitsExpr)
+    }
   }
 }
 
 object CometSha1 extends CometExpressionSerde[Sha1] {
 
-  override def getUnsupportedReasons(): Seq[String] = HashUtils.unsupportedReasons
-
   override def getSupportLevel(expr: Sha1): SupportLevel =
     HashUtils.supportLevelForChildren(expr)
 
@@ -120,27 +134,53 @@ private object HashUtils {
     "`DecimalType` with precision > 18 is not supported (Spark hashes via Java `BigDecimal`)"
   private val unsupportedTimeTypeReason = "`TimeType` is not supported"
 
+  // `TimeType` is omitted: `CodegenDispatchFallback` documents this list as JVM-dispatch cases.
   val unsupportedReasons: Seq[String] =
-    Seq(unsupportedDecimalReason, unsupportedTimeTypeReason, "Unsupported child data type")
+    Seq(unsupportedDecimalReason, "Unsupported child data type")
+
+  def containsTimeTypeInChildren(expr: Expression): Boolean =
+    expr.children.exists(c => containsTimeType(c.dataType))
 
   def supportLevelForChildren(expr: Expression): SupportLevel = {
-    expr.children.iterator
-      .flatMap(c => unsupportedReasonFor(c.dataType).iterator)
-      .toSeq
-      .headOption match {
-      case Some(reason) => Unsupported(Some(reason))
-      case None => Compatible()
+    // Compatible (not Unsupported) so `CodegenDispatchFallback` does not enroll TimeType.
+    if (containsTimeTypeInChildren(expr)) {
+      Compatible()
+    } else {
+      expr.children.iterator
+        .flatMap(c => unsupportedReasonFor(c.dataType).iterator)
+        .toSeq
+        .headOption match {
+        case Some(reason) => Unsupported(Some(reason))
+        case None => Compatible()
+      }
     }
   }
 
+  def convertNativeOrSparkFallback(expr: Expression)(
+      native: => Option[ExprOuterClass.Expr]): Option[ExprOuterClass.Expr] = {
+    if (containsTimeTypeInChildren(expr)) {
+      withFallbackReason(expr, unsupportedTimeTypeReason)
+      None
+    } else {
+      native
+    }
+  }
+
+  private def containsTimeType(dt: DataType): Boolean = dt match {
+    case t if isTimeType(t) => true
+    case s: StructType => s.fields.exists(f => containsTimeType(f.dataType))
+    case a: ArrayType => containsTimeType(a.elementType)
+    case m: MapType => containsTimeType(m.keyType) || containsTimeType(m.valueType)
+    case _ => false
+  }
+
   private def unsupportedReasonFor(dt: DataType): Option[String] = dt match {
     case d: DecimalType if d.precision > 18 => Some(unsupportedDecimalReason)
     case s: StructType =>
       s.fields.iterator.flatMap(f => unsupportedReasonFor(f.dataType).iterator).toSeq.headOption
     case a: ArrayType => unsupportedReasonFor(a.elementType)
     case m: MapType =>
       unsupportedReasonFor(m.keyType).orElse(unsupportedReasonFor(m.valueType))
-    case dt if isTimeType(dt) => Some(unsupportedTimeTypeReason)
     case _ if !supportedDataType(dt, allowComplex = true) =>
       Some(s"Unsupported child data type: $dt")
     case _ => None