From 3b0d9d405b477218fa6c664919f020ec1ae2bc99 Mon Sep 17 00:00:00 2001 From: Hung Date: Fri, 21 Aug 2026 23:49:52 +0800 Subject: [PATCH 1/5] feat: default native replace for non-empty UTF8_BINARY literal search (#5354) --- .../expression-audits/string_funcs.md | 5 +- .../org/apache/comet/serde/strings.scala | 46 ++++++++++++---- .../expressions/string/string_replace.sql | 17 +++++- .../org/apache/comet/CometCodegenSuite.scala | 53 +++++++++++++++++++ 4 files changed, 109 insertions(+), 12 deletions(-) diff --git a/docs/source/contributor-guide/expression-audits/string_funcs.md b/docs/source/contributor-guide/expression-audits/string_funcs.md index 58b300ebc12..ad286e86206 100644 --- a/docs/source/contributor-guide/expression-audits/string_funcs.md +++ b/docs/source/contributor-guide/expression-audits/string_funcs.md @@ -168,7 +168,10 @@ ## 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`). DataFusion `replace` instead inserts `replace` between every character. + - Comet evaluates `replace` natively by default when `search` is a non-empty `UTF8_BINARY` literal. + - Empty literal search, non-literal search, and non-default collations stay on the JVM codegen dispatcher. + - Users can still opt into the native (potentially incompatible) path for remaining cases via `spark.comet.expression.StringReplace.allowIncompatible=true`. - 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)). ## right diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index ebf45089882..451bae1f34b 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -21,6 +21,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, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper} import org.apache.spark.sql.types.{BinaryType, DataTypes, IntegerType, LongType, StringType} +import org.apache.spark.unsafe.types.UTF8String import org.apache.comet.CometConf import org.apache.comet.serde.ExprOuterClass.Expr @@ -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) } else { - // Run Spark's own generated code inside the Comet pipeline so the result matches Spark - // exactly. Falls back to Spark when the codegen dispatcher is disabled. + // Empty literal search, non-literal search, or a non-default collation: run Spark's + // own generated code inside the Comet pipeline. Falls back to Spark when the + // codegen dispatcher is disabled. CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding) } } diff --git a/spark/src/test/resources/sql-tests/expressions/string/string_replace.sql b/spark/src/test/resources/sql-tests/expressions/string/string_replace.sql index ad09525b6d9..bd88bd89ed9 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/string_replace.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/string_replace.sql @@ -19,7 +19,7 @@ statement CREATE TABLE test_str_replace(s string, search string, replace string) USING parquet statement -INSERT INTO test_str_replace VALUES ('hello world', 'world', 'there'), ('aaa', 'a', 'bb'), ('hello', 'xyz', 'abc'), ('', 'a', 'b'), (NULL, 'a', 'b'), ('hello', '', 'x') +INSERT INTO test_str_replace VALUES ('hello world', 'world', 'there'), ('aaa', 'a', 'bb'), ('hello', 'xyz', 'abc'), ('', 'a', 'b'), (NULL, 'a', 'b'), ('hello', '', 'x'), ('aaaa', 'aa', 'x'), ('你好你好', '你好', 'X'), ('😀a😀', '😀', 'x') query SELECT replace(s, search, replace) FROM test_str_replace @@ -27,7 +27,8 @@ 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. +-- Spark's own doGenCode handles this case. Non-empty UTF8_BINARY literal +-- search takes the native path by default (#5354). -- https://github.com/apache/datafusion-comet/issues/4497 query SELECT replace('hello', '', 'x') @@ -41,6 +42,18 @@ SELECT replace(NULL, '', 'x') query SELECT replace('hello', '', NULL) +-- Overlapping candidates: Spark replaces non-overlapping left-to-right +-- ('aaaa' + 'aa' -> 'xx'). Column source + literal search takes the native path. +query +SELECT replace(s, 'aa', 'x') FROM test_str_replace WHERE s = 'aaaa' + +-- Multi-byte UTF-8 values. Column source + literal search takes the native path. +query +SELECT replace(s, '你好', 'X') FROM test_str_replace WHERE s = '你好你好' + +query +SELECT replace(s, '😀', 'x') FROM test_str_replace WHERE s = '😀a😀' + -- column + literal + literal query SELECT replace(s, 'world', 'there') FROM test_str_replace diff --git a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala index 5806cb35015..f5cd11f8db4 100644 --- a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala @@ -31,6 +31,7 @@ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.spark.unsafe.types.UTF8String +import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus import org.apache.comet.CometSparkSessionExtensions.isSpark41Plus import org.apache.comet.codegen.CometBatchKernelCodegen import org.apache.comet.codegen.CometBatchKernelCodegen.ArrowColumnSpec @@ -328,6 +329,58 @@ class CometCodegenSuite } } + test("replace routes native vs JVM codegen dispatcher based on non-empty literal search") { + withTable("t") { + sql("CREATE TABLE t (s STRING, search STRING) USING parquet") + sql("INSERT INTO t VALUES ('hello world', 'world'), ('abcabc', 'world')") + + withSQLConf( + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXEC_PROJECT_ENABLED.key -> "true", + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key -> + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE) { + + val dfNative = sql("SELECT replace(s, 'world', 'comet') FROM t") + checkSparkAnswerAndOperator(dfNative) + val explainNative = + new ExtendedExplainInfo().generateExtendedInfo(dfNative.queryExecution.executedPlan) + assert( + !explainNative.contains("JVM codegen dispatcher: replace"), + s"expected native path for non-empty literal search, got:\n$explainNative") + + val dfEmptySearch = sql("SELECT replace(s, '', 'comet') FROM t") + checkSparkAnswerAndOperator(dfEmptySearch) + val explainEmptySearch = + new ExtendedExplainInfo().generateExtendedInfo( + dfEmptySearch.queryExecution.executedPlan) + assert( + explainEmptySearch.contains("JVM codegen dispatcher: replace"), + s"expected dispatcher path for empty literal search, got:\n$explainEmptySearch") + + val dfNonLiteralSearch = sql("SELECT replace(s, search, 'comet') FROM t") + checkSparkAnswerAndOperator(dfNonLiteralSearch) + val explainNonLiteralSearch = + new ExtendedExplainInfo().generateExtendedInfo( + dfNonLiteralSearch.queryExecution.executedPlan) + assert( + explainNonLiteralSearch.contains("JVM codegen dispatcher: replace"), + s"expected dispatcher path for non-literal search, got:\n$explainNonLiteralSearch") + + if (isSpark40Plus) { + val dfCollated = + sql("SELECT replace(CAST(s AS STRING COLLATE UTF8_LCASE), 'world', 'comet') FROM t") + checkSparkAnswerAndOperator(dfCollated) + val explainCollated = + new ExtendedExplainInfo().generateExtendedInfo(dfCollated.queryExecution.executedPlan) + assert( + explainCollated.contains("JVM codegen dispatcher: replace"), + s"expected dispatcher path for non-UTF8_BINARY collation, got:\n$explainCollated") + } + } + } + } + test("codegen dispatch fallback reasons name the expression") { // Flag-off short-circuit tags the expression `: ` so distinct expressions // don't collapse in the `Set[String]` roll-up. From d223e8f1f90d15abb594eec941b917a1cd2fa780 Mon Sep 17 00:00:00 2001 From: Hung Date: Sat, 22 Aug 2026 09:43:18 +0800 Subject: [PATCH 2/5] fix: narrow replace native-safe subset for expression-boundary cases --- .../expression-audits/string_funcs.md | 4 +- .../org/apache/comet/serde/strings.scala | 67 +++++++++++++----- .../org/apache/comet/CometCodegenSuite.scala | 68 ++++++++++++++++++- 3 files changed, 117 insertions(+), 22 deletions(-) diff --git a/docs/source/contributor-guide/expression-audits/string_funcs.md b/docs/source/contributor-guide/expression-audits/string_funcs.md index ad286e86206..2159cdc66b0 100644 --- a/docs/source/contributor-guide/expression-audits/string_funcs.md +++ b/docs/source/contributor-guide/expression-audits/string_funcs.md @@ -169,8 +169,8 @@ - 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. - - Comet evaluates `replace` natively by default when `search` is a non-empty `UTF8_BINARY` literal. - - Empty literal search, non-literal search, and non-default collations stay on the JVM codegen dispatcher. + - Comet evaluates `replace` natively by default when `search` is a short, well-formed, non-empty `UTF8_BINARY` literal and `replace` is a short well-formed literal or a column. + - Empty, malformed, or oversized literals, a non-literal / non-column replacement (including expressions that can throw), and non-default collations stay on the JVM codegen dispatcher. The kernel match is not enough: `CometLiteral` is not byte-preserving, DataFusion evaluates every child before `replace`, and large scalars overflow Arrow `Utf8` offsets when broadcast. - Users can still opt into the native (potentially incompatible) path for remaining cases via `spark.comet.expression.StringReplace.allowIncompatible=true`. - 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)). diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index 451bae1f34b..618711136f3 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -19,7 +19,10 @@ 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, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper} +import java.nio.charset.StandardCharsets +import java.util.Arrays + +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, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper} import org.apache.spark.sql.types.{BinaryType, DataTypes, IntegerType, LongType, StringType} import org.apache.spark.unsafe.types.UTF8String @@ -180,37 +183,66 @@ object CometStringReplace 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 DataFusion `replace` kernel matches Spark only for a non-empty search string. Kernel + * compatibility is not enough: `CometLiteral` serializes strings via `UTF8String.toString` + * (malformed UTF-8 becomes U+FFFD), DataFusion evaluates every child before `replace` (so a + * NULL `src` does not skip a throwing replacement), and scalar literals are broadcast into + * Arrow `Utf8` arrays that overflow 32-bit offsets on a large batch. * - * 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 + * The default native path is therefore limited to a plan-time subset that avoids those + * boundaries. Non-default collations stay on the dispatcher. + * https://github.com/apache/datafusion-comet/issues/4496 */ - private def nativeSafeSearchSubset(expr: StringReplace): Boolean = { + private def nativeSafeSubset(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 + val searchIsSafe = children(1) match { + case Literal(v: UTF8String, _) => isNativeSafeStringLiteral(v, allowEmpty = false) + case _ => false + } + val replacementIsSafe = children(2) match { + case Literal(null, _) => true + case Literal(v: UTF8String, _) => isNativeSafeStringLiteral(v, allowEmpty = true) + case _: Attribute | _: BoundReference => true case _ => false } val utf8BinaryCollation = !children.exists(c => QueryPlanSerde.isStringCollationType(c.dataType)) - utf8BinaryCollation && searchIsNonEmptyLiteral + utf8BinaryCollation && searchIsSafe && replacementIsSafe + } + + /** + * `CometLiteral` encodes a string as `UTF8String.toString`, so only literals whose bytes + * survive that round-trip can be sent natively. The size cap keeps a broadcast scalar under + * Arrow `Utf8`'s 32-bit offset limit at `spark.comet.batchSize` rows. + */ + private def isNativeSafeStringLiteral(v: UTF8String, allowEmpty: Boolean): Boolean = { + if (v == null) { + return false + } + if (!allowEmpty && v.numBytes() == 0) { + return false + } + val maxBytes = Int.MaxValue / math.max(CometConf.COMET_BATCH_SIZE.get(), 1) + if (v.numBytes() > maxBytes) { + return false + } + Arrays.equals(v.getBytes, v.toString.getBytes(StandardCharsets.UTF_8)) } override def getCompatibleNotes(): Seq[String] = Seq( - "When `search` is a non-empty `UTF8_BINARY` literal, Comet evaluates `replace` natively " + + "When `search` is a short, well-formed, non-empty `UTF8_BINARY` literal and `replace` " + + "is a short well-formed literal or a column, 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)) || nativeSafeSearchSubset(expr)) { + if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) || nativeSafeSubset(expr)) { Compatible() } else { Compatible(nativeOptIn = @@ -221,14 +253,13 @@ object CometStringReplace expr: StringReplace, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = { - 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. + if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) || nativeSafeSubset(expr)) { + // Native path for the plan-time safe subset, or when the user has opted in. super.convert(expr, inputs, binding) } else { - // Empty literal search, non-literal search, or a non-default collation: run Spark's - // own generated code inside the Comet pipeline. Falls back to Spark when the - // codegen dispatcher is disabled. + // Empty / malformed / oversized search, a throwing or non-column replacement, or a + // non-default collation: run Spark's own generated code inside the Comet pipeline. + // Falls back to Spark when the codegen dispatcher is disabled. CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding) } } diff --git a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala index f5cd11f8db4..529836e9144 100644 --- a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala @@ -331,8 +331,8 @@ class CometCodegenSuite test("replace routes native vs JVM codegen dispatcher based on non-empty literal search") { withTable("t") { - sql("CREATE TABLE t (s STRING, search STRING) USING parquet") - sql("INSERT INTO t VALUES ('hello world', 'world'), ('abcabc', 'world')") + sql("CREATE TABLE t (s STRING, search STRING, r STRING) USING parquet") + sql("INSERT INTO t VALUES ('hello world', 'world', 'comet'), ('abcabc', 'world', 'comet')") withSQLConf( CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true", @@ -349,6 +349,14 @@ class CometCodegenSuite !explainNative.contains("JVM codegen dispatcher: replace"), s"expected native path for non-empty literal search, got:\n$explainNative") + val dfColReplace = sql("SELECT replace(s, 'world', r) FROM t") + checkSparkAnswerAndOperator(dfColReplace) + val explainColReplace = + new ExtendedExplainInfo().generateExtendedInfo(dfColReplace.queryExecution.executedPlan) + assert( + !explainColReplace.contains("JVM codegen dispatcher: replace"), + s"expected native path for column replacement, got:\n$explainColReplace") + val dfEmptySearch = sql("SELECT replace(s, '', 'comet') FROM t") checkSparkAnswerAndOperator(dfEmptySearch) val explainEmptySearch = @@ -381,6 +389,62 @@ class CometCodegenSuite } } + test("replace stays on dispatcher for expression-boundary incompatibilities") { + withSQLConf( + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.key -> "true", + CometConf.COMET_EXEC_PROJECT_ENABLED.key -> "true", + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key -> + CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE) { + + def assertDispatcher(df: org.apache.spark.sql.DataFrame, clue: String): Unit = { + checkSparkAnswerAndOperator(df) + val explain = + new ExtendedExplainInfo().generateExtendedInfo(df.queryExecution.executedPlan) + assert(explain.contains("JVM codegen dispatcher: replace"), s"$clue, got:\n$explain") + } + + // Malformed search: CometLiteral would normalize 0xFF to U+FFFD, incorrectly matching + // a well-formed U+FFFD in the source. + withTable("t") { + sql("CREATE TABLE t (s STRING) USING parquet") + sql("INSERT INTO t VALUES ('\uFFFD'), ('ok')") + assertDispatcher( + sql("SELECT replace(s, CAST(X'FF' AS STRING), 'x') FROM t"), + "expected dispatcher path for malformed search literal") + } + + // Malformed replacement has the same serialization hazard. + withTable("t") { + sql("CREATE TABLE t (s STRING) USING parquet") + sql("INSERT INTO t VALUES ('a'), ('b')") + assertDispatcher( + sql("SELECT replace(s, 'a', CAST(X'FF' AS STRING)) FROM t"), + "expected dispatcher path for malformed replacement literal") + } + + // Spark skips replacement evaluation when src is NULL; native evaluates every child. + withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") { + withTable("t") { + sql("CREATE TABLE t (s STRING, n INT) USING parquet") + sql("INSERT INTO t VALUES (NULL, 0), ('a', 1)") + assertDispatcher( + sql("SELECT replace(s, 'a', CAST(1 / n AS STRING)) FROM t"), + "expected dispatcher path for throwing replacement expression") + } + } + + // A 256 KiB scalar replacement overflows Arrow Utf8 offsets when broadcast to 8192 rows. + withTable("t") { + sql("CREATE TABLE t (s STRING) USING parquet") + sql("INSERT INTO t VALUES ('hello')") + assertDispatcher( + sql("SELECT replace(s, 'notfound', repeat('x', 262144)) FROM t"), + "expected dispatcher path for oversized replacement literal") + } + } + } + test("codegen dispatch fallback reasons name the expression") { // Flag-off short-circuit tags the expression `: ` so distinct expressions // don't collapse in the `Set[String]` roll-up. From db8e31d2cb9d1d63798fb26cbf7318920aecd394 Mon Sep 17 00:00:00 2001 From: Hung Date: Sun, 23 Aug 2026 10:30:21 +0800 Subject: [PATCH 3/5] fix: whitelist replace source the same way as replacement --- .../expression-audits/string_funcs.md | 4 +- .../org/apache/comet/serde/strings.scala | 35 ++++++++++------- .../org/apache/comet/CometCodegenSuite.scala | 38 +++++++++++++++++++ 3 files changed, 61 insertions(+), 16 deletions(-) diff --git a/docs/source/contributor-guide/expression-audits/string_funcs.md b/docs/source/contributor-guide/expression-audits/string_funcs.md index 2159cdc66b0..69b1f19c64e 100644 --- a/docs/source/contributor-guide/expression-audits/string_funcs.md +++ b/docs/source/contributor-guide/expression-audits/string_funcs.md @@ -169,8 +169,8 @@ - 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. - - Comet evaluates `replace` natively by default when `search` is a short, well-formed, non-empty `UTF8_BINARY` literal and `replace` is a short well-formed literal or a column. - - Empty, malformed, or oversized literals, a non-literal / non-column replacement (including expressions that can throw), and non-default collations stay on the JVM codegen dispatcher. The kernel match is not enough: `CometLiteral` is not byte-preserving, DataFusion evaluates every child before `replace`, and large scalars overflow Arrow `Utf8` offsets when broadcast. + - Comet evaluates `replace` natively by default when `src` and `replace` are each a short well-formed literal or a column, and `search` is a short, well-formed, non-empty `UTF8_BINARY` literal. + - Nested source/replacement expressions (`substring`, `concat`, casts that can throw), empty / malformed / oversized literals, and non-default collations stay on the JVM codegen dispatcher. The kernel match is not enough: `CometLiteral` is not byte-preserving, DataFusion evaluates every child before `replace`, and large scalars overflow Arrow `Utf8` offsets when broadcast. - Users can still opt into the native (potentially incompatible) path for remaining cases via `spark.comet.expression.StringReplace.allowIncompatible=true`. - 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)). diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index 618711136f3..3bf03732aa9 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -190,28 +190,35 @@ object CometStringReplace * Arrow `Utf8` arrays that overflow 32-bit offsets on a large batch. * * The default native path is therefore limited to a plan-time subset that avoids those - * boundaries. Non-default collations stay on the dispatcher. - * https://github.com/apache/datafusion-comet/issues/4496 + * boundaries. `src` uses the same whitelist as `replace`: a short well-formed literal or a + * column. Nested expressions (`substring`, `concat`, …) stay on the dispatcher so a throwing or + * malformed child cannot hide inside the source tree. Non-default collations stay on the + * dispatcher. https://github.com/apache/datafusion-comet/issues/4496 */ private def nativeSafeSubset(expr: StringReplace): Boolean = { val children = expr.children if (children.length != 3) { return false } + val sourceIsSafe = isNativeSafeStringArg(children(0), allowEmptyLiteral = true) val searchIsSafe = children(1) match { case Literal(v: UTF8String, _) => isNativeSafeStringLiteral(v, allowEmpty = false) case _ => false } - val replacementIsSafe = children(2) match { + val replacementIsSafe = isNativeSafeStringArg(children(2), allowEmptyLiteral = true) + val utf8BinaryCollation = + !children.exists(c => QueryPlanSerde.isStringCollationType(c.dataType)) + utf8BinaryCollation && sourceIsSafe && searchIsSafe && replacementIsSafe + } + + /** A column, null literal, or a short well-formed string literal. */ + private def isNativeSafeStringArg(expr: Expression, allowEmptyLiteral: Boolean): Boolean = + expr match { case Literal(null, _) => true - case Literal(v: UTF8String, _) => isNativeSafeStringLiteral(v, allowEmpty = true) + case Literal(v: UTF8String, _) => isNativeSafeStringLiteral(v, allowEmptyLiteral) case _: Attribute | _: BoundReference => true case _ => false } - val utf8BinaryCollation = - !children.exists(c => QueryPlanSerde.isStringCollationType(c.dataType)) - utf8BinaryCollation && searchIsSafe && replacementIsSafe - } /** * `CometLiteral` encodes a string as `UTF8String.toString`, so only literals whose bytes @@ -234,9 +241,9 @@ object CometStringReplace override def getCompatibleNotes(): Seq[String] = Seq( - "When `search` is a short, well-formed, non-empty `UTF8_BINARY` literal and `replace` " + - "is a short well-formed literal or a column, Comet evaluates `replace` natively " + - "by default.") + "When `src` and `replace` are each a short well-formed literal or a column, and " + + "`search` is a short, well-formed, 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") @@ -257,9 +264,9 @@ object CometStringReplace // Native path for the plan-time safe subset, or when the user has opted in. super.convert(expr, inputs, binding) } else { - // Empty / malformed / oversized search, a throwing or non-column replacement, or a - // non-default collation: run Spark's own generated code inside the Comet pipeline. - // Falls back to Spark when the codegen dispatcher is disabled. + // Nested / malformed / oversized / throwing children, or a non-default collation: run + // Spark's own generated code inside the Comet pipeline. Falls back to Spark when the + // codegen dispatcher is disabled. CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding) } } diff --git a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala index 529836e9144..f3387dfa163 100644 --- a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala @@ -442,6 +442,44 @@ class CometCodegenSuite sql("SELECT replace(s, 'notfound', repeat('x', 262144)) FROM t"), "expected dispatcher path for oversized replacement literal") } + + // Source is not on the whitelist: Spark short-circuits inside substring when s is NULL. + withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") { + withTable("t") { + sql("CREATE TABLE t (s STRING, n INT) USING parquet") + sql("INSERT INTO t VALUES (NULL, 0), ('a', 1)") + assertDispatcher( + sql("SELECT replace(substring(s, 1, CAST(1 / n AS INT)), 'a', 'x') FROM t"), + "expected dispatcher path for throwing expression nested in source") + } + } + + // Malformed source literal: same CometLiteral byte-normalization as search/replacement. + withTable("t") { + sql("CREATE TABLE t (r STRING) USING parquet") + sql("INSERT INTO t VALUES ('x')") + assertDispatcher( + sql("SELECT replace(CAST(X'FF' AS STRING), 'a', r) FROM t"), + "expected dispatcher path for malformed source literal") + } + + // Malformed literal nested under concat is still in the source tree. + withTable("t") { + sql("CREATE TABLE t (r STRING) USING parquet") + sql("INSERT INTO t VALUES ('x')") + assertDispatcher( + sql("SELECT replace(concat(CAST(X'FF' AS STRING), r), 'a', 'x') FROM t"), + "expected dispatcher path for malformed literal nested in source") + } + + // Oversized source literal has the same broadcast / offset-overflow hazard. + withTable("t") { + sql("CREATE TABLE t (r STRING) USING parquet") + sql("INSERT INTO t VALUES ('x')") + assertDispatcher( + sql("SELECT replace(repeat('x', 262144), 'notfound', r) FROM t"), + "expected dispatcher path for oversized source literal") + } } } From 80c0737c67e8d2ea1f7e655e9e2b6810677c4166 Mon Sep 17 00:00:00 2001 From: Hung Date: Sun, 23 Aug 2026 16:54:30 +0800 Subject: [PATCH 4/5] fix: bound native explode output batch size --- .../src/execution/operators/batch_split.rs | 154 ++++++++++++++++++ native/core/src/execution/operators/mod.rs | 2 + native/core/src/execution/planner.rs | 15 +- .../org/apache/comet/serde/strings.scala | 4 +- .../org/apache/comet/CometCodegenSuite.scala | 33 ++++ 5 files changed, 204 insertions(+), 4 deletions(-) create mode 100644 native/core/src/execution/operators/batch_split.rs diff --git a/native/core/src/execution/operators/batch_split.rs b/native/core/src/execution/operators/batch_split.rs new file mode 100644 index 00000000000..cb536370d52 --- /dev/null +++ b/native/core/src/execution/operators/batch_split.rs @@ -0,0 +1,154 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::fmt::Formatter; +use std::sync::Arc; + +use datafusion::common::{internal_err, Result}; +use datafusion::execution::TaskContext; +use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet, SplitMetrics}; +use datafusion::physical_plan::stream::BatchSplitStream; +use datafusion::physical_plan::{ + DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, SendableRecordBatchStream, +}; + +/// Splits oversized batches emitted by an input plan using the runtime batch size. +/// +/// DataFusion 54.1.0's `UnnestExec` can emit more rows than the configured batch size. This +/// wrapper is a temporary downstream boundary for Comet's explode path until Comet upgrades to a +/// DataFusion version containing https://github.com/apache/datafusion/pull/24384. +#[derive(Debug)] +pub struct BatchSplitExec { + input: Arc, + cache: Arc, + metrics: ExecutionPlanMetricsSet, +} + +impl BatchSplitExec { + pub fn new(input: Arc) -> Self { + Self { + cache: Arc::clone(input.properties()), + input, + metrics: ExecutionPlanMetricsSet::new(), + } + } +} + +impl DisplayAs for BatchSplitExec { + fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result { + match t { + DisplayFormatType::Default | DisplayFormatType::Verbose => { + write!(f, "CometBatchSplitExec") + } + DisplayFormatType::TreeRender => unimplemented!(), + } + } +} + +impl ExecutionPlan for BatchSplitExec { + fn name(&self) -> &str { + "CometBatchSplitExec" + } + + fn properties(&self) -> &Arc { + &self.cache + } + + fn children(&self) -> Vec<&Arc> { + vec![&self.input] + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> Result> { + if children.len() != 1 { + return internal_err!( + "CometBatchSplitExec expects one child, got {}", + children.len() + ); + } + Ok(Arc::new(Self::new(Arc::clone(&children[0])))) + } + + fn execute( + &self, + partition: usize, + context: Arc, + ) -> Result { + let batch_size = context.session_config().batch_size(); + let input = self.input.execute(partition, context)?; + Ok(Box::pin(BatchSplitStream::new( + input, + batch_size, + SplitMetrics::new(&self.metrics, partition), + ))) + } + + fn metrics(&self) -> Option { + Some(self.metrics.clone_inner()) + } + + fn maintains_input_order(&self) -> Vec { + vec![true] + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use arrow::array::{AsArray, Int32Array, RecordBatch}; + use arrow::datatypes::{DataType, Field, Schema}; + use datafusion::datasource::memory::MemorySourceConfig; + use datafusion::datasource::source::DataSourceExec; + use datafusion::execution::SessionStateBuilder; + use datafusion::prelude::{SessionConfig, SessionContext}; + use futures::StreamExt; + + #[tokio::test] + async fn splits_batches_at_the_runtime_batch_size_and_preserves_order() { + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])); + let input_batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int32Array::from_iter_values(0..10))], + ) + .unwrap(); + let source = MemorySourceConfig::try_new(&[vec![input_batch]], schema, None).unwrap(); + let input: Arc = Arc::new(DataSourceExec::new(Arc::new(source))); + let split = BatchSplitExec::new(input); + + let config = SessionConfig::new().with_batch_size(4); + let state = SessionStateBuilder::new().with_config(config).build(); + let context = SessionContext::new_with_state(state); + let mut stream = split.execute(0, context.task_ctx()).unwrap(); + + let mut batch_sizes = vec![]; + let mut values = vec![]; + while let Some(batch) = stream.next().await { + let batch = batch.unwrap(); + batch_sizes.push(batch.num_rows()); + let column = batch + .column(0) + .as_primitive::(); + values.extend(column.values().iter().copied()); + } + + assert_eq!(batch_sizes, vec![4, 4, 2]); + assert_eq!(values, (0..10).collect::>()); + } +} diff --git a/native/core/src/execution/operators/mod.rs b/native/core/src/execution/operators/mod.rs index b9b2b0fbd73..62c393e7f81 100644 --- a/native/core/src/execution/operators/mod.rs +++ b/native/core/src/execution/operators/mod.rs @@ -20,11 +20,13 @@ pub use crate::errors::ExecutionError; pub use aligned_stream_reader::*; +pub use batch_split::BatchSplitExec; pub use copy::*; pub use iceberg_scan::*; pub use scan::*; mod aligned_stream_reader; +mod batch_split; mod copy; mod expand; pub use expand::ExpandExec; diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index d109627e825..3818a53de4e 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -38,8 +38,8 @@ use crate::execution::{ expressions::list_positions::ListPositionsExpr, expressions::subquery::Subquery, operators::{ - ExecutionError, ExpandExec, ParquetCompression, ParquetWriterExec, SampleExec, ScanExec, - ShuffleScanExec, + BatchSplitExec, ExecutionError, ExpandExec, ParquetCompression, ParquetWriterExec, + SampleExec, ScanExec, ShuffleScanExec, }, planner::expression_registry::ExpressionRegistry, planner::operator_registry::OperatorRegistry, @@ -2116,11 +2116,20 @@ impl PhysicalPlanner { output_schema, unnest_options, )?); + // DataFusion 54.1.0's UnnestExec can emit more than the runtime batch size. + // Bound batches before downstream native projections until Comet upgrades to a + // DataFusion version containing https://github.com/apache/datafusion/pull/24384. + let bounded_unnest: Arc = + Arc::new(BatchSplitExec::new(unnest_exec)); Ok(( scans, shuffle_scans, - Arc::new(SparkPlan::new(spark_plan.plan_id, unnest_exec, vec![child])), + Arc::new(SparkPlan::new( + spark_plan.plan_id, + bounded_unnest, + vec![child], + )), )) } OpStruct::SortMergeJoin(join) => { diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index 3bf03732aa9..e6654411026 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -223,7 +223,9 @@ object CometStringReplace /** * `CometLiteral` encodes a string as `UTF8String.toString`, so only literals whose bytes * survive that round-trip can be sent natively. The size cap keeps a broadcast scalar under - * Arrow `Utf8`'s 32-bit offset limit at `spark.comet.batchSize` rows. + * Arrow `Utf8`'s 32-bit offset limit. Native operators feeding projections must emit batches no + * larger than `spark.comet.batchSize`; Comet's explode path enforces that with + * `BatchSplitExec`. */ private def isNativeSafeStringLiteral(v: UTF8String, allowEmpty: Boolean): Boolean = { if (v == null) { diff --git a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala index f3387dfa163..061e3298415 100644 --- a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala @@ -26,6 +26,7 @@ import org.apache.spark.{SparkConf, SparkEnv, TaskContext} import org.apache.spark.sql.CometTestBase import org.apache.spark.sql.api.java.UDF1 import org.apache.spark.sql.catalyst.expressions.{BoundReference, CreateArray, CreateMap, CreateNamedStruct, Expression, Literal, MapConcat} +import org.apache.spark.sql.comet.CometExplodeExec import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ @@ -483,6 +484,38 @@ class CometCodegenSuite } } + test("replace remains safe after native explode expands the batch") { + withTable("t") { + sql(""" + |CREATE TABLE t USING parquet AS + |SELECT IF(id % 2 = 0, 'a', 'b') AS s + |FROM range(8192) + |""".stripMargin) + + withSQLConf(CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") { + val df = sql(""" + |SELECT replace(e, 'notfound', repeat('x', 131072)) + |FROM t + |LATERAL VIEW explode(array(s, s)) a AS e + |""".stripMargin) + + val (_, cometPlan) = checkSparkAnswerAndOperator(df) + assert( + stripAQEPlan(cometPlan).exists(_.isInstanceOf[CometExplodeExec]), + s"expected native CometExplodeExec, got:\n$cometPlan") + + val rows = df.collect() + assert(rows.length == 16384) + assert(rows.forall(row => Set("a", "b").contains(row.getString(0)))) + + val explain = new ExtendedExplainInfo().generateExtendedInfo(cometPlan) + assert( + !explain.contains("JVM codegen dispatcher: replace"), + s"expected native replace after native explode, got:\n$explain") + } + } + } + test("codegen dispatch fallback reasons name the expression") { // Flag-off short-circuit tags the expression `: ` so distinct expressions // don't collapse in the `Set[String]` roll-up. From a96957cce86116427e6a056e878807c9c0bf7f05 Mon Sep 17 00:00:00 2001 From: Hung Date: Mon, 24 Aug 2026 16:25:29 +0800 Subject: [PATCH 5/5] fix: forward UnnestExec metrics through BatchSplitExec --- .../src/execution/operators/batch_split.rs | 24 +++++++++++++++++-- .../org/apache/comet/CometCodegenSuite.scala | 11 ++++++--- 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/native/core/src/execution/operators/batch_split.rs b/native/core/src/execution/operators/batch_split.rs index cb536370d52..4036ca25495 100644 --- a/native/core/src/execution/operators/batch_split.rs +++ b/native/core/src/execution/operators/batch_split.rs @@ -100,7 +100,18 @@ impl ExecutionPlan for BatchSplitExec { } fn metrics(&self) -> Option { - Some(self.metrics.clone_inner()) + let mut metrics = self.metrics.clone_inner(); + + // BatchSplitExec is a transparent execution wrapper. Preserve metrics produced by the + // wrapped UnnestExec so the Spark explode node can continue reporting its original SQL + // metrics. + if let Some(input_metrics) = self.input.metrics() { + for metric in input_metrics.iter() { + metrics.push(metric.to_owned()); + } + } + + Some(metrics) } fn maintains_input_order(&self) -> Vec { @@ -117,6 +128,7 @@ mod tests { use datafusion::datasource::memory::MemorySourceConfig; use datafusion::datasource::source::DataSourceExec; use datafusion::execution::SessionStateBuilder; + use datafusion::physical_plan::limit::GlobalLimitExec; use datafusion::prelude::{SessionConfig, SessionContext}; use futures::StreamExt; @@ -129,7 +141,8 @@ mod tests { ) .unwrap(); let source = MemorySourceConfig::try_new(&[vec![input_batch]], schema, None).unwrap(); - let input: Arc = Arc::new(DataSourceExec::new(Arc::new(source))); + let source: Arc = Arc::new(DataSourceExec::new(Arc::new(source))); + let input: Arc = Arc::new(GlobalLimitExec::new(source, 0, None)); let split = BatchSplitExec::new(input); let config = SessionConfig::new().with_batch_size(4); @@ -150,5 +163,12 @@ mod tests { assert_eq!(batch_sizes, vec![4, 4, 2]); assert_eq!(values, (0..10).collect::>()); + + let metrics = split.metrics().unwrap().aggregate_by_name(); + let output_rows = metrics + .iter() + .find(|metric| metric.value().name() == "output_rows") + .expect("wrapped input output_rows metric should be forwarded"); + assert_eq!(output_rows.value().as_usize(), 10); } } diff --git a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala index 061e3298415..d6c4849eaf6 100644 --- a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala @@ -500,14 +500,19 @@ class CometCodegenSuite |""".stripMargin) val (_, cometPlan) = checkSparkAnswerAndOperator(df) - assert( - stripAQEPlan(cometPlan).exists(_.isInstanceOf[CometExplodeExec]), - s"expected native CometExplodeExec, got:\n$cometPlan") val rows = df.collect() assert(rows.length == 16384) assert(rows.forall(row => Set("a", "b").contains(row.getString(0)))) + val explode = stripAQEPlan(df.queryExecution.executedPlan) + .collectFirst { case e: CometExplodeExec => e } + .getOrElse(fail("expected CometExplodeExec")) + assert(explode.metrics("input_rows").value == 8192) + assert(explode.metrics("output_rows").value == 16384) + assert(explode.metrics("input_batches").value > 0) + assert(explode.metrics("output_batches").value > 0) + val explain = new ExtendedExplainInfo().generateExtendedInfo(cometPlan) assert( !explain.contains("JVM codegen dispatcher: replace"),