test: move expression-level codegen dispatcher tests to Comet SQL tests #6637

Part of #6615.

What / Why

CometCodegenSuite tests the JVM codegen dispatcher, which compiles Spark's own doGenCode into an Arrow batch kernel. Most of it is dispatcher internals and stays in Scala: ScalaUDF registration, kernel caching, Nondeterministic state, TaskContext, memory lifecycle and explain text. But about a third of the suite tests how a built-in Spark expression routes through the dispatcher, or what it returns. expect_dispatch / expect_native / expect_fallback can express those now. Static analysis on main at 3bc2faa found the following:

A built-in parent that always dispatches can stand in for an identity ScalaUDF. CometScalaUDF.emitJvmCodegenDispatch binds the whole subtree under the dispatched root: only column references become data arguments, and every inner node is reported as dispatched. Several serdes always dispatch:

• map(...), map_concat
• transform, transform_values, map_filter, zip_with, aggregate
• hypot, nanvl, pmod, add_months, Invoke

So map('k', array_max(flatten(a))) puts array_max and flatten in one kernel over the nested input, the same as idBin(array_max(flatten(a))), and expect_dispatch(array_max, flatten) proves it. The same trick lets the CometCodegenHOFSuite tests in #6628 convert fully rather than split.

Work

Line numbers refer to CometCodegenSuite.

JSON routing

• New: json/routing_json_{disabled,enabled,opt_in,opt_in_enabled}.sql, one file per allowIncompatible × codegen cell.
  • Covers L80 (json_array_length) and the two L151 instances (from_json, to_json). Issue Test dual-impl (native + codegen-dispatch) expressions consistently across the full routing matrix #4616's routing-matrix retrofit never reached JSON.
  • Use the Scala inputs: '[1,2,3]', '[]', 'not an array', NULL, and the struct/array JSON rows.
  • For from_json, use expect_fallback(from_json: spark.comet.exec.scalaUDF.codegen.enabled=false), expect_dispatch(from_json) or expect_native(from_json) per cell.
  • On Spark 4.x, json_array_length and to_json can't be named in expect_native / expect_dispatch. Both lower to StaticInvoke / Invoke there. Spark4xCometExprShim rebuilds the original expression but copies back only the fallback reasons, so the coverage tags are lost.
    • Use expect_fallback where a cell falls back, which works on 4.x, and a plain query elsewhere.
    • A fully Comet plan with the dispatcher off proves the native path. A dispatcher-off fallback with allowIncompatible=false proves there is no native path, and then a fully Comet plan with the dispatcher on proves dispatch.
    • Only "opt-in on, dispatcher on" for natively supported shapes stays unpinned on 4.x. A MaxSparkVersion: 3.5 copy could pin it on 3.x.
• struct/json_to_structs.sql: change L24 and L28 from spark_answer_only to expect_dispatch(from_json).

Arrays

• array/sequence.sql
  • L482: expect_native(sequence) on the explicit-step and default-step column queries.
  • L530: a d DATE table with sequence(d, DATE'2024-06-01', INTERVAL 1 MONTH) as expect_dispatch(sequence).
  • Upgrade L182-189 to expect_dispatch(sequence).
• array/routing_arrays_{enabled,disabled,opt_in}.sql (L514): sequence(x, y, CASE WHEN x <= y THEN 2 ELSE -2 END).
• array/transform.sql (L1815, L1832): transform(sequence(0, 15), i -> i * 10) and a named_struct variant, both expect_dispatch(transform) with no FROM, so 16 elements land in a one-row batch (JVM codegen dispatcher miscompiles map-typed (MapType) output #4539).
• array/dispatch_nested_input.sql (new, spark.comet.batchSize=16): L1863, L1884, L1934, L1947, L1960, L1976.
  • map('k', array_distinct(s)) over ARRAY<STRUCT<a INT, b STRING>>.
  • map('k', array_max(flatten(a))) over BINARY, STRING, DECIMAL(10,2) and DECIMAL(30,2) nested arrays with inner NULLs, with expect_dispatch(array_max, flatten).
  • L1884's 100 deterministic rows are generated in SQL as one file.
• array/dispatch_collection_output_sizes.sql (new, spark.comet.batchSize=8): L2324, 10 instances.
  • A 256-row single-file seed table, with one query per output type built by always-dispatch HOFs.

Strings, math, datetime, decimal, misc

• string/upper.sql (L2023): expect_dispatch(upper, substring) on upper(substring(s, 1, 2)).
• string/upper_ansi.sql (new, L2033): expect_error(DIVIDE_BY_ZERO) on IF(flag, upper(substring('abc', CAST(1L DIV 0L AS INT), n)), NULL), plus a sentinel of the same shape (Codegen dispatcher single-ordinal null short-circuit swallows ANSI errors from literal subtrees #5608).
• string/lpad_binary.sql + _dispatch_disabled.sql (new, L2357): lpad(b, 8, unhex('FF')) over binary.
  • Dispatcher on: expect_dispatch(staticinvoke).
  • Dispatcher off: expect_fallback(...) (Route unrecognized StaticInvoke and Invoke through the codegen dispatcher instead of falling back #5575).
• string/replace_dispatch_boundary.sql + _ansi.sql (new, L701)
  • Eight replace shapes as expect_dispatch(replace).
  • The ANSI null short-circuit shapes go in the ANSI file.
• string/is_valid_utf8.sql (new, MinSparkVersion: 4.0, L2379): expect_dispatch(invoke) on is_valid_utf8(s) and make_valid_utf8(s), including a malformed value.
• math/pmod.sql (L2066)
  • Add the (NULL, NULL) row, and make L26-27 expect_dispatch(pmod).
  • Add map('k', a + b) as expect_dispatch(add).
• math/pmod_ansi.sql (new, L2087)
  • A (NULL, 0) sentinel row with expect_dispatch(pmod).
  • (7, 0) as expect_error(BY_ZERO), or use a 4.0 / 4.1 file pair to keep DIVIDE_BY_ZERO versus REMAINDER_BY_ZERO exact.
• math/hypot.sql (L602): expect_dispatch(hypot, cast, checkoverflow, add) on hypot(CAST(amount + amount AS DOUBLE), 4.0D) over a DECIMAL(10,2) column.
• math/hypot_dispatch_disabled.sql (new, L796): expect_fallback(hypot: spark.comet.exec.scalaUDF.codegen.enabled=false).
• datetime/add_months_ansi.sql (new, L1992, Codegen dispatcher: whole-tree NullIntolerant short-circuit suppresses ANSI errors, plus TIME type gaps between canHandle and the runtime dispatcher #5218): expect_error(CAST_INVALID_INPUT) on add_months(CAST(s AS DATE), i), plus an expect_dispatch(add_months, cast) sentinel over valid rows.
• decimal/decimal_ops.sql (L576): expect_native(checkoverflow, add) on a + b.
• misc/dispatch_sliced_boolean.sql (new, L1066)
  • Set spark.sql.shuffle.partitions=1 and spark.comet.batchSize=100.
  • expect_dispatch(regexp_replace) on regexp_replace(IF(b, s, 'zz'), '1', 'y') over a grouped aggregate.

The conversions above assume the kernel accepts output types such as Map<String, Binary> and Map<String, Array<Struct>>. The UDF tests' map and struct outputs make that likely, but it hasn't been verified, so run the new fixtures before deleting the Scala tests.

Blocked or partly kept

• L80 and the to_json instance of L151: blocked on the Spark 4.x coverage-tag loss described above. Pinning the mechanism on 4.x needs the shim to copy coverage tags back.
• L701: the malformed CAST(X'FF' AS STRING) and 256 KiB repeat(...) literal cases need ConstantFolding (test: let Comet SQL tests keep ConstantFolding, exclude optimizer rules and change configs mid-file #6618), or the 262,144-character literal spelled out.
• L2379: the 3.4/3.5 half (no SQL function reaches CometInvoke there) and the Invoke-with-arguments case on a folded literal target stay in Scala.

Stays in Scala

• ScalaUDF registration and type surface: L978, L987, L1086-L1155, L1296 (10 instances), L1307-L1416, L1545-L1771.
• UDF-specific semantics: L302, L498, L1030, L1075, L1440, L1461.
• Kernel cache and Nondeterministic state: L818, L847, L864, L895, L921, L947, L1487, L1505.
• Dispatcher counters: L317, L333.
• Direct kernel and serde tests: L240, L2127 (TIME input, which has no SQL route), L2336, L2396.
• Memory lifecycle: L270.
• Explain, COMET-INFO and coverage-stat text: L357, L544, L626, L684.
• Test-helper self-tests: L401, L430.
• Dataset API and DSv2 catalog functions: L1040, L2441.

Already covered: L1802, by map/map_concat.sql L46-47.

Findings

• The Codegen dispatcher: whole-tree NullIntolerant short-circuit suppresses ANSI errors, plus TIME type gaps between canHandle and the runtime dispatcher #5218 short-circuit tests L2066 and L2087 never generate the short-circuit. They wrap pmod / a + b in idInt(...), and a ScalaUDF root is not NullIntolerant. A bare pmod(a, b) does take the path, so the fixture versions are stronger.
• L302's comment describes the wrong mechanism, and its claim that Concat(null, x) = x is wrong for Spark.
• L847 never runs more than one batch per partition.
• L864's premise doesn't hold: the table read forces columns nullable.
• Explain on Spark 4.x misreports dispatched json_array_length and to_json. It likely does the same for parse_url, which goes through the same shim code. A dispatched call shows up as a native staticinvoke / invoke, and the COMET-INFO dispatcher line omits it. That is a product issue, not just a fixture gap. It was found by reading the code; worth confirming and filing separately.
• Fixture headers claim a mechanism nothing asserts: sequence.sql, json_array_length_default.sql, hypot.sql, nanvl.sql, pmod.sql, map_concat.sql, create_map.sql, transform.sql.

Done when

• Every test in the table below is accounted for. Either a fixture covers it (the same inputs, configs and assertion strength, or stronger), the fixture that already covers it is cited, or it stays in Scala for the reason given above.
• Each converted Scala test is deleted in the PR that adds its coverage. A suite left empty is deleted and removed from the suite lists in .github/workflows/pr_build_linux.yml and pr_build_macos.yml.
• The new and changed fixtures pass on every Spark profile. Apply run-all-spark-profiles, because pull requests run only the default profile.