feat: default native replace for non-empty UTF8_BINARY literal search

[sam-1112]

Which issue does this PR close?

Closes #5354.

Rationale for this change

DataFusion replace only diverges from Spark when search is empty: Spark returns src unchanged, while DataFusion inserts replace between every character. For the common case—a non-empty UTF8_BINARY literal—the native kernel is Spark compatible, so Comet can use it by default instead of the JVM codegen dispatcher. Routing is decided by search only. The replacement argument is not required to be a literal, so replace(col, 'literal', replacementColumn) also runs natively. Empty literal search, non-literal search, and non-default collations remain on the dispatcher. spark.comet.expression.StringReplace.allowIncompatible is unchanged and still opts the remaining cases into native execution.
This PR does not change the native kernel. Collation support remains out of scope (#4496).

What changes are included in this PR?

• CometStringReplace uses native DataFusion replace by default when search is a non-empty UTF8_BINARY literal.
• Empty literal search, non-literal search, and non-default collations continue to use the JVM codegen dispatcher.
• Updated the ## replace section in docs/source/contributor-guide/expression-audits/string_funcs.md.
• Added SQL coverage for overlapping replacements and multi-byte UTF-8 on the native path.
• Added routing coverage for native versus dispatcher execution.

How are these changes tested?

SQL tests

File: spark/src/test/resources/sql-tests/expressions/string/string_replace.sql

Existing unchanged coverage includes:

• Empty search: replace('hello', '', 'x') and related NULL/empty variants
• Empty source: ('', 'a', 'b') in the table scan
• No match: ('hello', 'xyz', 'abc')
• Single match: ('hello world', 'world', 'there')
• Multiple matches: ('aaa', 'a', 'bb')

This PR adds column-source, literal-search cases that take the native path:

• Overlapping candidates: replace(s, 'aa', 'x') on 'aaaa' → 'xx'
• Multi-byte UTF-8 values, including '你好你好' and '😀a😀'

Routing tests

CometCodegenSuite asserts the following routing behavior:

• Non-empty literal search → native
• Empty literal search → JVM codegen dispatcher
• Column search → JVM codegen dispatcher
• Non-default collation on Spark 4+ → JVM codegen dispatcher

The non-default-collation assertion is gated by isSpark40Plus.

Commands executed

The SQL file and routing tests were run on Spark 4.1, 3.5, and 3.4:

• ./mvnw test -Dtest=none -Dsuites="org.apache.comet.CometSqlFileTestSuite string_replace"
• ./mvnw test -Dtest=none -Dsuites="org.apache.comet.CometCodegenSuite replace routes native"

Spark 4.1 was tested with JDK 17. Spark 3.5 and 3.4 were tested with JDK 11.

Performance trade-off

Native execution is not faster in this microbenchmark. It uses 2.5–2.9× as much CPU time as the dispatcher; at 100% match density, 474.6 / 166.2 ≈ 2.86×.

This PR therefore does not claim a throughput improvement. It trades CPU throughput for reduced dispatcher-side JVM heap churn on a scan-bound path. Reviewer feedback on whether that trade-off justifies changing the default is especially welcome.

Benchmark query:

SELECT replace(c1, '123', 'ab')
FROM parquetV1Table

Rows: 1,048,576

All timings are in ns/row; lower is better.

Match density	Spark (ns/row)	Comet dispatcher (ns/row)	Comet native (ns/row)	Native / dispatcher
0%	34.1	153.1	395.9	2.59×
10%	48.2	162.2	407.5	2.51×
100%	71.6	166.2	474.6	2.86×

The dispatcher is estimated to allocate approximately 80–136 bytes/row for this path, based on the JVM-heap UTF8String analysis in #5354. This allocation was not re-measured in this benchmark run, and Spark Benchmark's allocated-bytes column was not recorded.

The native path avoids those dispatcher-side per-row JVM string allocations.

Benchmark environment and methodology

• Spark 4.1.2, using Comet's default profile at the time of measurement
• Apple M4
• OpenJDK 17
• Release native library built with make release
• Spark Benchmark defaults:
  • Warmup: 2 seconds
  • minNumIters: 2
  • minTime: 2 seconds

After this change, a non-empty literal search always takes the native path. Measuring the dispatcher arm therefore required temporarily forcing nativeSafeSearchSubset to false in a local benchmark harness. That harness change is not included in this PR.

[sunchao]

Summary

Reviewed head 3b0d9d405b477218fa6c664919f020ec1ae2bc99 against base f8f9b3356d29e776e02a9fca7a6b3c630790db2c with five independent review scopes. The patch keeps the implementation small and reuses the existing native kernel, but three P2 regressions survive verification: malformed search literals can change results, NULL-source rows can raise previously suppressed replacement-expression errors, and large scalar literals can overflow during batch broadcasting. All three reproduce with spark.comet.expression.StringReplace.allowIncompatible=false.

Prior state and problem

Previously, the default StringReplace path serialized the complete Spark expression into the JVM codegen dispatcher. Native DataFusion execution required explicit incompatibility opt-in because empty-search behavior differs from Spark. Issue #5354 proposes recognizing the non-empty-literal subset at planning time to avoid dispatcher-side JVM allocations.

That motivation is understandable, and ordinary valid-UTF-8 replacement semantics agree. The compatibility boundary also includes literal serialization, conditional evaluation of children, and the scalar-to-array adaptation performed before the native kernel runs. Those boundaries explain the three findings even though this PR does not change the kernel itself.

Design approach

nativeSafeSearchSubset requires three children, a non-empty UTF8String search literal, and no non-default string collation on any child. Both getSupportLevel and convert use this predicate, so eligible expressions become native by default while empty, non-literal, and collated searches retain dispatcher routing.

The replacement remains unrestricted and may be a column or expression. Explicit allowIncompatible=true still selects the existing native route for the other cases. The collation check is conservative and uses the existing version shims rather than introducing Spark-version-specific logic here.

Correctness / compatibility analysis

The pinned DataFusion 54.1.0 UDF matched 9,000 seeded valid-UTF-8 cases against each of Spark 3.4.3, 3.5.8, 4.0.4, and 4.1.3. The corpus included NULL source/replacement, empty source/replacement, overlaps, multibyte and combining characters, NUL bytes, and replacement columns. A focused Spark 4.0.4 probe also checked parsing, routing, and interpreted/generated agreement for UTF8_BINARY, UTF8_LCASE, and UTF8_BINARY_RTRIM.

For the regressions, I separately compiled the exact base and head versions of strings.scala and ran controlled A/B SQL probes on the same cached Spark 4.0.4/Comet support runtime. Both sides executed Comet projections, with the base retaining the dispatcher and the head taking the native route. The malformed-literal wrong result, ANSI division-by-zero failure, and unused-large-literal overflow all appeared only with the head serde. Actual pinned DataFusion/Arrow probes independently corroborated literal replacement and broadcasting behavior. This is focused serde/kernel validation, not a fresh full-head Maven/native build or a full test-suite run.

The exact-head CI workflow, Delta build gate, CodeQL, and PR-title workflows currently report action_required. They provide no executed CI test evidence for this head. The test runs and performance numbers in the PR description remain author-reported.

Key design decisions

Reusing the existing kernel avoids a second replacement implementation, and retaining dispatcher behavior outside the intended subset limits the change. Leaving the incompatibility opt-in intact also preserves the existing configuration surface. However, search shape alone does not establish that the complete expression is compatible with native execution.

The disclosed performance trade-off is explicit: native takes 2.5–2.9 times the dispatcher timing in the supplied benchmark, while eliminating dispatcher-side JVM string allocations. The allocation estimate was not remeasured in that benchmark. I have not treated the disclosed slowdown itself as an additional finding or inferred a production throughput or total-memory improvement from those numbers.

Implementation sketch

The introduced PR diff changes four files: the serde predicate and routing, the replacement audit entry, SQL cases, and the codegen routing test. The existing scalar-function registration and three-string DataFusion signature already provide the required native plumbing, so no new registration is missing.

I also inspected all nine paths in the direct base-to-head tree diff. The other five paths belong to newer base-only commits for signed-zero tests and workflow dependencies, not reversions introduced by this PR. The added SQL tests run through a harness that excludes constant folding, while the Scala routing test uses a column source and dispatcher annotations to distinguish execution paths.

Behavioral changes worth calling out

For the newly native subset, Spark no longer receives the whole expression tree in a serialized dispatcher closure. String literals pass through native literal serialization, replacement expressions are evaluated eagerly as native children, and scalar strings are expanded for the batch before matching. The inline reproducers show observable changes at each of these boundaries.

These are default-behavior regressions rather than changes confined to users who opted into incompatibility. Ordinary replacement-column, overlap, and valid-Unicode controls continued to match, so the findings do not argue against native execution for a more carefully bounded subset.

Suggested improvements

Please preserve byte-level behavior for malformed literals and Spark's NULL short-circuiting for replacement expressions before selecting the native default. The native path also needs scalar-aware handling, or dispatcher retention for affected calls, so a no-match replacement cannot overflow merely from broadcasting an unused literal.

Add targeted regressions for a valid U+FFFD source with a malformed search literal, a NULL source with an ANSI-erroring replacement expression, and a full batch with a large unused search/replacement literal. These tests should assert both the result or expected error behavior and the intended execution path. The three inline comments contain concrete reproducers and the relevant boundaries to address.

[sunchao]

[P2] Exclude malformed search literals from the native-safe subset

Could this guard also reject search literals whose bytes change during native serialization? With a valid Parquet source column containing U+FFFD (EF BF BD), replace(s, CAST(X'FF' AS STRING), 'x') returns the source unchanged in Spark and the base dispatcher because byte FF is absent. Catalyst folds the cast to a non-empty UTF8String, so this check accepts it, but CometLiteral serializes it through UTF8String.toString, changing the search to U+FFFD. The head's native path consequently returns x. I reproduced this with allowIncompatible=false and entirely valid scan data. Please retain dispatcher routing for malformed search literals unless their byte semantics can be preserved natively.

[sunchao]

[P2] Preserve NULL short-circuiting for replacement expressions

With ANSI enabled and Parquet rows (s=NULL, n=0) and (s='a', n=1), SELECT replace(s, 'a', CAST(1 / n AS STRING)) FROM t succeeds in Spark and the base dispatcher, returning NULL and '1.0'. This native conversion instead raises DIVIDE_BY_ZERO with allowIncompatible=false. Spark's ternary expression skips the replacement when the source is NULL, whereas the native scalar-function expression evaluates every child for the batch before replace receives the source null mask. Could the native eligibility check account for this conditional evaluation, or retain dispatcher routing when the replacement can throw? A nullable-source/erroring-replacement regression would protect this behavior.

[sunchao]

[P2] Avoid broadcasting large literals in the new default path

For an 8,192-row Parquet batch containing only short strings 'a'/'b', SELECT replace(s, 'notfound', repeat('x', 262144)) FROM t should return those inputs unchanged. The base dispatcher does, but the head's native Comet projection fails with CometNativeException: native panic: offset overflow, even though the output is tiny. Pinned DataFusion 54.1.0 broadcasts every Utf8 scalar before matching, so the unused 256 KiB replacement becomes a 2 GiB array and exceeds Arrow's i32 offsets at the default Comet batch size. A large search literal triggers the same problem. Please use scalar-aware native execution or retain dispatcher routing for these cases instead of enabling them solely from a non-empty search.