sunchao commented on code in PR #5409:
URL: https://github.com/apache/datafusion-comet/pull/5409#discussion_r3832146714
##########
spark/src/main/scala/org/apache/comet/serde/strings.scala:
##########
@@ -178,29 +179,56 @@ object CometStringReplace
extends CometScalarFunction[StringReplace]("replace")
with NativeOptInAvailable {
+ /**
+ * Native DataFusion `replace` differs from Spark only when the search
string is empty (Spark
+ * returns `src` unchanged; DataFusion inserts the replacement between every
character). That
+ * case is decidable at plan time when `search` is a literal.
+ *
+ * The native kernel is also byte-level `UTF8_BINARY` only, so non-default
collations stay on
+ * the dispatcher. https://github.com/apache/datafusion-comet/issues/4496
+ */
+ private def nativeSafeSearchSubset(expr: StringReplace): Boolean = {
+ val children = expr.children
+ if (children.length != 3) {
+ return false
+ }
+ val searchIsNonEmptyLiteral = children(1) match {
+ case Literal(v: UTF8String, _) => v != null && v.numBytes() > 0
Review Comment:
[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.
##########
spark/src/main/scala/org/apache/comet/serde/strings.scala:
##########
@@ -178,29 +179,56 @@ object CometStringReplace
extends CometScalarFunction[StringReplace]("replace")
with NativeOptInAvailable {
+ /**
+ * Native DataFusion `replace` differs from Spark only when the search
string is empty (Spark
+ * returns `src` unchanged; DataFusion inserts the replacement between every
character). That
+ * case is decidable at plan time when `search` is a literal.
+ *
+ * The native kernel is also byte-level `UTF8_BINARY` only, so non-default
collations stay on
+ * the dispatcher. https://github.com/apache/datafusion-comet/issues/4496
+ */
+ private def nativeSafeSearchSubset(expr: StringReplace): Boolean = {
+ val children = expr.children
+ if (children.length != 3) {
+ return false
+ }
+ val searchIsNonEmptyLiteral = children(1) match {
+ case Literal(v: UTF8String, _) => v != null && v.numBytes() > 0
+ case _ => false
+ }
+ val utf8BinaryCollation =
+ !children.exists(c => QueryPlanSerde.isStringCollationType(c.dataType))
+ utf8BinaryCollation && searchIsNonEmptyLiteral
+ }
+
+ override def getCompatibleNotes(): Seq[String] =
+ Seq(
+ "When `search` is a non-empty `UTF8_BINARY` literal, Comet evaluates
`replace` natively " +
+ "by default.")
+
override def getIncompatibleReasons(): Seq[String] =
Seq("Produces different results from Spark when the search string is
empty")
override def getSupportLevel(expr: StringReplace): SupportLevel =
- if (!CometConf.isExprAllowIncompat(getExprConfigName(expr))) {
+ if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) ||
nativeSafeSearchSubset(expr)) {
+ Compatible()
+ } else {
Compatible(nativeOptIn =
Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr)))))
- } else {
- Compatible()
}
override def convert(
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 (CometConf.isExprAllowIncompat(getExprConfigName(expr)) ||
nativeSafeSearchSubset(expr)) {
Review Comment:
[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.
##########
spark/src/main/scala/org/apache/comet/serde/strings.scala:
##########
@@ -178,29 +179,56 @@ object CometStringReplace
extends CometScalarFunction[StringReplace]("replace")
with NativeOptInAvailable {
+ /**
+ * Native DataFusion `replace` differs from Spark only when the search
string is empty (Spark
+ * returns `src` unchanged; DataFusion inserts the replacement between every
character). That
+ * case is decidable at plan time when `search` is a literal.
+ *
+ * The native kernel is also byte-level `UTF8_BINARY` only, so non-default
collations stay on
+ * the dispatcher. https://github.com/apache/datafusion-comet/issues/4496
+ */
+ private def nativeSafeSearchSubset(expr: StringReplace): Boolean = {
+ val children = expr.children
+ if (children.length != 3) {
+ return false
+ }
+ val searchIsNonEmptyLiteral = children(1) match {
+ case Literal(v: UTF8String, _) => v != null && v.numBytes() > 0
+ case _ => false
+ }
+ val utf8BinaryCollation =
+ !children.exists(c => QueryPlanSerde.isStringCollationType(c.dataType))
+ utf8BinaryCollation && searchIsNonEmptyLiteral
+ }
+
+ override def getCompatibleNotes(): Seq[String] =
+ Seq(
+ "When `search` is a non-empty `UTF8_BINARY` literal, Comet evaluates
`replace` natively " +
+ "by default.")
+
override def getIncompatibleReasons(): Seq[String] =
Seq("Produces different results from Spark when the search string is
empty")
override def getSupportLevel(expr: StringReplace): SupportLevel =
- if (!CometConf.isExprAllowIncompat(getExprConfigName(expr))) {
+ if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) ||
nativeSafeSearchSubset(expr)) {
+ Compatible()
+ } else {
Compatible(nativeOptIn =
Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr)))))
- } else {
- Compatible()
}
override def convert(
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 (CometConf.isExprAllowIncompat(getExprConfigName(expr)) ||
nativeSafeSearchSubset(expr)) {
+ // Native DataFusion `replace` matches Spark when search is a non-empty
UTF8_BINARY
+ // literal (the common case, selected by default) and when the user has
opted in.
super.convert(expr, inputs, binding)
Review Comment:
[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
