sunchao commented on code in PR #5725:
URL: https://github.com/apache/datafusion-comet/pull/5725#discussion_r3942316460
##########
spark/src/main/scala/org/apache/comet/serde/strings.scala:
##########
@@ -287,44 +287,25 @@ object CometConcat
}
object CometConcatWs extends CometExpressionSerde[ConcatWs] with
CodegenDispatchFallback {
-
- // Spark's ConcatWs accepts `array<string>` arguments after the separator
and flattens their
- // elements into the list of strings to join (null elements are skipped,
like null strings).
- // DataFusion's `concat_ws` accepts only string arguments and fails at
execution time when it is
- // handed a list, so these calls have no native path. Because this serde
mixes in
- // `CodegenDispatchFallback`, they run through the JVM codegen dispatcher
(Spark's own
- // `ConcatWs.doGenCode` inside the Comet pipeline) instead, and fall back to
Spark only when the
- // dispatcher is disabled. See
https://github.com/apache/datafusion-comet/issues/5675.
- private val arrayArgumentReason =
- "`concat_ws` with `array<string>` arguments: Spark flattens the array
elements into the " +
- "strings to join, which DataFusion's `concat_ws` does not support " +
- "(https://github.com/apache/datafusion-comet/issues/5675)"
-
- override def getUnsupportedReasons(): Seq[String] = Seq(arrayArgumentReason)
+ override def getUnsupportedReasons(): Seq[String] = Seq("all arguments are
foldable")
override def getSupportLevel(expr: ConcatWs): SupportLevel =
expr.children.headOption match {
- // A NULL separator converts directly to a NULL result, so it stays
supported.
case Some(Literal(null, _)) => Compatible()
- case _ if expr.children.exists(_.dataType.isInstanceOf[ArrayType]) =>
- Unsupported(Some(arrayArgumentReason))
- // Decline all-literal args so that Spark's ConstantFolding handles them
(it normally folds
- // them before they reach Comet). With the dispatcher enabled they run
through Spark's own
- // generated code in-pipeline; otherwise the projection falls back to
Spark.
case _ if expr.children.forall(_.foldable) =>
+ // The upstream kernel expands scalar arguments to one row but iterates
over the batch
+ // length. Keep codegen dispatch for this shape until it supports
all-scalar batches.
Unsupported(Some("all arguments are foldable"))
Review Comment:
### Correctness
**[P2] Handle runtime scalars as well as foldable expressions**
A non-foldable child can still evaluate to a scalar. For example, with a
multi-row Parquet `fact` table and a string `lookup.s`, `SELECT id,
concat_ws(',', (SELECT max(s) FROM lookup)) FROM fact` passes this guard:
Spark's `ScalarSubquery` is non-foldable, while Comet's `Subquery::evaluate`
returns `ColumnarValue::Scalar(Utf8(...))`. All arguments to this call are
therefore scalars at runtime. In locked DataFusion Spark 55, `values_to_arrays`
expands them to length 1, but `spark_concat_ws` loops over
`ScalarFunctionArgs.number_rows`, which is the surrounding batch length. The
separator access at row 1 then exceeds the one-element array. The previous
DataFusion string `concat_ws` handles this shape by returning a scalar.
Please handle all-scalar inputs at the native invocation boundary, or use a
corrected kernel, and add a multi-row regression containing a scalar subquery.
The current column-input tests and `.foldable` check do not cover this
contract. This is confirmed by tracing the exact sources; I have not run the
example locally.
##########
native/spark-expr/src/comet_scalar_funcs.rs:
##########
@@ -118,6 +118,9 @@ pub fn create_comet_physical_fun_with_eval_mode(
) -> Result<Arc<ScalarUDF>, DataFusionError> {
let fail_on_error = fail_on_error.unwrap_or(false);
match fun_name {
+ "concat_ws" => Ok(Arc::new(ScalarUDF::new_from_impl(
+
datafusion_spark::function::string::concat_ws::SparkConcatWs::new(),
+ ))),
Review Comment:
### Performance
**[P2] Benchmark both the new array path and the replaced string path**
This arm changes every native `concat_ws` call, including previously native
string-only calls. The new kernel broadcasts scalars through
`values_to_arrays`, creates list-row slices, and reserves a fixed 16 bytes per
output row; the previous string kernel retains scalar references and uses an
input-derived output capacity. There are no benchmark results in the
PR/discussion, the benchmark check is skipped, and the existing
`CometStringExpressionBenchmark` only contains a string-column case.
Please add an array/mixed-argument benchmark and publish matched
Comet-versus-Spark codegen results before enabling this path, including
realistic array lengths, nulls, column/literal separators and short/long
strings. Also run the existing string-only case before and after the factory
change to check for regression. Verify the intended native/dispatch/Spark plans
for each measurement. The new correctness regressions do not establish this
hot-path performance tradeoff.
--
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]
