diff --git a/docs/source/contributor-guide/expression-audits/hash_funcs.md b/docs/source/contributor-guide/expression-audits/hash_funcs.md
index a294069df14..d6eddffeee5 100644
--- a/docs/source/contributor-guide/expression-audits/hash_funcs.md
+++ b/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.
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
index 784b93f5ebf..36d3ddafe39 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/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 |
 
 ---
diff --git a/spark/src/main/scala/org/apache/comet/serde/hash.scala b/spark/src/main/scala/org/apache/comet/serde/hash.scala
index 04cd184fe41..a0227a8a241 100644
--- a/spark/src/main/scala/org/apache/comet/serde/hash.scala
+++ b/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,19 +74,10 @@ 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,
@@ -94,7 +85,9 @@ object CometSha2 extends CometExpressionSerde[Sha2] {
       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))
   }
 }
 
diff --git a/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala
index 7e823fb347c..fc4d868d8d2 100644
--- a/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala
+++ b/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()}")
+  }
 }
diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometHashCodegenDispatchBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometHashCodegenDispatchBenchmark.scala
new file mode 100644
index 00000000000..97a9d43cc24
--- /dev/null
+++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometHashCodegenDispatchBenchmark.scala
@@ -0,0 +1,234 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.benchmark
+
+import java.nio.charset.StandardCharsets
+
+import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.optimizer.ConstantFolding
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+import org.apache.comet.udf.codegen.CometScalaUDFCodegen
+
+/**
+ * Benchmark of native non-foldable `sha2(payload, numBits)` against Spark.
+ *
+ * Both Comet arms execute SHA-2 natively, with the codegen dispatcher enabled and disabled. The
+ * mixed projection also measures native expressions alongside `sha2`.
+ *
+ * Before timing anything, every case is run through all three arms and the rows are compared, so
+ * a timing cannot come from an arm that computed something else.
+ *
+ * To run this benchmark:
+ * {{{
+ *   SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometHashCodegenDispatchBenchmark
+ * }}}
+ * Results will be written to "spark/benchmarks/CometHashCodegenDispatchBenchmark-**results.txt".
+ */
+object CometHashCodegenDispatchBenchmark extends CometBenchmarkBase {
+
+  /** Fewer rows than one Comet batch, so the query is a single batch of real work. */
+  private val SmallRows = 1024
+
+  /** ~128 batches at the default `spark.comet.batchSize`. */
+  private val LargeRows = 1024 * 1024
+
+  /**
+   * @param name
+   *   Case name, as it appears in the results table.
+   * @param query
+   *   The query to time. Every argument is a column so that the expression is evaluated per row.
+   */
+  private case class DispatchCase(name: String, query: String)
+
+  private def cases: Seq[DispatchCase] = Seq(
+    DispatchCase("sha2(payload, numBits)", "select sha2(c_str, c_bits) from parquetV1Table"),
+    DispatchCase(
+      "sha2 mixed projection",
+      "select length(c_str), c_int + 1, hash(c_int), sha2(c_str, c_bits) from parquetV1Table"))
+
+  private def noConstantFolding: (String, String) =
+    SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> excludedRulesWith(ConstantFolding.ruleName)
+
+  private def sparkConfigs: Seq[(String, String)] =
+    Seq(noConstantFolding, CometConf.COMET_ENABLED.key -> "false")
+
+  /** Native execution with the dispatcher enabled. */
+  private def dispatchConfigs: Seq[(String, String)] =
+    cometConfigs(dispatch = true)
+
+  /** Native execution with the dispatcher disabled. */
+  private def nativeConfigs: Seq[(String, String)] =
+    cometConfigs(dispatch = false)
+
+  private def cometConfigs(dispatch: Boolean): Seq[(String, String)] = Seq(
+    noConstantFolding,
+    CometConf.COMET_ENABLED.key -> "true",
+    CometConf.COMET_EXEC_ENABLED.key -> "true",
+    CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> dispatch.toString)
+
+  private val DispatchCaseName = "Comet native, dispatch on"
+  private val NativeCaseName = "Comet native, dispatch off"
+  private val SparkCaseName = "Spark (Comet disabled)"
+
+  override def runCometBenchmark(mainArgs: Array[String]): Unit = {
+    val selected = cases
+    runBenchmark("SHA-2 native: environment") {
+      emitEnvironment(selected)
+    }
+
+    withCorpus(SmallRows) {
+      selected.foreach(verifyArmsAgree(_, SmallRows))
+      selected.foreach(runSteadyState(_, SmallRows))
+    }
+    withCorpus(LargeRows) {
+      selected.foreach(verifyArmsAgree(_, LargeRows))
+      selected.foreach(runSteadyState(_, LargeRows))
+    }
+  }
+
+  private def emitEnvironment(selected: Seq[DispatchCase]): Unit = {
+    emit(s"Spark version: ${spark.version}")
+    emit(
+      s"Java version: ${System.getProperty("java.version")} " +
+        s"(${System.getProperty("java.vm.name")})")
+    emit(s"Scala version: ${scala.util.Properties.versionNumberString}")
+    emit(s"spark.master: ${spark.conf.get("spark.master", "<unset>")}")
+    emit(
+      s"${CometConf.COMET_BATCH_SIZE.key}: " +
+        CometConf.COMET_BATCH_SIZE.get(spark.sessionState.conf))
+    emit(
+      s"${SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key}: " +
+        spark.conf.get(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key))
+    emit(s"Dispatcher conf: ${CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key}")
+    emit("Steady-state tables: Spark's Benchmark defaults -- 2s of untimed warmup per case, then")
+    emit(
+      "  at least 2 iterations and at least 2s of timed iterations; the table reports best and")
+    emit("  average of the timed iterations.")
+    emit(s"Row counts: $SmallRows (sub-batch) and $LargeRows (multi-batch).")
+    emit("Payloads include nulls, empty strings and varying lengths; numBits varies per row.")
+    emit("Both Comet arms execute SHA-2 natively without using the dispatcher.")
+    emit(s"Cases: ${selected.map(_.name).mkString(", ")}")
+  }
+
+  /**
+   * Fails if the three arms disagree on `query`. Rows are compared as a sorted multiset because
+   * output order is not guaranteed across the different plans.
+   */
+  private def verifyArmsAgree(c: DispatchCase, rows: Int): Unit = {
+    def collect(configs: Seq[(String, String)]): Array[String] = {
+      // Assigned to a local rather than returned from the block: Spark 3.4 and 3.5 declare
+      // `SQLHelper.withSQLConf` as returning `Unit`; only Spark 4 has the result-returning form.
+      var collected: Array[Row] = Array.empty
+      withSQLConf(configs: _*) {
+        collected = spark.sql(c.query).collect()
+      }
+      collected
+        .map(
+          _.toSeq
+            .map {
+              case bytes: Array[Byte] => bytes.mkString("[", ",", "]")
+              case other => String.valueOf(other)
+            }
+            .mkString("|"))
+        .sorted
+    }
+
+    val expected = collect(sparkConfigs)
+    Seq(DispatchCaseName -> dispatchConfigs, NativeCaseName -> nativeConfigs).foreach {
+      case (armName, configs) =>
+        val actual = collect(configs)
+        assert(
+          expected.length == actual.length,
+          s"${c.name} @ $rows rows: Spark produced ${expected.length} rows, " +
+            s"$armName ${actual.length}")
+        expected.indices.find(i => expected(i) != actual(i)).foreach { i =>
+          throw new AssertionError(
+            s"${c.name} @ $rows rows: row $i differs -- Spark ${expected(i)}, " +
+              s"$armName ${actual(i)}")
+        }
+    }
+  }
+
+  private def runSteadyState(c: DispatchCase, rows: Int): Unit = {
+    runBenchmark(s"${c.name} -- $rows rows") {
+      val benchmark = new Benchmark(s"${c.name} -- $rows rows", rows.toLong, output = output)
+      checkPlans(c)
+      // Spark goes first so `Relative` reports the native speedup.
+      benchmark.addCase(SparkCaseName)(_ => runQuery(c.query, sparkConfigs))
+      benchmark.addCase(NativeCaseName)(_ => runQuery(c.query, nativeConfigs))
+      benchmark.addCase(DispatchCaseName)(_ => runQuery(c.query, dispatchConfigs))
+      benchmark.run()
+    }
+  }
+
+  /** Refuses to time a case whose actual routing does not match its label. */
+  private def checkPlans(c: DispatchCase): Unit = {
+    Seq(dispatchConfigs, nativeConfigs).foreach { configs =>
+      withSQLConf(configs: _*) {
+        CometScalaUDFCodegen.resetStats()
+        val df = spark.sql(c.query)
+        df.noop()
+        val nonComet = findFirstNonCometOperator(stripAQEPlan(df.queryExecution.executedPlan))
+        assert(nonComet.isEmpty, s"native projection fell back to Spark: $nonComet")
+        assert(CometScalaUDFCodegen.stats().totalLookups == 0, "native SHA-2 used the dispatcher")
+      }
+    }
+  }
+
+  private def runQuery(query: String, configs: Seq[(String, String)]): Unit =
+    withSQLConf(configs: _*) {
+      spark.sql(query).noop()
+    }
+
+  /** Builds `parquetV1Table` with `rows` rows of the corpus and drops it afterwards. */
+  private def withCorpus(rows: Int)(f: => Unit): Unit = {
+    withTempPath { dir =>
+      withTempTable(tbl, "parquetV1Table") {
+        spark.range(rows.toLong).createOrReplaceTempView(tbl)
+        prepareTable(dir, spark.sql(corpusQuery))
+        f
+      }
+    }
+  }
+
+  /**
+   * Inputs vary per row, including nulls and empty payloads, so neither engine can hoist the
+   * expression out of the row loop.
+   */
+  private def corpusQuery: String = {
+    val columns = Seq(
+      "c_str" -> ("CASE WHEN pmod(id, 31) = 0 THEN NULL WHEN pmod(id, 31) = 1 THEN '' " +
+        "ELSE REPEAT(CAST(id AS STRING), CAST(pmod(id, 16) AS INT) + 1) END"),
+      "c_int" -> "CAST(id AS INT)",
+      "c_bits" -> "element_at(array(0, 224, 256, 384, 512, 128, NULL), CAST(pmod(id, 7) AS INT) + 1)")
+    s"SELECT ${columns.map { case (name, expr) => s"$expr AS $name" }.mkString(", ")} FROM $tbl"
+  }
+
+  /** [[Benchmark]] tees console and results file; this benchmark's own tables need the same. */
+  private def emit(line: String): Unit = {
+    // scalastyle:off println
+    println(line)
+    // scalastyle:on println
+    output.foreach(_.write(s"$line\n".getBytes(StandardCharsets.UTF_8)))
+  }
+}
diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometHashExpressionBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometHashExpressionBenchmark.scala
index 6450b822fe2..e07e62f11c9 100644
--- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometHashExpressionBenchmark.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometHashExpressionBenchmark.scala
@@ -84,7 +84,7 @@ object CometHashExpressionBenchmark extends CometBenchmarkBase {
   }
 
   // Run only these cases with --wide-decimal. The fallback arm is the path used by
-  // wide-decimal hash on main before native support; codegen dispatch is tracked in #5835.
+  // wide-decimal hash on main before native support.
   private def runWideDecimalBenchmarks(values: Int): Unit = {
     runBenchmarkWithTable("Wide decimal hash", values) { rows =>
       withTempPath { dir =>