fix: route length/bit_length/octet_length binary input through codegen dispatcher

[adibmbrk]

Which issue does this PR close?

Closes #5584.

Rationale for this change

length, bit_length, and octet_length rejected BinaryType input and fell the entire projection back to Spark, even though the operation is trivial (numBytes()) and BinaryType is already supported by the codegen dispatcher.

What changes are included in this PR?

• Mix CodegenDispatchFallback into CometLength, CometBitLength, and CometOctetLength so binary input routes through the JVM codegen dispatcher (Spark's own doGenCode, run inside the Comet pipeline) instead of falling back to Spark.
• Regenerate the affected docs rows (length, len, char_length, character_length, bit_length, octet_length) from Native to Hybrid.
• Add CometBinaryLengthBenchmark (see below).

How are these changes tested?

Updated the length.sql, bit_length.sql, and octet_length.sql fixtures: the binary cases now assert native-vs-Spark parity (checkSparkAnswerAndOperator) instead of expect_fallback. Added binary coverage to length.sql.

Ran against Spark 4.1:

• CometStringExpressionSuite — 33 succeeded, 0 failed.
• CometSqlFileTestSuite — 467 succeeded, 0 failed.

Benchmark

Added CometBinaryLengthBenchmark, per review request. CometStringExpressionBenchmark covers these three roots only on StringType, which takes the native DataFusion kernel and so never exercises the binary route.

Three arms, all cases of one Benchmark so warmup, iteration count, data and SQL settings are matched by construction:

Arm	Configuration
Spark	spark.comet.enabled=false
Comet (Spark fallback)	Comet on, spark.comet.exec.scalaUDF.codegen.enabled=false — the pre-PR path
Comet (codegen dispatch)	Comet on, dispatcher enabled — the path this PR adds

Shapes span payload width (8 B / 64 B / 1 KB) and null fraction (0 / 50 / 90%). Each case prints its physical plan and a result digest before the timings; all three arms agreed in all 15 cases.

length, best time over 2M rows, Apple M4 / OpenJDK 21.0.8 / Spark 4.1 (bit_length and octet_length track it closely):

Shape	Spark	Comet (Spark fallback)	Comet (codegen dispatch)
8 B	169 ms	101 ms	135 ms
64 B	201 ms	164 ms	206 ms
1 KB	754 ms	1193 ms	1522 ms
64 B, 50% null	167 ms	104 ms	150 ms
64 B, 90% null	92 ms	68 ms	80 ms

Known follow-up

Review surfaced a pre-existing correctness hole in CometBatchKernelCodegen.canShortCircuitNulls: its single-ordinal null short-circuit can swallow an ANSI error raised by a foldable-but-throwing subtree between the dispatched root and its single input ordinal (e.g. length(substring(X'00', CAST(1L DIV 0L AS INT), n))). This is not introduced by this PR — upper on main reproduces the same issue — and does not block this PR. Tracked and fixed separately in #5608.

[sunchao]

[P2] Preserve child evaluation order for compound binary inputs

This marker dispatches the entire binary-producing child, so the existing kernel null shortcut can now suppress an earlier ANSI error. An unexecuted source-derived diagnostic is IF(flag, length(substring(X'00', CAST(1L DIV 0L AS INT), n)), 0) over persisted Parquet rows (true, NULL) and (false, NULL), with flag BOOLEAN, nullable n INT, and ANSI/Comet projection/codegen dispatch enabled. In the inspected Spark 3.5/4.0 source, the conditional keeps the failing constant inside a branch. Spark evaluates Substring's position before its later length argument, so the selected branch must raise division by zero even when n is null.

Here CometScalaUDF captures the Length tree with only n bound. Its nodes pass allNullIntolerant and the single-input-ordinal guard in CometBatchKernelCodegen, which writes NULL before evaluating the generated child code. At BASE the unsupported binary Length has no dispatcher marker and the enclosing projection falls back to Spark. The new BitLength and OctetLength markers expose the same issue. Please preserve Spark's evaluation order, or retain fallback for these unsafe compound trees, and add a regression asserting the ANSI error for all three roots. This is a source trace, not an executed reproduction.

[andygrove]

Thanks @sunchao, I ran this down and the trace holds. Reproduced on this branch (Spark 4.1, ANSI on):

CREATE TABLE t (flag BOOLEAN, n INT) USING parquet;
INSERT INTO t VALUES (true, NULL), (false, NULL);
SELECT IF(flag, length(substring(X'00', CAST(1L DIV 0L AS INT), n)), 0) FROM t;

Spark raises [DIVIDE_BY_ZERO], Comet returns a row. Same for bit_length and octet_length.

For anyone reading later, the three pieces that have to line up:

• ConstantFolding refuses to fold 1L DIV 0L because it sits under an If branch (it tags FAILED_TO_EVALUATE and leaves the node alone), so the throwing literal survives into the physical plan.
• TernaryExpression.nullSafeCodeGen emits Substring's pos code before it tests len's null, so Spark evaluates the division even though n is NULL.
• Length, Substring, Cast and IntegralDivide are all null-intolerant and the dispatched tree reads exactly one ordinal, so canShortCircuitNulls takes its single-ordinal branch and the kernel writes NULL before ev.code runs.

One correction on scope: this isn't introduced here, it's the residual hole in #5219. The single-ordinal branch assumes "there is nothing left for Spark to evaluate ahead of that ordinal's own null check", and that's false whenever the tree carries a literal-only subtree that throws. upper reproduces it on main today, unchanged by this PR:

SELECT IF(flag, upper(substring('abc', CAST(1L DIV 0L AS INT), n)), NULL) FROM t;

I confirmed that one on the same build: Spark raises, Comet doesn't.

So I'd rather fix canShortCircuitNulls than special-case the three length serdes, otherwise we paper over three of the ~70 expressions that share the hole. Filed as #5608, with the suggested guard and a regression test covering upper plus all three roots from this PR.

@adibmbrk I don't think this needs to block the PR. Please add a link to #5608 in the PR description so the connection isn't lost.

[sunchao]

Re-reviewed 37f12bed after the reproduction discussion. [P2] Thanks for reproducing the evaluation-order issue and tracking the shared fix in #5608. I agree the shared dispatcher is the right place to address it, with these three newly exposed binary callers covered alongside Upper. I found no additional P1/P2 issues.

Could you share one focused binary-input microbenchmark comparing this dispatcher path with the previous Spark-fallback path? Please use matched Spark/Comet build settings, data and warmup, with representative payload widths and null fractions, and include the actual execution plans and matching results. The existing string-expression benchmark does not exercise this binary route.

[sunchao]

Thanks for documenting the separate fix. I verified that #5608 was closed by merged #5623, preserving the agreed separate-fix scope for the existing P2 evaluation-order issue. That guard is absent from this exact head, and I have not validated the combined changes.

The updated description also says the binary-input benchmark was added. At 37f12bed, CometStringExpressionBenchmark is unchanged from the base and still builds string c1 and integer c2. Could you push or link the benchmark commit and its results for the stated payload widths and null fractions, including dispatcher-on/off plans and matching outputs, so the earlier request can be closed?

No new P1/P2 findings. This pass was source and discussion verification, not a benchmark or Spark/JNI run. Current-head checks are 63 successful and 9 skipped. A separate title-check workflow remains action_required.

[adibmbrk]

Hi @sunchao @andygrove, I have committed the microbenchmark and the results from it are in the PR description.

[sunchao]

Thanks for adding the binary benchmark. [P2] Could we retain Spark fallback for these three binary roots until a narrow implementation improves the comparison, or provide evidence that justifies the tradeoff? Your reported length times increase from 164 to 206 ms at 64 B (+25.6%) and from 1193 to 1522 ms at 1 KB (+27.6%). With Comet execution and the default dispatcher setting enabled, the new markers select that route for ordinary binary-column projections. I inspected the source but have not independently reproduced these timings.

A native binary-length implementation or a direct-vector length path that reads offsets seems worth measuring, while preserving null handling, indexing and compound-child evaluation. The generated dispatcher getter allocates and copies the payload, but CometPlainVector.getBinary in the fallback also does so. These numbers do not isolate copying as the cause of the extra time. A length-specific optimization should leave the general getBinary contents contract intact.

For the comparison, could you capture the actual timed write-command plans and matching results on your Spark 4.1 build? In the maintained Spark 3.5/4.0 sources, noop() creates a separate command QueryExecution. The SELECT plan printed by describe therefore does not establish the timed writer operators. I have not independently verified that boundary on your Spark 4.1 build.

The previously deferred #5608 guard is now in the reported base and inspected merge source, but not raw HEAD. Its accepted separate-fix scope is unchanged. This pass was source review, not a Spark/JNI or benchmark run. All four current-head workflows are awaiting approval (action_required), with no check results yet.

[sunchao]

Reviewed head b517646 against base 90c1dd9 with five independent scopes. The change adds little code, but the default execution choice is not justified by the supplied performance results. I also found a new build failure.

1. [P1] The benchmark breaks Spark 3.4/3.5 test compilation.
   describe returns String, but those Spark versions’ withSQLConf returns Unit. Current CI reports exactly found: Unit, required: String. I independently reproduced this with an extracted compilation. Capturing the result inside the configuration block and returning it afterward compiles with both API signatures. CI failure

2. [P2] The new default route is slower in every reported benchmark shape.
   The author’s length results show 164 → 206 ms at 64 B (+25.6%) and 1193 → 1522 ms at 1 KB (+27.6%) versus the previous Comet/Spark fallback. These are author-reported measurements; I did not independently reproduce the timings. They argue for retaining fallback until a faster implementation or representative pipeline measurements justify changing it. Reported results

   The dispatcher adds a JVM callback and Arrow bridging per expression/batch. Its binary getter also allocates and copies the payload merely to read its length. The old fallback copies binary values too, so copying alone does not establish the cause of the measured regression.

3. The benchmark should capture the actual timed execution plan.
   describe invokes df.noop() and then prints the original SELECT’s plan. I verified that Spark 4.1.3’s noop() executes a separate write command with its own QueryExecution. Capture that command through a query listener and assert the intended operators. Also measure native downstream consumers and a projection computing all three lengths from the same input. Those cases would test the claimed benefit of preserving the Comet pipeline.

For design and complexity, I favor a small native binary-length implementation. Reusing CodegenDispatchFallback is consistent with existing code and introduces no unnecessary framework. The execution machinery is nevertheless expensive for an operation that only needs adjacent Arrow offsets.

Arrow 58.4 already provides binary length kernels. A narrow adapter could share byte-length handling between length and octet_length, preserve Spark’s wrapping multiplication for bit_length, and avoid payload materialization and JVM bridging. Three local component tests passed for nulls, empty/arbitrary bytes, sliced arrays, and wide values. This validates the underlying kernels, not a completed Comet implementation.

The performance and plan-reporting concerns were already raised in the discussion; the compilation failure is the new finding. At the last CI check: 4 failed, 13 passed, 6 running, 6 skipped. No full Comet build or end-to-end benchmark was run locally.