viirya commented on code in PR #7234:
URL: https://github.com/apache/datafusion-comet/pull/7234#discussion_r4381374872
##########
native/shuffle/src/writers/shuffle_block_writer.rs:
##########
@@ -154,6 +192,17 @@ impl ShuffleBlockWriter {
}
}
+ /// True when the schema contains a view type at any depth; see
[`contains_view_type`].
+ ///
+ /// Arrow's IPC writer serializes every data buffer a view array
references, whatever the
+ /// views actually point at. A batch gathered by `interleave` from many
buffered batches
Review Comment:
Small accuracy point on this doc. `interleave_views` only carries over the
data buffers that the selected views actually reference. It doesn't carry all
of the source batches' buffers. The concern is still real when the strings are
long and come from many source batches. But "shares all of their data buffers"
overstates it. Could this say "every data buffer the selected views reference"?
##########
native/shuffle/src/writers/buf_batch_writer.rs:
##########
@@ -46,32 +56,68 @@ pub(crate) struct BufBatchWriter<S:
Borrow<ShuffleBlockWriter>, W: Write> {
writer: W,
buffer_max_size: usize,
/// Coalesces small batches into target_batch_size before serialization.
- /// Lazily initialized on first write to capture the schema.
+ /// Lazily initialized on first write to capture the schema. Never set in
passthrough mode.
coalescer: Option<BatchCoalescer>,
- /// Target batch size for coalescing
- batch_size: usize,
+ /// Target batch size for coalescing; `None` selects passthrough mode.
+ coalesce_batch_size: Option<usize>,
/// Address of the scratch `Vec` seen on first use; every later call must
pass the same
/// one, or unflushed bytes in the other buffer would be silently
abandoned.
#[cfg(debug_assertions)]
scratch_addr: Option<usize>,
- /// Running total of bytes serialized through this writer, used to report
spilled bytes when
- /// the underlying writer does not implement [`Seek`] (e.g. a `Box<dyn
SpillWriter>`).
+ /// Running total of bytes serialized through this writer, used to report
spilled bytes and
+ /// to track output offsets without a `Seek` on the underlying writer.
total_bytes_written: u64,
}
impl<S: Borrow<ShuffleBlockWriter>, W: Write> BufBatchWriter<S, W> {
+ /// A coalescing writer; see the type-level docs for when to use which
mode.
pub(crate) fn new(
shuffle_block_writer: S,
writer: W,
buffer_max_size: usize,
batch_size: usize,
+ ) -> Self {
+ Self::with_mode(
+ shuffle_block_writer,
+ writer,
+ buffer_max_size,
+ Some(batch_size),
+ )
+ }
+
+ /// A writer for one output partition of a multi-partition shuffle:
passthrough, except for
+ /// schemas with view types, which still need the coalescer to compact
sparse views (see
+ /// [`ShuffleBlockWriter::has_view_types`]).
+ pub(crate) fn new_for_partition(
Review Comment:
I'm not sure the view-type exception buys much here. Three things make me
doubt it:
- Even on the coalescing path, full `batch_size` chunks skip the coalescer
because of `with_biggest_coalesce_batch_size(Some(batch_size - 1))`, as the
comment in `write` says. So for a view schema, only the one tail per partition
per cycle is ever compacted. Every full chunk is written with whatever buffers
`interleave` gave it, both before and after this PR.
- `BatchCoalescer` only gc's top-level `Utf8View`/`BinaryView` columns,
through `InProgressByteViewArray`. Everything else goes through
`GenericInProgressArray`, whose `concat` returns the input slice unchanged when
there is only one input. That includes a view nested in a struct or list, a
dictionary of views, and `ListView`/`LargeListView`. Yet `has_view_types`
matches all of those, so for them we keep the coalescer and get nothing for it.
- As far as I can tell, view types can't reach the shuffle writer today.
`planner.rs` casts `Utf8View` results back to `Utf8` with the note that Comet
does not support view types yet.
Would it be simpler to make the per-partition writer always pass through?
That would mean dropping `has_view_types`, the `coalesce_batch_size` branch,
and `for_partition_coalesces_only_view_schemas`, and leaving view handling to
the existing round-robin planner check. If you'd rather keep the guard as
future-proofing, could the docs say it only helps the tail of top-level view
columns?
##########
native/shuffle/src/writers/local/spill.rs:
##########
@@ -334,7 +321,8 @@ mod tests {
let block_writer =
ShuffleBlockWriter::try_new(batch.schema_ref().as_ref(),
CompressionCodec::None)
.unwrap();
- // batch_size below the row count so a write serializes into the
scratch
+ // batch_size below the row count so a view-schema write, which
coalesces, serializes into
Review Comment:
This helper builds an `Int64` schema, so there's no view-schema write here.
With passthrough, every write serializes into the scratch whatever the batch
size. Could the comment be removed or reworded? Otherwise it reads as if these
tests exercise the coalescing path. It also means `batch_size = 10` no longer
matters to these tests.
--
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]
