mbutrovich commented on code in PR #5032:
URL: https://github.com/apache/datafusion-comet/pull/5032#discussion_r3658190951
##########
spark/src/main/scala/org/apache/comet/serde/strings.scala:
##########
@@ -109,15 +109,38 @@ object CometOctetLength extends
CometScalarFunction[OctetLength]("octet_length")
}
}
-object CometStringTranslate extends
CometScalarFunction[StringTranslate]("translate") {
+object CometStringTranslate
+ extends CometExpressionSerde[StringTranslate]
+ with NativeOptInAvailable {
private val incompatReason =
"DataFusion's translate iterates over Unicode graphemes (Spark uses code
points) and" +
" substitutes U+0000 instead of treating it as a deletion sentinel"
override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason)
- override def getSupportLevel(expr: StringTranslate): SupportLevel =
Incompatible(
- Some(incompatReason))
+ override def getSupportLevel(expr: StringTranslate): SupportLevel =
+ if (!CometConf.isExprAllowIncompat(getExprConfigName(expr))) {
+ Compatible(nativeOptIn =
+
Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr)))))
+ } else {
+ Compatible()
+ }
+
+ override def convert(
+ expr: StringTranslate,
+ inputs: Seq[Attribute],
+ binding: Boolean): Option[Expr] = {
+ if (CometConf.isExprAllowIncompat(getExprConfigName(expr))) {
+ val childExprs = expr.children.map(exprToProtoInternal(_, inputs,
binding))
+ val optExpr = scalarFunctionExprToProto("translate", childExprs: _*)
+ optExprWithFallbackReason(optExpr, expr, expr.children: _*)
Review Comment:
First, lines 134-136 are a verbatim copy of `CometScalarFunction.convert`
(`spark/src/main/scala/org/apache/comet/serde/CometScalarFunction.scala:29-33`),
so the function name `"translate"` now lives in two places. Dropping the
`CometScalarFunction[StringTranslate]("translate")` base to hand-inline its
body is also a step away from the established idiom for this exact case, which
keeps the base and overrides `convert` to choose a branch: `CometInitCap`
(`strings.scala:162-174`), `CometStringReplace` (`:192-199`),
`CometCaseConversionBase` (`:67-78`).
Second, and more to the point: the whole `getSupportLevel` plus `convert`
branching structure is what `CodegenDispatchFallback` exists to express
declaratively. `QueryPlanSerde.exprToProtoInternal` already routes a
non-opted-in `Incompatible` result from such a serde through
`emitJvmCodegenDispatch` and falls back to Spark only when the dispatcher
refuses (`QueryPlanSerde.scala:838-870`, `dispatchIfFallback` at `:1109-1117`),
and `CodegenDispatchFallback extends NativeOptInAvailable` so `GenerateDocs`
renders the same compatible-by-default row (`GenerateDocs.scala:193`). So the
entire diff to this object should be one mixin on the original:
```scala
object CometStringTranslate
extends CometScalarFunction[StringTranslate]("translate")
with CodegenDispatchFallback {
private val incompatReason = ...
override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason)
override def getSupportLevel(expr: StringTranslate): SupportLevel =
Incompatible(Some(incompatReason))
}
```
That keeps the existing `Incompatible` declaration honest, keeps the
function name in one place, and gets the `logWarning` on the opted-in path
(`QueryPlanSerde.scala:841-846`) that the hand-rolled version drops.
##########
spark/src/main/scala/org/apache/comet/serde/structs.scala:
##########
@@ -259,50 +259,50 @@ object CometJsonToStructs extends
CometCodegenDispatch[JsonToStructs] with Nativ
}
}
-object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] {
+object CometStructsToCsv extends CometCodegenDispatch[StructsToCsv] with
NativeOptInAvailable {
private val incompatibleDataTypes = Seq(DateType, TimestampType,
TimestampNTZType, BinaryType)
override def getIncompatibleReasons(): Seq[String] = Seq(
"Date, Timestamp, TimestampNTZ, and Binary data types may produce
different results" +
" (https://github.com/apache/datafusion-comet/issues/3232)")
- override def getUnsupportedReasons(): Seq[String] = Seq(
- "Complex types (arrays, maps, structs) in the schema are not supported")
-
- override def getSupportLevel(expr: StructsToCsv): SupportLevel = {
+ // The native ToCsv path only supports non-complex, compatible field types.
Everything else
+ // (and the default, unless opted in) runs through the codegen dispatcher,
which is bit-exact.
+ private def nativeSupported(expr: StructsToCsv): Boolean = {
val dataTypes = expr.inputSchema.fields.map(_.dataType)
- val containsComplexType = dataTypes.exists(DataTypeSupport.isComplexType)
- if (containsComplexType) {
- return Unsupported(
- Some(
- s"The schema ${expr.inputSchema} is not supported because it
includes a complex type"))
- }
- val containsIncompatibleDataTypes =
dataTypes.exists(incompatibleDataTypes.contains)
- if (containsIncompatibleDataTypes) {
- return Incompatible(
- Some(
- s"The schema ${expr.inputSchema} is not supported because " +
- s"it includes a incompatible data types: $incompatibleDataTypes"))
- }
- // https://github.com/apache/datafusion-comet/issues/3232
- Incompatible()
+ !dataTypes.exists(DataTypeSupport.isComplexType) &&
+ !dataTypes.exists(incompatibleDataTypes.contains)
}
+ override def getSupportLevel(expr: StructsToCsv): SupportLevel =
+ if (!CometConf.isExprAllowIncompat(getExprConfigName(expr)) &&
nativeSupported(expr)) {
+ Compatible(nativeOptIn =
+
Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr)))))
+ } else {
+ Compatible()
+ }
+
override def convert(
expr: StructsToCsv,
inputs: Seq[Attribute],
binding: Boolean): Option[ExprOuterClass.Expr] = {
- for {
- childProto <- exprToProtoInternal(expr.child, inputs, binding)
- } yield {
- val optionsProto = options2Proto(expr.options, expr.timeZoneId)
- val toCsv = ExprOuterClass.ToCsv
- .newBuilder()
- .setChild(childProto)
- .setOptions(optionsProto)
- .build()
- ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build()
+ if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) &&
nativeSupported(expr)) {
Review Comment:
When `isExprAllowIncompat` is true and `nativeSupported` is false, the user
asked for the native path and silently gets the dispatcher with nothing
recorded. `CometStructsToJson` has the same hole so this is not new, but it
deserves a `logDebug` naming the disqualifying field type. Using
`CodegenDispatchFallback` gets this for free through the `Unsupported` arm.
##########
spark/src/main/scala/org/apache/comet/serde/structs.scala:
##########
@@ -259,50 +259,50 @@ object CometJsonToStructs extends
CometCodegenDispatch[JsonToStructs] with Nativ
}
}
-object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] {
+object CometStructsToCsv extends CometCodegenDispatch[StructsToCsv] with
NativeOptInAvailable {
private val incompatibleDataTypes = Seq(DateType, TimestampType,
TimestampNTZType, BinaryType)
override def getIncompatibleReasons(): Seq[String] = Seq(
"Date, Timestamp, TimestampNTZ, and Binary data types may produce
different results" +
" (https://github.com/apache/datafusion-comet/issues/3232)")
- override def getUnsupportedReasons(): Seq[String] = Seq(
- "Complex types (arrays, maps, structs) in the schema are not supported")
-
- override def getSupportLevel(expr: StructsToCsv): SupportLevel = {
+ // The native ToCsv path only supports non-complex, compatible field types.
Everything else
+ // (and the default, unless opted in) runs through the codegen dispatcher,
which is bit-exact.
+ private def nativeSupported(expr: StructsToCsv): Boolean = {
val dataTypes = expr.inputSchema.fields.map(_.dataType)
- val containsComplexType = dataTypes.exists(DataTypeSupport.isComplexType)
- if (containsComplexType) {
- return Unsupported(
- Some(
- s"The schema ${expr.inputSchema} is not supported because it
includes a complex type"))
- }
- val containsIncompatibleDataTypes =
dataTypes.exists(incompatibleDataTypes.contains)
- if (containsIncompatibleDataTypes) {
- return Incompatible(
- Some(
- s"The schema ${expr.inputSchema} is not supported because " +
- s"it includes a incompatible data types: $incompatibleDataTypes"))
- }
- // https://github.com/apache/datafusion-comet/issues/3232
- Incompatible()
+ !dataTypes.exists(DataTypeSupport.isComplexType) &&
+ !dataTypes.exists(incompatibleDataTypes.contains)
}
+ override def getSupportLevel(expr: StructsToCsv): SupportLevel =
+ if (!CometConf.isExprAllowIncompat(getExprConfigName(expr)) &&
nativeSupported(expr)) {
+ Compatible(nativeOptIn =
+
Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr)))))
+ } else {
+ Compatible()
+ }
+
override def convert(
expr: StructsToCsv,
inputs: Seq[Attribute],
binding: Boolean): Option[ExprOuterClass.Expr] = {
- for {
- childProto <- exprToProtoInternal(expr.child, inputs, binding)
- } yield {
- val optionsProto = options2Proto(expr.options, expr.timeZoneId)
- val toCsv = ExprOuterClass.ToCsv
- .newBuilder()
- .setChild(childProto)
- .setOptions(optionsProto)
- .build()
- ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build()
+ if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) &&
nativeSupported(expr)) {
+ for {
+ childProto <- exprToProtoInternal(expr.child, inputs, binding)
+ } yield {
+ val optionsProto = options2Proto(expr.options, expr.timeZoneId)
+ val toCsv = ExprOuterClass.ToCsv
+ .newBuilder()
+ .setChild(childProto)
+ .setOptions(optionsProto)
+ .build()
+ ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build()
+ }
+ } else {
+ // Default: route through the codegen dispatcher so Spark's own
doGenCode runs inside the
Review Comment:
The same argument applies here, and more strongly because the pre-existing
`getSupportLevel` already returned exactly the right classifications:
`Unsupported` for complex field types, `Incompatible` for the #3232 types,
`Incompatible()` otherwise. With `with CodegenDispatchFallback` on line 262
those three outcomes route as intended with no other change: `Unsupported` to
the dispatcher plus a `logDebug` (`QueryPlanSerde.scala:827-833`), non-opted-in
`Incompatible` to the dispatcher plus the `[COMET-INFO]` opt-in hint
(`:854-860`), opted-in `Incompatible` to the native proto (`:847`). That turns
a +30/-30 rewrite into a one-line change and removes the duplicated
`nativeSupported(expr)` test that currently has to be kept in sync between
`:279` and `:290`.
Deleting `getUnsupportedReasons` (was `structs.scala:270-271`) also loses
real information. The native path still cannot do complex types, and under
`NativeOptInAvailable` the documented surface for native-path limitations is
`getIncompatibleReasons`
(`spark/src/main/scala/org/apache/comet/serde/CometExpressionSerde.scala:52-63`).
Keeping `CodegenDispatchFallback` preserves both accessors as-is; if you keep
the rewrite instead, the complex-type limitation needs folding into
`getIncompatibleReasons` at `:266-268`, otherwise a user who sets
`allowIncompatible` has no way to find out why they did not get the native path.
##########
spark/src/test/resources/sql-tests/expressions/string/string_translate.sql:
##########
@@ -15,29 +15,29 @@
-- specific language governing permissions and limitations
-- under the License.
--- translate is gated as Incompatible by default. DataFusion's translate
iterates over Unicode
--- graphemes (Spark uses code points) and substitutes U+0000 instead of
treating it as a deletion
--- sentinel, so the native path silently diverges from Spark for
combining-mark inputs and for
--- to=NUL. These default-config tests assert that the expression falls back
cleanly to Spark.
--- See string_translate_enabled.sql for the opt-in native path.
+-- translate runs through the codegen dispatcher by default so results match
Spark exactly. The
+-- native path diverges from Spark (DataFusion iterates over Unicode graphemes
where Spark uses code
+-- points, and substitutes U+0000 instead of treating it as a deletion
sentinel), so it is opt-in
+-- via spark.comet.expression.StringTranslate.allowIncompatible. See
string_translate_enabled.sql
+-- for the opt-in native path.
statement
CREATE TABLE test_translate(s string, from_str string, to_str string) USING
parquet
statement
INSERT INTO test_translate VALUES ('hello', 'el', 'ip'), ('hello', 'aeiou',
'12345'), ('', 'a', 'b'), (NULL, 'a', 'b'), ('hello', '', ''), ('abc', 'abc',
'x')
Review Comment:
The four `query` blocks do assert native execution, so flipping them off
`expect_fallback` is real new coverage of the routing. What is missing is the
payload: the header comment at `:18-22` names two specific ways the native path
diverges, and the six rows on line 28 are ASCII-only and hit neither. The file
is now identical to `string_translate_enabled.sql` in CREATE TABLE, INSERT rows
and queries, differing only in the config header, so nothing distinguishes the
dispatcher from the native path here. Please add the two rows that are only
assertable now that the default is bit-exact: a combining-mark input for the
grapheme versus code-point difference, and a `to` argument containing U+0000
for the deletion-sentinel difference. To keep the fixture ASCII, build them
with `decode(X'65CC81', 'UTF-8')` and `decode(X'00', 'UTF-8')` rather than
embedding the characters. Those two rows are the regression test for this PR.
##########
spark/src/main/scala/org/apache/comet/serde/structs.scala:
##########
@@ -259,50 +259,50 @@ object CometJsonToStructs extends
CometCodegenDispatch[JsonToStructs] with Nativ
}
}
-object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] {
+object CometStructsToCsv extends CometCodegenDispatch[StructsToCsv] with
NativeOptInAvailable {
private val incompatibleDataTypes = Seq(DateType, TimestampType,
TimestampNTZType, BinaryType)
override def getIncompatibleReasons(): Seq[String] = Seq(
"Date, Timestamp, TimestampNTZ, and Binary data types may produce
different results" +
" (https://github.com/apache/datafusion-comet/issues/3232)")
- override def getUnsupportedReasons(): Seq[String] = Seq(
- "Complex types (arrays, maps, structs) in the schema are not supported")
-
- override def getSupportLevel(expr: StructsToCsv): SupportLevel = {
+ // The native ToCsv path only supports non-complex, compatible field types.
Everything else
+ // (and the default, unless opted in) runs through the codegen dispatcher,
which is bit-exact.
+ private def nativeSupported(expr: StructsToCsv): Boolean = {
val dataTypes = expr.inputSchema.fields.map(_.dataType)
- val containsComplexType = dataTypes.exists(DataTypeSupport.isComplexType)
- if (containsComplexType) {
- return Unsupported(
- Some(
- s"The schema ${expr.inputSchema} is not supported because it
includes a complex type"))
- }
- val containsIncompatibleDataTypes =
dataTypes.exists(incompatibleDataTypes.contains)
- if (containsIncompatibleDataTypes) {
- return Incompatible(
- Some(
- s"The schema ${expr.inputSchema} is not supported because " +
- s"it includes a incompatible data types: $incompatibleDataTypes"))
- }
- // https://github.com/apache/datafusion-comet/issues/3232
- Incompatible()
+ !dataTypes.exists(DataTypeSupport.isComplexType) &&
+ !dataTypes.exists(incompatibleDataTypes.contains)
}
Review Comment:
`nativeSupported` is the new branch predicate driving both `getSupportLevel`
(`:279`) and `convert` (`:290`), and no test takes it false. The two cases it
excludes are the reason the change exists:
Date, Timestamp, TimestampNTZ and Binary fields were `Incompatible` before
this PR and fell the operator back to Spark. That is #3232.
Complex field types were `Unsupported` before and also fell back. They are
legal `to_csv` input: `StructsToCsv.checkInputDataTypes` accepts arrays, maps
and nested structs and rejects only Variant (Spark
`csvExpressions.scala:240-261` on master and branch-4.0; branch-3.5 has no gate
at all). Nothing tests that the kernel can hand `converter()` a nested
`InternalRow`, which is the new and untried part of this path.
The cheapest place to cover the first group is the existing suite rather
than a new fixture. `CometCsvExpressionSuite`'s test named "to_csv - default
options"
(`spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala:36`) is
actually run under `allowIncompatible=true` (`:50`), and it selects c0-c5, c7,
c8, c9 and c12 out of the 14-column fuzz schema, skipping exactly c6 Double,
c10 Timestamp, c11 TimestampNTZ and c13 Binary
(`FuzzDataGenerator.scala:303-317`). Those are the columns the native path
cannot match Spark on. After this PR the dispatcher matches them by
construction, so that test can drop the `allowIncompatible` wrapper and add
those four columns, which is what would actually justify the name it already
has.
##########
spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql:
##########
@@ -0,0 +1,41 @@
+-- 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.
+
+-- to_csv runs through the codegen dispatcher by default so results match
Spark exactly, including
+-- quoting and escaping. The native path is opt-in via
+-- spark.comet.expression.StructsToCsv.allowIncompatible.
+
+statement
+CREATE TABLE test_to_csv(a int, b string, c double) USING parquet
+
+statement
+INSERT INTO test_to_csv VALUES
+ (1, 'x', 2.5),
+ (-3, 'hello,world', 0.0),
+ (0, 'has "quote"', -1.5),
+ (NULL, NULL, NULL),
+ (7, '', 3.0)
+
+-- column struct: values with delimiters and quotes exercise Spark's CSV
quoting rules
+query
+SELECT to_csv(named_struct('a', a, 'b', b, 'c', c)) FROM test_to_csv
Review Comment:
`options` is the main axis of behavior difference for this expression and
nothing in the new fixture varies it. `CometCsvExpressionSuite:71-161` covers
delimiter, escape, quoteAll and nullValue but only under
`allowIncompatible=true`, so the dispatcher path has no options coverage
anywhere. `to_csv(s, map('sep', ';'))` plus a `timestampFormat` over a
timestamp field would cover both this and the #3232 types in one query.
--
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]
