sunchao commented on code in PR #7251:
URL: https://github.com/apache/datafusion-comet/pull/7251#discussion_r4391396936
##########
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:
Confirmed at `f5cfc51e29`. A bounded Catalyst probe using
`stackAsStructArray` extracted unchanged from this head expands `Stack(10000,
42)` from 3 expression nodes to 30,001 nodes, including 10,000 structs. At 51
rows the rewrite already proceeds although Spark's `supportCodegen` is false.
The supported Spark releases all cap stack code generation at 50 rows.
This allocation happens during `convert`, even before the `childOp.isEmpty`
check, so planning/EXPLAIN for a large valid row count can exhaust driver heap
without executing the query. Could we keep larger stacks on Spark before
entering this rewrite and add a planning regression case above the bound? This
is a method-level reproduction, not a full Comet execution test.
##########
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:
Confirmed the native failure at `f5cfc51e29` with the extracted
`flatten_struct_cols` function and Arrow 59.3.1 from this PR's lockfile. A
`StructArray::new_empty_fields(2, None)` with no carried columns fails with
`Invalid argument error: must either specify a row count or at least one
column`. The control using `RecordBatchOptions::with_row_count(Some(2))`
returns the expected two rows.
The Iceberg case is reachable: `CometScanRule` explicitly exempts metadata
columns such as empty `_partition` structs from schema validation, and the
attribute/array serdes accept that type. Could we preserve the unnested row
count when constructing the batch and add a zero-field `inline`/`inline_outer`
regression? I verified the Arrow failure and traced the serde path, but did not
run an Iceberg integration test.
--
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]
