viirya commented on code in PR #7251:
URL: https://github.com/apache/datafusion-comet/pull/7251#discussion_r4385722617
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1920,33 +1918,43 @@ 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"))
+ Unsupported(Some(s"Unsupported generator: ${other.nodeName}"))
}
}
override def convert(
op: GenerateExec,
builder: Operator.Builder,
childOp: OperatorOuterClass.Operator*):
Option[OperatorOuterClass.Operator] = {
- val childExpr = op.generator.children.head
+ val (childExpr, isPosExplode, isInline) = op.generator match {
+ case inline: Inline => (inline.child, false, true)
+ case stack: Stack => (stackAsStructArray(stack), false, true)
+ case explode: ExplodeBase => (explode.child, explode.position, false)
+ case other => throw new IllegalStateException(s"Unsupported generator:
${other.nodeName}")
+ }
val childExprProto = exprToProto(childExpr, op.child.output)
if (childExprProto.isEmpty) {
+ // `stack` builds its array of structs here rather than taking it from
the plan, so carry any
+ // reason recorded on the built nodes to the generator, where the
operator collects it.
+ liftFallbackReasons(childExpr, op.generator)
Review Comment:
I'm not sure this lift does what the comment describes. `stackAsStructArray`
puts the original value expressions into the synthesized tree as the same
instances, so a reason recorded on the `plus_one` UDF in the new test is
already on a node under `Stack.children`. `rollUpFallbackReasons` walks
`op.expressions` and should find it without the lift. Does the
`CometGenerateExecSuite` test still pass with this line removed?
What the lift adds on top is the generic `unsupported arguments for
CreateNamedStruct` and `unsupported arguments for CreateArray` reasons from
nodes the user never wrote. Those will show up in the extended explain for a
plain `stack(...)`. It also runs for `explode` and `inline`, whose child is
already in the plan. If there is a case that needs the lift, could you scope it
to `Stack`, add a test that fails without it, and keep the synthesized
wrappers' generic reasons out of what gets reported? Otherwise dropping it
seems simpler.
##########
native/operators/src/explode.rs:
##########
@@ -1872,4 +1877,21 @@ mod tests {
);
assert_output_lens(with_nulls.slice(1, 3), [&[0, 0, 3], &[1, 0, 3],
&[1, 1, 3]]);
}
+
+ #[test]
+ fn flatten_struct_cols_nulls_fields_under_a_null_struct() {
Review Comment:
The second commit dropped the sliced-input case and the non-nullable field
from this test, but the PR description still says it covers a sliced input. I
think the slice case is worth keeping. The contiguous fast path in
`unnest_list_array` returns `values.slice(first, capacity)`, so for every chunk
after the first, `flatten_struct_cols` receives a `StructArray` with a non-zero
offset. That is the normal production input for `inline`. The SQL tests fit in
one batch and never reach it, and `struct_nulls.rs` has no slice case either.
Could you restore the `structs.slice(..)` assertion and the non-nullable field?
--
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]
