viirya commented on code in PR #7251:
URL: https://github.com/apache/datafusion-comet/pull/7251#discussion_r4391835654
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1961,18 +1966,34 @@ object CometExplodeExec extends
CometOperatorSerde[GenerateExec] {
return None
}
- val isPosExplode = op.generator.nodeName.toLowerCase(Locale.ROOT) ==
"posexplode"
-
val explodeBuilder = OperatorOuterClass.Explode
.newBuilder()
.setChild(childExprProto.get)
.setOuter(op.outer)
.setPosition(isPosExplode)
+ .setInline(isInline)
.addAllProjectList(projectExprs.map(_.get).asJava)
Some(builder.setExplode(explodeBuilder).build())
}
+ /**
+ * `stack(n, v1, ..., vk)` emits `n` rows of `ceil(k / n)` columns, filled
row by row and padded
+ * with nulls. That is `inline` over an array holding one struct per output
row.
+ */
+ private def stackAsStructArray(stack: Stack): Expression = {
+ val numRows = stack.children.head.eval().asInstanceOf[Int]
+ val values = stack.children.tail
+ val fields = stack.elementSchema.fields.toSeq
+ CreateArray(Seq.tabulate(numRows) { row =>
Review Comment:
+1 on bounding this. I think the executor side is the stronger reason. The
native planner evaluates the synthesized `make_array` in the `ProjectionExec`
in front of `ExplodeExec`, so each input batch materializes `batch_size * n`
structs before `ExplodeStream` gets to chunk the output. Spark's `Stack.eval`
only holds the `n` rows of one input row at a time. Checking
`stack.supportCodegen` in `getSupportLevel` would keep the rewrite from being
built at all above Spark's 50-row limit, and that still covers every realistic
unpivot.
##########
native/core/src/execution/planner/explode.rs:
##########
@@ -162,11 +163,30 @@ impl OperatorBuilder for ExplodeBuilder {
}
};
- output_fields.push(Field::new(
- array_field.name(),
- element_type,
- true, // Element is nullable after unnesting
- ));
+ // `inline` flattens each struct element into one column per field.
Every field is
+ // nullable after unnesting: a null struct element, or the row
`inline_outer` emits for a
+ // NULL or empty array, yields a row of nulls.
+ let struct_column_indices = if explode.inline {
+ let DataType::Struct(struct_fields) = &element_type else {
+ return Err(ExecutionError::GeneralError(format!(
+ "Expected List of Struct for inline, got {:?}",
+ array_field.data_type()
+ )));
+ };
+ output_fields.extend(
Review Comment:
I'd lean toward falling back here rather than threading a row count through
`flatten_struct_cols`. Even with a row count, a zero-column batch still has to
leave the native plan and be exported to the JVM, and I'm not sure that path
handles a batch with no columns. Returning `Unsupported` from `getSupportLevel`
when the `inline` element struct has no fields keeps the previous Spark
behavior and costs nothing, since `array<struct<>>` carries no data. `stack`
cannot hit this because it always has at least one column. Would a plain
`SELECT inline(array(struct())) FROM t` also reach it? Spark accepts
`struct()`, so if Comet converts the zero-field `CreateNamedStruct` natively,
that would make a simpler regression test than the Iceberg case.
--
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]
