andygrove commented on code in PR #7253:
URL: https://github.com/apache/datafusion-comet/pull/7253#discussion_r4456348569
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1920,59 +1918,102 @@ object CometExplodeExec extends
CometOperatorSerde[GenerateExec] {
if (!op.generator.deterministic) {
return Unsupported(Some("Only deterministic generators are supported"))
}
- if (op.generator.children.length != 1) {
- return Unsupported(Some("generators with multiple inputs are not
supported"))
- }
- val nodeName = op.generator.nodeName.toLowerCase(Locale.ROOT)
- if (nodeName != "explode" && nodeName != "posexplode") {
- return Unsupported(Some(s"Unsupported generator:
${op.generator.nodeName}"))
- }
- op.generator.children.head.dataType match {
- case _: ArrayType =>
+ op.generator match {
+ // Spark only accepts an array of structs for `inline`, and `stack` is
planned as `inline`;
+ // see `stackAsStructArray`.
+ case _: Inline | _: Stack =>
Compatible()
- case _: MapType =>
- // TODO add support for map types
- // https://github.com/apache/datafusion-comet/issues/2837
- Unsupported(Some("Comet only supports explode/explode_outer for
arrays, not maps"))
+ case explode: ExplodeBase =>
+ explode.child.dataType match {
+ case _: ArrayType =>
+ Compatible()
+ case _: MapType =>
+ // TODO add support for map types
+ // https://github.com/apache/datafusion-comet/issues/2837
+ Unsupported(Some("Comet only supports explode/explode_outer for
arrays, not maps"))
+ case other =>
+ Unsupported(Some(s"Unsupported data type: $other"))
+ }
case other =>
- Unsupported(Some(s"Unsupported data type: $other"))
+ CometGeneratorUDF.declineReason(other) match {
+ case Some(reason) => Unsupported(Some(reason))
+ case None => Compatible()
+ }
}
}
override def convert(
op: GenerateExec,
builder: Operator.Builder,
childOp: OperatorOuterClass.Operator*):
Option[OperatorOuterClass.Operator] = {
- val childExpr = op.generator.children.head
- val childExprProto = exprToProto(childExpr, op.child.output)
+ val explodeBuilder =
OperatorOuterClass.Explode.newBuilder().setOuter(op.outer)
+ val childExprProto = op.generator match {
+ case inline: Inline =>
+ explodeBuilder.setInline(true)
+ exprToProto(inline.child, op.child.output)
+ case stack: Stack =>
+ explodeBuilder.setInline(true)
+ val array = stackAsStructArray(stack)
+ exprToProto(array, op.child.output).orElse {
+ // The array of structs is built here rather than taken from the
plan, so carry any
+ // reason recorded on the built nodes to the generator, where the
operator collects it.
+ liftFallbackReasons(array, op.generator)
+ None
+ }
+ case explode: ExplodeBase =>
+ explodeBuilder.setPosition(explode.position)
+ exprToProto(explode.child, op.child.output)
+ case generator =>
Review Comment:
Done in 4c335a8a5. `ReplicateRows` now lowers to
`inline(array_repeat(named_struct(...), n))`, and the JVM generator remains the
fallback if that does not convert. Checked against `ReplicateRows.eval`: a NULL
count unboxes to 0 rows and `array_repeat` gives a NULL array, which `inline`
drops; a negative count gives an empty range and an empty array. The `Long`
count is clamped to `Int.MaxValue` instead of wrapped by the cast. The
`except_all.sql` types above pass natively, and `EXCEPT ALL replicates rows
with array_repeat, not a JVM generator` asserts the plan holds no
`JvmScalarUdf`. A new test also covers counts of zero, one and many, for both
`EXCEPT ALL` and `INTERSECT ALL`.
--
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]
