sunchao commented on code in PR #6107:
URL: https://github.com/apache/datafusion-comet/pull/6107#discussion_r4417380935
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1928,12 +1928,8 @@ object CometExplodeExec extends
CometOperatorSerde[GenerateExec] {
return Unsupported(Some(s"Unsupported generator:
${op.generator.nodeName}"))
}
op.generator.children.head.dataType match {
- case _: ArrayType =>
+ case _: ArrayType | _: MapType =>
Review Comment:
Confirmed fixed at `259c891c2`. I reran the original case through
`MapEntriesExpr` and `ExplodeExec` with one 2,049-entry map, a 1 MiB carried
string, and a batch size of 256. It returned eight batches of 256 and one final
row, with every key, value and string length intact. The Spark execution suite
also passed the wide-map aggregate tests for all four generator variants.
Resolving this overflow report.
##########
native/operators/src/explode.rs:
##########
@@ -346,6 +356,82 @@ impl PendingInput {
.as_ref()
.map(|lens| lens.slice(self.row_offset, rows))
}
+
+ fn next_chunk(
+ &mut self,
+ batch_size: usize,
+ list_columns: &[ListUnnest],
+ ) -> Result<(RecordBatch, Option<Int64Array>)> {
+ if let Some(limit) = self.split_limit {
+ let row_length =
self.output_lens.as_ref().unwrap().value(self.row_offset) as usize;
+ if row_length > limit {
+ let length = limit.min(row_length - self.element_offset);
+ let row = self.batch.slice(self.row_offset, 1);
+ let mut columns = row.columns().to_vec();
+ for unnest in list_columns {
+ let list =
columns[unnest.index_in_input_schema].as_list::<i32>();
+ let DataType::List(field) = list.data_type() else {
+ unreachable!();
+ };
+ let available = if list.is_null(0) {
+ 0
+ } else {
+ list.value_length(0) as usize
+ };
+ let skip = self.element_offset.min(available);
+ let start = list.value_offsets()[0] + skip as i32;
+ let end = start + length.min(available - skip) as i32;
+ // Narrow the entries and parallel positions together,
preserving their
+ // original values and sharing their buffers. Shorter
lists still pad.
+ columns[unnest.index_in_input_schema] =
Arc::new(ListArray::new(
+ Arc::clone(field),
+ OffsetBuffer::new(vec![start, end].into()),
+ Arc::clone(list.values()),
+ list.nulls().cloned(),
+ ));
+ }
+ self.element_offset += length;
+ if self.element_offset == row_length {
+ self.row_offset += 1;
+ self.element_offset = 0;
+ }
+ return Ok((
+ RecordBatch::try_new(row.schema(), columns)?,
+ Some(Int64Array::from(vec![length as i64])),
+ ));
+ }
+ }
+ let rows = self.next_chunk_rows(batch_size);
+ let chunk = self.batch.slice(self.row_offset, rows);
+ let lengths = self.chunk_lengths(rows);
+ self.row_offset += rows;
+ Ok((chunk, lengths))
+ }
+}
+
+/// If every row repeats at most R times, no copied buffer or child offset
span exceeds R
+/// times its entire source span. Count elements too: List<Null> has no child
value buffer.
+fn repetition_limit(batch: &RecordBatch, list_columns: &[ListUnnest]) -> usize
{
+ fn max_span(data: &ArrayData) -> usize {
+ std::iter::once(data.len())
+ .chain(data.buffers().iter().map(|buffer| buffer.len()))
+ .chain(data.child_data().iter().map(max_span))
+ .max()
+ .unwrap_or(0)
+ }
+ let span = batch
+ .columns()
+ .iter()
+ .enumerate()
+ .filter(|(index, _)| {
+ !list_columns
+ .iter()
+ .any(|list| list.index_in_input_schema == *index)
+ })
+ .map(|(_, column)| max_span(&column.to_data()))
+ .max()
+ .unwrap_or(0);
+ (i32::MAX as usize / span.max(1)).max(1)
Review Comment:
I reproduced the batch fragmentation on the existing array path at
`259c891c2`. With a batch size of 8,192, the input was 33 rows. One carried a
16 MiB string and an array with one element. Each of the other 32 carried a
one-byte string and an array with 300 elements. The base (`330b29cf`) and
previous head (`8f93e97a`) emitted **2 batches**. This head emitted **97**,
with exactly the same 9,601 rows and 16,786,816 payload bytes. The wide string
is emitted only once, but its size caps every other row at 127 repetitions, so
each 300-element row gets three builds.
A second operator probe carried a single one-byte string sliced from a
larger Arrow buffer. With 64 MiB of unselected bytes behind that slice,
exploding 1,024 elements changed from 1 batch to 34 for just 1 KiB of logical
string output. Across 200 runs the previous head took about 4.9 ms total and
this head took 48.5 ms. The no-padding control stayed at one batch on both
heads. These are isolated operator timings with unoptimized dependencies, not
Spark benchmark numbers.
Could the split decision use the bytes and child-offset spans actually
repeated in the candidate chunk? That would retain the wide-map fix without
penalizing small array rows for another row's payload or a slice's unused
backing buffer. The 33-row input would make a regression test for both output
and batching.
--
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]
