chore(audit): address PR review feedback on string-expressions audit

Apply the recurring patterns surfaced in PR review comments rather than
documenting them as Known Limitation prose.

- Remove the unreachable `CometScalarFunction("btrim")` serde mapping.
  `StringTrimBoth` is `RuntimeReplaceable` and rewritten to `StringTrim`
  before serde runs, so the entry was dead code. Updated the support-doc
  sub-bullet for `btrim` to point readers at the `trim` entry.
- Rewrite `CometCaseConversionBase` to use the standard `Incompatible`
  opt-in instead of the bespoke `spark.comet.caseConversion.enabled`
  conf. Override `getSupportLevel` to return
  `Incompatible(Some(reason))` and drop the in-`convert` config check.
  Remove `COMET_CASE_CONVERSION_ENABLED` and update the two callers and
  the string-expression benchmark to use the per-expression
  `spark.comet.expression.{Upper,Lower,InitCap}.allowIncompatible`
  keys. Closes the gating asymmetry called out in #4467.
- Mark `replace` as `Incompatible` when `search` is a literal empty
  string. DataFusion's `replace` inserts the replacement between every
  character whereas Spark short-circuits to return `src` unchanged.
  Refile under #4497 (#3344 was closed previously with the bug body
  describing the divergence in the wrong direction).
- Drop the spurious `CometStringRepeat.getCompatibleNotes()` claim that
  DataFusion `repeat` throws on negative counts. Verified in
  `expressions/string/string_repeat.sql` that negative counts produce
  the empty string Spark expects. The audit's prose claim was wrong, so
  #4462 was closed as already-fixed.
- Update the `replace` SQL fixture to assert the new fallback message
  rather than the prior `ignore(...)` shape. Update `lower.sql` /
  `upper.sql` `expect_fallback(...)` strings to match the new reason
  text. Switch `lower_enabled.sql` / `upper_enabled.sql` from
  `spark.comet.caseConversion.enabled` to the per-expression
  `allowIncompatible` keys.
- Cross-reference the collation umbrella issue
  (#4496) from every support-doc sub-bullet that previously documented
  "non-default collations may diverge silently" without an issue link.
  Add the cross-reference to the entries that route through Spark 4.0
  `CollationSupport.X.exec` but had no collation note at all
  (`bit_length`, `concat_ws`, `endswith`, `initcap`, `instr`, `lpad`,
  `octet_length`, `repeat`, `replace`, `right`, `rpad`, `rtrim`,
  `split`, `startswith`, `substring`, `substring_index`, `translate`).
- Replace the stale "Known limitation: enforced inside `convert`" prose
  on `left`, `right`, `substring`, and `length`. The literal-only
  argument restriction has already been moved into `getSupportLevel`;
  the prose just hadn't been updated.

Tests:
- `./mvnw test -Dsuites="org.apache.comet.CometSqlFileTestSuite expressions/string/"` (42 pass)
- `./mvnw test -Dsuites="org.apache.comet.CometStringExpressionSuite"` (33 pass)

Author: andygrove

docs/source/contributor-guide/spark_expressions_support.md

@@ -745,14 +745,6 @@ object CometConf extends ShimCometConf {
       .toSequence
       .createWithDefault(Seq("Range,InMemoryTableScan,RDDScan,OneRowRelation"))
 
-  val COMET_CASE_CONVERSION_ENABLED: ConfigEntry[Boolean] =
-    conf("spark.comet.caseConversion.enabled")
-      .category(CATEGORY_EXEC)
-      .doc("Java uses locale-specific rules when converting strings to upper or lower case and " +
-        "Rust does not, so we disable upper and lower by default.")
-      .booleanConf
-      .createWithDefault(false)
-
   val COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK: ConfigEntry[Boolean] =
     conf("spark.comet.scan.unsignedSmallIntSafetyCheck")
       .category(CATEGORY_SCAN)

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -191,14 +191,13 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       classOf[StartsWith] -> CometScalarFunction("starts_with"),
       classOf[StringInstr] -> CometScalarFunction("instr"),
       classOf[StringRepeat] -> CometStringRepeat,
-      classOf[StringReplace] -> CometScalarFunction("replace"),
+      classOf[StringReplace] -> CometStringReplace,
       classOf[StringRPad] -> CometStringRPad,
       classOf[StringLPad] -> CometStringLPad,
       classOf[StringSpace] -> CometScalarFunction("space"),
       classOf[StringSplit] -> CometStringSplit,
       classOf[StringTranslate] -> CometScalarFunction("translate"),
       classOf[StringTrim] -> CometScalarFunction("trim"),
-      classOf[StringTrimBoth] -> CometScalarFunction("btrim"),
       classOf[StringTrimLeft] -> CometScalarFunction("ltrim"),
       classOf[StringTrimRight] -> CometScalarFunction("rtrim"),
       classOf[Left] -> CometLeft,

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

@@ -21,7 +21,7 @@ package org.apache.comet.serde
 
 import java.util.Locale
 
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Concat, ConcatWs, Expression, GetJsonObject, If, InitCap, IsNull, Left, Length, Like, Literal, Lower, RegExpReplace, Right, RLike, StringLPad, StringRepeat, StringRPad, StringSplit, Substring, SubstringIndex, Upper}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Concat, ConcatWs, Expression, GetJsonObject, If, InitCap, IsNull, Left, Length, Like, Literal, Lower, RegExpReplace, Right, RLike, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, Substring, SubstringIndex, Upper}
 import org.apache.spark.sql.types.{BinaryType, DataTypes, LongType, StringType}
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -33,10 +33,6 @@ import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInter
 
 object CometStringRepeat extends CometExpressionSerde[StringRepeat] {
 
-  override def getCompatibleNotes(): Seq[String] = Seq(
-    "A negative argument for the number of times to repeat throws an exception" +
-      " instead of returning an empty string as Spark does")
-
   override def convert(
       expr: StringRepeat,
       inputs: Seq[Attribute],
@@ -54,21 +50,13 @@ object CometStringRepeat extends CometExpressionSerde[StringRepeat] {
 class CometCaseConversionBase[T <: Expression](function: String)
     extends CometScalarFunction[T](function) {
 
-  override def getIncompatibleReasons(): Seq[String] = Seq(
-    "Results can vary depending on locale and character set." +
-      s" Requires `${CometConf.COMET_CASE_CONVERSION_ENABLED.key}=true` to enable.")
+  private val incompatReason =
+    "Results can vary depending on locale and character set " +
+      "(https://github.com/apache/datafusion-comet/issues/2190)."
 
-  override def convert(expr: T, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = {
-    if (!CometConf.COMET_CASE_CONVERSION_ENABLED.get()) {
-      withInfo(
-        expr,
-        "Comet is not compatible with Spark for case conversion in " +
-          s"locale-specific cases. Set ${CometConf.COMET_CASE_CONVERSION_ENABLED.key}=true " +
-          "to enable it anyway.")
-      return None
-    }
-    super.convert(expr, inputs, binding)
-  }
+  override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason)
+
+  override def getSupportLevel(expr: T): SupportLevel = Incompatible(Some(incompatReason))
 }
 
 object CometUpper extends CometCaseConversionBase[Upper]("upper")
@@ -228,6 +216,21 @@ object CometRight extends CometExpressionSerde[Right] {
   }
 }
 
+object CometStringReplace extends CometScalarFunction[StringReplace]("replace") {
+
+  private val emptySearchReason =
+    "Empty `search` string produces different output: Spark returns `str` unchanged, " +
+      "DataFusion inserts `replace` between every character " +
+      "(https://github.com/apache/datafusion-comet/issues/4497)."
+
+  override def getIncompatibleReasons(): Seq[String] = Seq(emptySearchReason)
+
+  override def getSupportLevel(expr: StringReplace): SupportLevel = expr.searchExpr match {
+    case Literal(s: UTF8String, _) if s.numBytes == 0 => Incompatible(Some(emptySearchReason))
+    case _ => Compatible()
+  }
+}
+
 object CometConcat extends CometScalarFunction[Concat]("concat") {
   val unsupportedReason = "CONCAT supports only string input parameters"
 

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

@@ -21,9 +21,9 @@ CREATE TABLE test_lower(s string) USING parquet
 statement
 INSERT INTO test_lower VALUES ('HELLO'), ('hello'), ('Hello World'), (''), (NULL), ('123ABC')
 
-query expect_fallback(case conversion)
+query expect_fallback(locale and character set)
 SELECT lower(s) FROM test_lower
 
 -- literal arguments
-query expect_fallback(case conversion)
+query expect_fallback(locale and character set)
 SELECT lower('HELLO'), lower(''), lower(NULL)

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

@@ -15,8 +15,8 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
--- Test lower() with case conversion enabled (happy path)
--- Config: spark.comet.caseConversion.enabled=true
+-- Test lower() with the standard allowIncompatible opt-in (happy path)
+-- Config: spark.comet.expression.Lower.allowIncompatible=true
 
 statement
 CREATE TABLE test_lower_enabled(s string) USING parquet

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

@@ -24,7 +24,8 @@ INSERT INTO test_str_replace VALUES ('hello world', 'world', 'there'), ('aaa', '
 query
 SELECT replace(s, search, replace) FROM test_str_replace
 
-query ignore(https://github.com/apache/datafusion-comet/issues/3344)
+-- Comet returns 'xhxexlxlxox' where Spark returns 'hello' (short-circuits on empty search).
+query expect_fallback(Empty `search`)
 SELECT replace('hello', '', 'x')
 
 -- column + literal + literal

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

@@ -21,9 +21,9 @@ CREATE TABLE test_upper(s string) USING parquet
 statement
 INSERT INTO test_upper VALUES ('hello'), ('HELLO'), ('Hello World'), (''), (NULL), ('123abc')
 
-query expect_fallback(case conversion)
+query expect_fallback(locale and character set)
 SELECT upper(s) FROM test_upper
 
 -- literal arguments
-query expect_fallback(case conversion)
+query expect_fallback(locale and character set)
 SELECT upper('hello'), upper(''), upper(NULL)

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

@@ -15,8 +15,8 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
--- Test upper() with case conversion enabled (happy path)
--- Config: spark.comet.caseConversion.enabled=true
+-- Test upper() with the standard allowIncompatible opt-in (happy path)
+-- Config: spark.comet.expression.Upper.allowIncompatible=true
 
 statement
 CREATE TABLE test_upper_enabled(s string) USING parquet

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

@@ -261,7 +261,9 @@ class CometStringExpressionSuite extends CometTestBase {
   }
 
   test("Upper and Lower") {
-    withSQLConf(CometConf.COMET_CASE_CONVERSION_ENABLED.key -> "true") {
+    withSQLConf(
+      CometConf.getExprAllowIncompatConfigKey("Upper") -> "true",
+      CometConf.getExprAllowIncompatConfigKey("Lower") -> "true") {
       val table = "names"
       withTable(table) {
         sql(s"create table $table(id int, name varchar(20)) using parquet")
@@ -339,7 +341,7 @@ class CometStringExpressionSuite extends CometTestBase {
   }
 
   test("trim") {
-    withSQLConf(CometConf.COMET_CASE_CONVERSION_ENABLED.key -> "true") {
+    withSQLConf(CometConf.getExprAllowIncompatConfigKey("Upper") -> "true") {
       val table = "test"
       withTable(table) {
         sql(s"create table $table(col varchar(20)) using parquet")