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.