feat: support non-foldable sha2 natively

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

@@ -54,14 +54,14 @@
 ## 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 `SparkSha2` UDF for both foldable and column `numBits`. Tests cover STRING and BINARY payload columns with row-varying bit lengths, including invalid values and NULL, with the JVM codegen dispatcher enabled and disabled.
 - 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.
 
@@ -70,7 +70,7 @@
 - Performance (tuned 2026-09-28, issue #5994): the shared typed wide-decimal list path was 2.1–15.8x faster in local Criterion measurements across 8,192-row `decimal(38,0)` List, LargeList and FixedSizeList inputs with 2/32 elements and no/sparse/dense nulls; paired scalar remeasurement found no significant slowdown. Benchmark: `benches/decimal_hash.rs`.
 
 - 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 decimals, including nested decimals, hash natively as for `hash`; `TimeType` falls back to Spark.
 - Spark 4.0.1 (audited 2026-05-27): semantics unchanged.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
 - Native routing: Comet's native `xxhash64` UDF delegates compatible arguments at Spark's default seed (`42`) to `datafusion-spark::SparkXxhash64`. Differential tests in `native/spark-expr/src/hash_funcs/xxhash64_diff.rs` compare Comet's kernel against `SparkXxhash64` for primitives, narrow decimals, dictionaries, lists, maps, and nested combinations, and record where the two differ on a NaN whose bits are not canonical. Wide-decimal tests verify that the expression stays on Comet's kernel: the pinned DataFusion implementation uses fixed-width little-endian bytes rather than Spark's minimal signed big-endian bytes. The Comet kernel is retained for wide decimals (also within nested types), a non-default seed (`SparkXxhash64` hardcodes 42), `Struct` (upstream does not push a parent null mask into children; see #5753), a `Dictionary` nested in a list/map (upstream restarts those hashes from 42), `Time64`, and `Float32`/`Float64`, including values nested in lists and maps (upstream hashes the raw bits of a NaN, where Spark hashes every NaN as the canonical NaN; see [apache/datafusion#25913](https://github.com/apache/datafusion/issues/25913)). `create_xxhash64_hashes` remains for `approx_count_distinct` and the non-delegated native path.

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

@@ -360,9 +360,9 @@ to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enab
 | `crc32` | ✅ | Native |  |
 | `hash` | ✅ | Native | Includes decimals with precision 19–38 and nested decimals |
 | `md5` | ✅ | Native |  |
-| `sha` | ✅ | Native |  |
+| `sha` | ✅ | Native | Alias of `sha1` |
 | `sha1` | ✅ | Native |  |
-| `sha2` | ✅ | Native |  |
+| `sha2` | ✅ | Native | Supports literal and column `numBits` ([audit](../../contributor-guide/expression-audits/hash_funcs.md#sha2)) |
 | `xxhash64` | ✅ | Native | Includes decimals with precision 19–38 and nested decimals |
 
 ---

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

@@ -22,7 +22,7 @@ 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, IntegerType, LongType, MapType, StringType, StructType}
 
-import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, isTimeType, scalarFunctionExprToProtoWithReturnType, serializeDataType, supportedDataType}
+import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, isTimeType, scalarFunctionExprToProtoWithReturnType, serializeDataType, supportedDataType, withNullShortCircuit}
 
 object CometXxHash64 extends CometExpressionSerde[XxHash64] {
 
@@ -74,27 +74,20 @@ object CometMurmur3Hash extends CometExpressionSerde[Murmur3Hash] {
 
 object CometSha2 extends CometExpressionSerde[Sha2] {
 
-  private val nonFoldableNumBitsReason =
-    "The `numBits` argument must be a foldable literal value"
-
-  override def getUnsupportedReasons(): Seq[String] =
-    HashUtils.unsupportedReasons :+ nonFoldableNumBitsReason
+  override def getUnsupportedReasons(): Seq[String] = HashUtils.unsupportedReasons
 
-  override def getSupportLevel(expr: Sha2): SupportLevel = {
-    if (!expr.right.foldable) {
-      Unsupported(Some(nonFoldableNumBitsReason))
-    } else {
-      HashUtils.supportLevelForChildren(expr)
-    }
-  }
+  override def getSupportLevel(expr: Sha2): SupportLevel =
+    HashUtils.supportLevelForChildren(expr)
 
   override def convert(
       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)
+    withNullShortCircuit(
+      expr,
+      scalarFunctionExprToProtoWithReturnType("sha2", StringType, false, leftExpr, numBitsExpr))
   }
 }
 

spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala

@@ -24,14 +24,16 @@ import scala.util.Random
 import org.apache.spark.sql.CometTestBase
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 
+import org.apache.comet.CometSparkSessionExtensions.isSpark41Plus
 import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions}
 import org.apache.comet.udf.codegen.CometScalaUDFCodegen
 
 /**
  * Test suite for Spark `hash` (murmur3) and `xxhash64` compatibility between Spark and Comet.
  *
- * These tests verify that Comet's native implementations produce identical results to Spark for
- * all supported data types.
+ * Native kernels are asserted for supported input shapes, including wide and nested decimals.
+ * `sha2` supports both literal and column `numBits` natively. `TimeType` hash inputs fall back to
+ * Spark.
  */
 class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
 
@@ -113,6 +115,77 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe
     }
   }
 
+  test("hash - TimeType falls back to Spark") {
+    assume(isSpark41Plus, "TimeType requires Spark 4.1+")
+    withSQLConf("spark.sql.timeType.enabled" -> "true") {
+      withTable("t") {
+        sql("CREATE TABLE t(c STRING) USING parquet")
+        sql("INSERT INTO t VALUES ('12:34:56'), ('00:00:00'), (null)")
+        checkSparkAnswerAndFallbackReasons(
+          "SELECT hash(to_time(c)) FROM t",
+          Set("`TimeType` is not supported"))
+        checkSparkAnswerAndFallbackReasons(
+          "SELECT xxhash64(to_time(c)) FROM t",
+          Set("`TimeType` is not supported"))
+      }
+    }
+  }
+
+  Seq("STRING", "BINARY").foreach { payloadType =>
+    test(s"sha2 - non-foldable numBits stays native with $payloadType payload") {
+      withTable("t") {
+        sql(s"CREATE TABLE t(payload $payloadType, num_bits INT) USING parquet")
+        val payload = if (payloadType == "BINARY") "X'0080FF68656C6C6F'" else "'hello'"
+        val empty = if (payloadType == "BINARY") "X''" else "''"
+        val values = Seq("0", "224", "256", "384", "512", "128", "-1", "NULL")
+          .map(bits => s"($payload, $bits)") ++
+          Seq(s"($empty, 256)", s"($empty, 512)", "(NULL, 256)")
+        sql(s"INSERT INTO t VALUES ${values.mkString(",")}")
+        Seq("true", "false").foreach { enabled =>
+          withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> enabled) {
+            Seq(
+              "SELECT sha2(payload, num_bits) FROM t",
+              "SELECT sha2(payload, num_bits + 0) FROM t",
+              s"SELECT sha2($payload, num_bits) FROM t").foreach(assertNativeHash)
+          }
+        }
+      }
+    }
+  }
+
+  Seq("STRING", "BINARY").foreach { payloadType =>
+    test(s"sha2 - NULL $payloadType payload short-circuits ANSI numBits cast") {
+      withSQLConf("spark.sql.ansi.enabled" -> "true") {
+        withTable("t") {
+          sql(s"CREATE TABLE t(payload $payloadType, bits STRING) USING parquet")
+          val payload = if (payloadType == "BINARY") "X'0080FF'" else "'hello'"
+          sql(
+            s"INSERT INTO t VALUES (NULL, 'invalid'), ($payload, '256'), " +
+              s"($payload, NULL), (NULL, '256')")
+          Seq("true", "false").foreach { enabled =>
+            withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> enabled) {
+              assertNativeHash("SELECT sha2(payload, CAST(bits AS INT)) FROM t")
+            }
+          }
+        }
+      }
+    }
+  }
+
+  test("sha2 - literal numBits stays native") {
+    withTable("t") {
+      sql("CREATE TABLE t(payload STRING) USING parquet")
+      sql("INSERT INTO t VALUES ('hello'), (''), (NULL)")
+      Seq("true", "false").foreach { enabled =>
+        withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> enabled) {
+          Seq("0", "224", "256", "384", "512", "128", "-1", "128 + 128").foreach { bits =>
+            assertNativeHash(s"SELECT sha2(payload, $bits) FROM t")
+          }
+        }
+      }
+    }
+  }
+
   test("hash - array of integers") {
     withTable("t") {
       sql("CREATE TABLE t(c ARRAY<INT>) USING parquet")
@@ -395,4 +468,12 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe
       }
     }
   }
+
+  private def assertNativeHash(query: String): Unit = {
+    CometScalaUDFCodegen.resetStats()
+    checkSparkAnswerAndOperator(query)
+    assert(
+      CometScalaUDFCodegen.stats().totalLookups == 0,
+      s"expected native hash execution for $query, got ${CometScalaUDFCodegen.stats()}")
+  }
 }