feat: run length on binary input natively

[dwsmith1983]

Which issue does this PR close?

Closes #2348.

Rationale for this change

length(binary) currently falls back to Spark because Comet registered DataFusion's character_length, which accepts string types only. datafusion-spark now ships a Spark-compatible length that returns the byte count for binary and the character count for strings, so the fallback can go.

What changes are included in this PR?

• Register SparkLengthFunc in the native session under length, char_length, character_length and len, replacing character_length. The result is Int32 for every input width, matching Spark.
• Drop the BinaryType fallback from CometLength.
• A Rust roundtrip test through the Parquet schema adapter shows dictionary-typed string and binary columns are cast to Utf8/Binary before any expression sees them, so the replaced function's dictionary handling is not needed downstream. A Scala test reads a small Arrow-written Parquet file whose embedded schema declares dictionary-typed columns and checks length runs natively on it.
• Refresh the audit note in the string function audit doc and the stale notes in the bit_length/octet_length fixtures.

How are these changes tested?

• length.sql gains a binary table and queries covering multi-byte text, bytes that are not valid UTF-8, embedded NUL bytes, empty, NULL and all-NULL columns, binary reached through a struct field, an array element, a map value, unhex, substring and CAST(string AS BINARY), the Int32 result in arithmetic, a filter, a GROUP BY aggregate and after a native shuffle, and the char_length/character_length aliases. Every block asserts a fully native plan.
• CometStringExpressionSuite runs length on 1000 binary rows with and without dictionary encoding, and on the dictionary-typed file described above.
• A Rust test asserts length resolves to the Spark function returning Int32 for Utf8, LargeUtf8, Utf8View, Binary, LargeBinary and BinaryView with no cast inserted.
• Verified locally on Spark 3.5 and 4.0: CometSqlFileTestSuite length, CometStringExpressionSuite, cargo test, cargo clippy -D warnings, spotless.

[sunchao]

Correctness

Reviewed 35f4206b against 3810936b. Previously, length(binary) fell back to Spark because the registered DataFusion function accepted strings. This change registers SparkLengthFunc and removes that fallback. The locked DataFusion Spark 55.1 implementation counts bytes for binary and Unicode code points for strings, preserves nulls, and returns Int32. That matches the maintained Spark 3.5 and 4.0 implementations for the reviewed inputs, including empty values, multibyte text and binary values containing invalid UTF-8 or NUL bytes. Maintained Spark 3.4 and 4.1 sources were unavailable, so I am not claiming source coverage for those versions.

The registration is used by the actual scalar planner, including return-type inference. I also checked the dictionary concern: Spark-input scans and native shuffle scans already unpack dictionaries, and the native Parquet adapter converts columns to the required Spark types. The new Parquet roundtrip test exercises that adapter, while the registration test checks accepted types and the return field. The SQL fixture has 15 native-plan query blocks with constant folding disabled by the harness. The Scala cases add repeated binary data with dictionary encoding on and off. I found no verified correctness regression.

These are source and test-path checks. The author reports local Spark 3.5/4.0 and Rust tests, but I have not independently reproduced them. CI, CodeQL, the Delta gate and the title check are awaiting approval. Only the label job has succeeded. The fetched merge commit has the assigned base/head parents and the same tree as the reviewed head, but no product CI execution is credited.

Performance

The binary kernel reads each value's length without decoding or copying its payload, and allocates an Int32 result vector. The string path retains the previous ASCII fast path and character-count loop. This is a reasonable implementation, but source inspection does not establish the benefit of replacing Spark fallback.

The P2 inline request asks for representative binary measurements and a string comparison against the base. The existing benchmark covers only 1,024 string rows, and this PR provides no timings for the new binary path. No benchmark was run during this review.

Design

Reusing the upstream Spark-compatible function keeps both input types behind one established serde and registration path. Its aliases replace the existing registry entries, and later Comet registrations do not replace it again. The planner can infer the correct return type from this registration, so an additional custom kernel or return-type workaround is unnecessary. The fallback documentation is updated at its source.

Abstraction & complexity

The production change removes a type-specific support override and adds one registration. It introduces no wrapper or separate execution mechanism. The SQL fixtures cover expressions and composition, while the Scala fixture and Rust roundtrip test address dictionary-backed storage and schema adaptation. Those tests have distinct purposes and fit the existing test infrastructure.

[sunchao]

Performance

[P2] Could you add a representative length(binary) microbenchmark and report Spark-versus-Comet results before enabling this path? This registration also replaces the existing string implementation, so please include a base-versus-head string comparison. The current CometStringExpressionBenchmark only calls length(c1) on 1,024 string rows, and this PR contains no binary timings. The added 1,000-row Scala cases verify answers and native plans, but do not measure performance. Short and long binary values, nulls, and repeated versus varied values would establish the benefit of the new path and check the existing string path for regressions.

[andygrove]

The function you are replacing handled dictionaries and this one does not. CharacterLengthFunc has an explicit DataType::Dictionary arm that recurses into the values, while SparkLengthFunc falls through to exec_err!("Unsupported data type ..."). Comet also stops coercing the input when it makes that swap, because SparkLengthFunc uses Signature::uniform and needs_fields_coercion in planner.rs returns false for TypeSignature::Uniform, so create_scalar_function_expr keeps the raw input type and inserts no cast. Anything outside those six types becomes a hard native error rather than a clean fallback to Spark.

I think you are right that no dictionary reaches an expression today. ScanExec and the shuffle scan both unpack, and the Parquet adapter casts to the required type. But that invariant is load bearing for length now and nothing in the code says so. Could you add a short comment at the registration recording what it depends on? The new test does not cover it either, because it asserts through fields_with_udf, and that is exactly the call the planner skips for a Uniform signature. Would it be better to assert through create_scalar_function_expr so the test follows the path the planner actually takes?

There is also an open PR covering the same ground. #5607 mixes CodegenDispatchFallback into CometLength, CometBitLength and CometOctetLength for binary and edits the same three fixtures, and it closes #5584, which overlaps #2348. Both cannot land as they are. I think this PR is the better outcome for length, since a native kernel beats routing back through the JVM dispatcher, and #5607's own numbers put the dispatcher at 1522 ms against Spark's 754 ms for 1 KB payloads. Could you and @adibmbrk agree on which PR owns length and have the other drop that part? #5607 also adds CometBinaryLengthBenchmark, which is the harness @sunchao is asking for above, so borrowing it would be cheaper than writing a third one.

On the same theme, octet_length on binary is the kernel you just registered and bit_length on binary is that times eight, but both still gate BinaryType as Unsupported here. Is there a reason to leave those to #5607 rather than finishing the family? The branch also conflicts with main in native/core/src/execution/jni_api.rs, so it needs a rebase before CI can tell us anything.

[dwsmith1983]

Could you add a short comment at the registration recording what it depends on? ... Would it be better to assert through create_scalar_function_expr so the test follows the path the planner actually takes?

Both done. The registration comment records that SparkLengthFunc unwraps no dictionaries and gets no planner coercion, so it relies on ScanExec, the shuffle scan and the Parquet adapter delivering plain Utf8 or Binary. The test now builds the ScalarFunc proto over a bound column and plans it through PhysicalPlanner::create_expr, asserting an Int32 scalar function over the column with no cast for both string and binary; the wider and view encodings keep a return-type check only, since no Comet input path produces them. The branch is merged with main and the jni_api.rs conflict resolved.

Could you and @adibmbrk agree on which PR owns length and have the other drop that part?

I agree this PR should own length, for the reason you give: a native kernel over routing through the dispatcher. @adibmbrk, would you drop the length part and the length.sql edits from #5607 and keep bit_length and octet_length there? I am happy to borrow CometBinaryLengthBenchmark for the numbers @sunchao asked for rather than add a third harness.

[sunchao]

Re-reviewed 2e344c494d9f against 1d0ce5fe8dfa, including the changes since 35f4206bbad8.

The registration now documents its dependency on dictionary unpacking. The revised test uses PhysicalPlanner::create_expr and checks the Int32 result and uncast column for string and binary inputs. The length implementation, SQL/Scala fixtures and dictionary roundtrip test are unchanged. I found no new correctness issue against maintained Spark 3.5/4.0 semantics.

The existing P2 performance request remains: there are still no representative Spark-versus-Comet binary timings or base-versus-head string measurements. Offering to reuse the benchmark harness does not yet supply those results.

Local validation was source and test-path analysis, with no native/JNI execution or benchmark run. Current-head CI, the Delta gate and CodeQL await approval with zero jobs. Only labeling passed. Maintained-source qualification remains limited to Spark 3.5/4.0.

[dwsmith1983]

Could you add a representative length(binary) microbenchmark and report Spark-versus-Comet results before enabling this path? ... please include a base-versus-head string comparison.

Added CometBinaryLengthBenchmark: 1M rows, five shapes, best time over the harness's iterations, Spark 3.5 on an M-series laptop. Base is main at 1d0ce5f with its own native build, head is this branch. On base the binary cases fall back to Spark, so the base Comet column there is the fallback path including the columnar-to-row conversion.

case	Spark	Comet base	Comet head	head vs Spark
length(string), 10-char varied	600 ms	73 ms	71 ms	8.4x
length(binary), 8 B varied	45 ms	43 ms	36 ms	1.2x
length(binary), 1 KiB varied	530 ms	713 ms	502 ms	1.1x
length(binary), 8 B, 20% null	43 ms	41 ms	32 ms	1.3x
length(binary), 5 repeated values	28 ms	20 ms	17 ms	1.7x

The string path is unchanged within noise, 73 ms base against 71 ms head. The binary path goes from a fallback that is slower than Spark on wide payloads (713 ms against 530 ms for 1 KiB values, where the conversion dominates) to native and ahead of Spark on every shape; the 1 KiB case is scan-bound on both sides, which is why the ratio is small there. The benchmark file is on the branch so the numbers can be reproduced with make benchmark-org.apache.spark.sql.benchmark.CometBinaryLengthBenchmark.