andygrove commented on code in PR #5032:
URL: https://github.com/apache/datafusion-comet/pull/5032#discussion_r3692978646
##########
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:
You are right on both counts, and the mixin does the whole job. Done in
26994c919 — the entire change to this object is now `with
CodegenDispatchFallback` on the original declaration, exactly as you wrote it.
The `CometScalarFunction[StringTranslate]("translate")` base is restored, so
`"translate"` lives in one place again, and the opted-in `logWarning` at
`QueryPlanSerde.scala:841-846` comes back with it. The pre-existing
`Incompatible(Some(incompatReason))` was already the right classification, so
it stays untouched.
##########
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:
Agreed — reverted to the original three-way `getSupportLevel` plus the mixin
in 26994c919. The +30/-30 rewrite is now one mixin and a comment.
Confirming your reading of the routing, since it is the whole basis for the
change: `Unsupported` (complex field types) reaches the dispatcher plus
`logDebug` at `QueryPlanSerde.scala:819-831`, non-opted-in `Incompatible`
(#3232 types, and the `Incompatible()` catch-all) reaches the dispatcher plus
the `[COMET-INFO]` hint at `:848-860`, and opted-in `Incompatible` reaches
`handler.convert` — the native proto — at `:846`. So all three original
outcomes route as intended with no other change.
`getUnsupportedReasons` is restored, and the duplicated `nativeSupported`
predicate is gone along with the sync hazard between the two call sites. I also
dropped `CometCodegenDispatch` as the base: it hardcodes `getSupportLevel =
Compatible()` and `convert = emitJvmCodegenDispatch` for expressions with no
native path at all, so overriding `convert` to sometimes go native was working
against it.
##########
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:
Resolved by the redesign rather than by adding a `logDebug`: with
`CodegenDispatchFallback` this case lands in the framework's `Unsupported` arm,
which already logs at debug naming the disqualifying reason
(`QueryPlanSerde.scala:819-831`). So it is covered for free, as you noted.
--
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]
