feat: enable native replace for safe inputs

docs/source/contributor-guide/expression-audits/string_funcs.md

@@ -169,8 +169,34 @@
 ## replace
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `StringReplace(src, search, replace)`; when `search` is empty, Spark returns `src` unchanged (short-circuit on `search.numBytes == 0`). DataFusion `replace` instead inserts `replace` between every character, so `CometStringReplace` reports `Compatible` with a `NativeOptIn` and runs Spark's own generated code inside the Comet pipeline by default. The native DataFusion `replace` is used only when `spark.comet.expression.StringReplace.allowIncompatible=true`.
+- Spark 3.5.8 (audited 2026-05-27): baseline. `StringReplace(src, search, replace)`; when `search` is empty, Spark returns `src` unchanged (short-circuit on `search.numBytes == 0`). Earlier DataFusion versions inserted `replace` between every character, so Comet routed the expression through Spark's generated code by default.
 - Spark 4.0.1 (audited 2026-05-27): routes through `CollationSupport.StringReplace.exec`; semantics unchanged for `UTF8_BINARY`. Non-default collations not honoured by Comet ([#4496](https://github.com/apache/datafusion-comet/issues/4496)).
+- DataFusion 55.1.0 (implementation reviewed 2026-10-05, [#5354](https://github.com/apache/datafusion-comet/issues/5354)): an empty search now returns the source unchanged. A direct source column and non-null literal search/replacement use its scalar fast path, which does not broadcast the literals into Arrow `Utf8` arrays. With a replacement column, its general path broadcasts the search literal into an Arrow array, so Comet limits that search to 256 bytes and the configured batch's broadcast to 8 MiB. The 256-byte limit also avoids a measured regression for a 1024-byte search literal. The literal-replacement path allows valid UTF-8 literals of at most 64 KiB. Both paths require a non-empty search and `UTF8_BINARY` operands. Large literals retain the JVM codegen dispatcher, including the 256 KiB regression cases from [#5409](https://github.com/apache/datafusion-comet/pull/5409). All other shapes also retain the dispatcher by default. The explicit `spark.comet.expression.StringReplace.allowIncompatible=true` opt-in remains available. The table below measures literal replacement only.
+- Benchmark (2026-10-05): `CometStringExpressionBenchmark` on Apple M4, macOS 26.7.1, JDK 21.0.12, Spark 4.1, release build, 65,536 rows, one run per route. The same query and Parquet inputs were used with only the default `replace` routing changed. These end-to-end figures include scan and projection; 10% match density is approximate.
+
+  | Match density | Dispatcher ns/row | Native ns/row | Dispatcher M rows/s | Native M rows/s | Native speedup |
+  | ------------- | ----------------: | ------------: | ------------------: | --------------: | -------------: |
+  | 0%            |             207.5 |         149.5 |                 4.8 |             6.7 |          1.39× |
+  | ~10%          |             277.5 |         141.8 |                 3.6 |             7.1 |          1.96× |
+  | 100%          |             309.9 |         170.5 |                 3.2 |             5.9 |          1.82× |
+
+- Column replacement benchmark rerun (2026-10-05): same Apple M4, Spark 4.1 release build and 65,536-row setup, one run. Spark versus native Comet measured 263.5 versus 149.9 ns/row (3.8 versus 6.7 M rows/s) at 0% matches, 225.2 versus 136.9 (4.4 versus 7.3 M rows/s) at about 10%, and 286.5 versus 164.9 (3.5 versus 6.1 M rows/s) at 100%. End-to-end scan and projection times.
+- Column replacement dispatcher baseline (2026-10-05): same machine and setup, with only the default route temporarily forced through the JVM codegen dispatcher. Dispatcher versus native measured 184.4 versus 149.9 ns/row (5.4 versus 6.7 M rows/s, 1.23× native speedup) at 0% matches, 243.3 versus 136.9 (4.1 versus 7.3 M rows/s, 1.78×) at about 10%, and 281.5 versus 164.9 (3.6 versus 6.1 M rows/s, 1.71×) at 100%. The dispatcher and native runs were separate single runs; benchmark timing includes scan and projection.
+- Long search literal benchmark (2026-10-05): Apple M4, macOS 26.7.1, JDK 21.0.12, Spark 4.1 release build, 65,536 rows, one run per route and case. Source strings were 2,048 bytes; search literals were 16, 256, or 1,024 ASCII bytes with 0%, about 10%, or 100% match density. Each run includes scan and projection. Dispatcher / native ns per row and native speedup over dispatcher:
+
+  | Search bytes | Match density | Dispatcher ns/row | Native ns/row | Dispatcher M rows/s | Native M rows/s | Native speedup |
+  | -----------: | ------------- | ----------------: | ------------: | ------------------: | --------------: | -------------: |
+  |           16 | 0%            |           1,734.6 |         723.6 |                 0.6 |             1.4 |          2.40× |
+  |           16 | ~10%          |           3,071.5 |         715.8 |                 0.3 |             1.4 |          4.29× |
+  |           16 | 100%          |           2,667.7 |         715.1 |                 0.4 |             1.4 |          3.73× |
+  |          256 | 0%            |           2,823.3 |       1,270.9 |                 0.4 |             0.8 |          2.22× |
+  |          256 | ~10%          |           2,802.2 |       1,280.0 |                 0.4 |             0.8 |          2.19× |
+  |          256 | 100%          |           2,234.7 |       1,464.3 |                 0.4 |             0.7 |          1.53× |
+  |        1,024 | 0%            |           1,847.0 |       3,239.8 |                 0.5 |             0.3 |          0.57× |
+  |        1,024 | ~10%          |           1,808.6 |       3,287.6 |                 0.6 |             0.3 |          0.55× |
+  |        1,024 | 100%          |           1,195.0 |       4,004.1 |                 0.8 |             0.2 |          0.30× |
+
+  These results led to the 256-byte default limit for replacement-column routing. Runs are independent single benchmark invocations, so treat small differences as directional rather than statistically conclusive.
 
 ## right
 

docs/source/user-guide/latest/expressions.md

@@ -604,7 +604,7 @@ to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enab
 | `regexp_replace` | ✅ | Hybrid |  |
 | `regexp_substr` | ✅ | — | Rewrites to `nullif(regexp_extract(...), '')`; `regexp_extract` routes through the JVM codegen dispatcher by default |
 | `repeat` | ✅ | Native |  |
-| `replace` | ✅ | Hybrid |  |
+| `replace` | ✅ | Hybrid | Native by default for a direct string column with a non-empty, valid UTF-8 search literal and either a replacement literal (both at most 64 KiB) or a replacement column (search at most 256 bytes and at most 8 MiB when broadcast per batch), under UTF8_BINARY collation; other cases use the JVM codegen dispatcher by default |
 | `right` | ✅ | Native |  |
 | `rpad` | ✅ | Hybrid | String inputs use the native kernel with a column string and literal padding; literal strings and column padding use codegen dispatch. Binary inputs use codegen dispatch. |
 | `rtrim` | ✅ | Hybrid | Non-UTF8_BINARY collated input with a trim string routes through the JVM codegen dispatcher; other input runs natively |

spark/src/main/scala/org/apache/comet/serde/strings.scala

@@ -19,7 +19,7 @@
 
 package org.apache.comet.serde
 
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Base64, BitLength, Cast, Concat, ConcatWs, Contains, Elt, Empty2Null, EndsWith, Expression, FindInSet, FormatNumber, FormatString, GetJsonObject, InitCap, Left, Length, Levenshtein, Like, Literal, Lower, Mask, OctetLength, Overlay, RegExpExtract, RegExpExtractAll, RegExpInStr, RegExpReplace, Right, RLike, SoundEx, StartsWith, StringInstr, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, StringTrim, StringTrimLeft, StringTrimRight, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Base64, BitLength, BoundReference, Cast, Concat, ConcatWs, Contains, Elt, Empty2Null, EndsWith, Expression, FindInSet, FormatNumber, FormatString, GetJsonObject, InitCap, Left, Length, Levenshtein, Like, Literal, Lower, Mask, OctetLength, Overlay, RegExpExtract, RegExpExtractAll, RegExpInStr, RegExpReplace, Right, RLike, SoundEx, StartsWith, StringInstr, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, StringTrim, StringTrimLeft, StringTrimRight, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper}
 import org.apache.spark.sql.types.{ArrayType, BinaryType, DataTypes, IntegerType, LongType, StringType}
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -186,13 +186,58 @@ object CometInitCap extends CometScalarFunction[InitCap]("initcap") with NativeO
 
 object CometStringReplace
     extends CometScalarFunction[StringReplace]("replace")
-    with NativeOptInAvailable {
+    with NativeOptInAvailable
+    with CometTypeShim {
+
+  private val MaxNativeLiteralBytes = 64 * 1024
+  // With a replacement column, DataFusion broadcasts the scalar search into an Arrow array.
+  // Bound both the literal and its total expansion at the configured batch size.
+  private val MaxBroadcastSearchBytes = 256
+  private val MaxBroadcastSearchBytesPerBatch = 8 * 1024 * 1024
+
+  override def hasConditionalNativeDefault: Boolean = true
+
+  override def getCompatibleNotes(): Seq[String] = Seq(
+    "Native by default for a direct string column with a non-empty, valid UTF-8 search " +
+      "literal and either a replacement literal (both at most 64 KiB) or a replacement " +
+      "column (search at most 256 bytes and at most 8 MiB when broadcast per batch), under " +
+      "UTF8_BINARY collation. Other cases use " +
+      "the JVM codegen dispatcher by default.")
+
+  // DataFusion 55.1.0 keeps two non-null literal arguments as scalars when the source is an
+  // array. With a replacement column it broadcasts the search literal, hence the smaller bound.
+  // Direct column operands avoid Spark NULL short-circuit differences in nested expressions.
+  private def nativeSafeSubset(expr: StringReplace): Boolean = expr.children match {
+    case Seq(source, Literal(search: UTF8String, _), Literal(replacement: UTF8String, _)) =>
+      (source.isInstanceOf[Attribute] || source.isInstanceOf[BoundReference]) &&
+      search.numBytes() > 0 && search.numBytes() <= MaxNativeLiteralBytes &&
+      replacement.numBytes() <= MaxNativeLiteralBytes &&
+      isValidUtf8(search) && isValidUtf8(replacement) &&
+      !hasNonDefaultStringCollation(expr.dataType) &&
+      !expr.children.exists(child => hasNonDefaultStringCollation(child.dataType))
+    case Seq(source, Literal(search: UTF8String, _), replacement) =>
+      (source.isInstanceOf[Attribute] || source.isInstanceOf[BoundReference]) &&
+      (replacement.isInstanceOf[Attribute] || replacement.isInstanceOf[BoundReference]) &&
+      search.numBytes() > 0 && search.numBytes() <= MaxBroadcastSearchBytes &&
+      search.numBytes().toLong * CometConf.COMET_BATCH_SIZE.get() <=
+        MaxBroadcastSearchBytesPerBatch &&
+        isValidUtf8(search) &&
+        !hasNonDefaultStringCollation(expr.dataType) &&
+        !expr.children.exists(child => hasNonDefaultStringCollation(child.dataType))
+    case _ => false
+  }
+
+  private def useNative(expr: StringReplace): Boolean =
+    CometConf.isExprAllowIncompat(getExprConfigName(expr)) || nativeSafeSubset(expr)
 
   override def getIncompatibleReasons(): Seq[String] =
-    Seq("Produces different results from Spark when the search string is empty")
+    Seq(
+      "Unrestricted native arguments can change malformed UTF-8 results or Spark's NULL " +
+        "short-circuit behavior, ignore non-default collations, or overflow Arrow offsets when " +
+        "scalars are broadcast")
 
   override def getSupportLevel(expr: StringReplace): SupportLevel =
-    if (!CometConf.isExprAllowIncompat(getExprConfigName(expr))) {
+    if (!useNative(expr)) {
       Compatible(nativeOptIn =
         Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr)))))
     } else {
@@ -203,10 +248,7 @@ object CometStringReplace
       expr: StringReplace,
       inputs: Seq[Attribute],
       binding: Boolean): Option[Expr] = {
-    if (CometConf.isExprAllowIncompat(getExprConfigName(expr))) {
-      // The native DataFusion `replace` avoids the JVM allocations of the codegen
-      // dispatcher but is not Spark-compatible for an empty search string, so it is
-      // only used when incompatibility is explicitly allowed.
+    if (useNative(expr)) {
       super.convert(expr, inputs, binding)
     } else {
       // Run Spark's own generated code inside the Comet pipeline so the result matches Spark

spark/src/test/resources/sql-tests/expressions/string/routing_string_opt_in_disabled.sql

@@ -29,8 +29,13 @@ INSERT INTO routing_string_opt_in VALUES ('hello world', '{"a":1}'), ('', '{}'),
 query expect_fallback(initcap: spark.comet.exec.scalaUDF.codegen.enabled=false)
 SELECT initcap(s) FROM routing_string_opt_in
 
-query expect_fallback(replace: spark.comet.exec.scalaUDF.codegen.enabled=false)
+-- A direct column with bounded, non-empty literals is native even without opt-in.
+query expect_native(replace)
 SELECT replace(s, 'l', 'L') FROM routing_string_opt_in
 
+-- Empty search remains outside the native default subset and needs the dispatcher.
+query expect_fallback(replace: spark.comet.exec.scalaUDF.codegen.enabled=false)
+SELECT replace(s, '', 'L') FROM routing_string_opt_in
+
 query expect_fallback(get_json_object: spark.comet.exec.scalaUDF.codegen.enabled=false)
 SELECT get_json_object(j, '$.a') FROM routing_string_opt_in

spark/src/test/resources/sql-tests/expressions/string/routing_string_opt_in_enabled.sql

@@ -29,8 +29,13 @@ INSERT INTO routing_string_opt_in VALUES ('hello world', '{"a":1}'), ('', '{}'),
 query expect_dispatch(initcap)
 SELECT initcap(s) FROM routing_string_opt_in
 
-query expect_dispatch(replace)
+-- A direct column with bounded, non-empty literals is native even without opt-in.
+query expect_native(replace)
 SELECT replace(s, 'l', 'L') FROM routing_string_opt_in
 
+-- Empty search remains outside the native default subset and needs the dispatcher.
+query expect_dispatch(replace)
+SELECT replace(s, '', 'L') FROM routing_string_opt_in
+
 query expect_dispatch(get_json_object)
 SELECT get_json_object(j, '$.a') FROM routing_string_opt_in

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

@@ -24,10 +24,7 @@ INSERT INTO test_str_replace VALUES ('hello world', 'world', 'there'), ('aaa', '
 query
 SELECT replace(s, search, replace) FROM test_str_replace
 
--- Empty literal search: DataFusion's replace diverges from Spark
--- (Spark short-circuits and returns the source unchanged). The custom
--- CometStringReplace serde routes through the codegen dispatcher so
--- Spark's own doGenCode handles this case.
+-- Empty literal search: CometStringReplace retains the codegen dispatcher path.
 -- https://github.com/apache/datafusion-comet/issues/4497
 query
 SELECT replace('hello', '', 'x')