adriangb commented on code in PR #23522:
URL: https://github.com/apache/datafusion/pull/23522#discussion_r3591053454
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -168,6 +168,32 @@ struct OutputChannel {
shared_coalescer: Option<SharedCoalescer>,
}
+/// The set of spill-pool writers for a single output partition, before they
are handed to the
+/// per-input tasks. The variant encodes the repartition mode so the wrong
writer topology cannot
+/// be constructed for a given mode.
+enum PartitionSpillWriters {
+ /// `preserve_order`: one single-producer FIFO writer per input partition.
Each is `take`n
+ /// exactly once (moved into the matching input task), so the pool always
has one writer.
+ PerInput(Vec<Option<SpillPoolWriter>>),
+ /// Non-preserve-order: one shared writer, cloned into every input task.
+ Shared(SharedSpillPoolWriter),
+}
+
+impl PartitionSpillWriters {
+ /// Hand out the writer for input partition `input`.
+ ///
+ /// In `PerInput` mode this moves the dedicated writer out (it must only
be requested once per
+ /// input); in `Shared` mode it clones the shared writer.
+ fn take_for_input(&mut self, input: usize) -> SpillPoolWriter {
+ match self {
+ PartitionSpillWriters::PerInput(writers) => writers[input]
+ .take()
+ .expect("spill writer for input partition requested more than
once"),
Review Comment:
It looks like this gets called from a fallible function, we could (not
saying we should) make this error instead of panic.
##########
datafusion/physical-plan/src/spill/spill_pool.rs:
##########
@@ -476,6 +469,42 @@ pub fn channel(
(writer, Box::pin(reader))
}
+/// Creates a paired writer and reader for a spill pool with MPSC
(multi-producer,
+/// single-consumer) semantics. See [`channel`] for the general architecture
description
+/// of the spill pool.
+///
+/// Additional writers can be created by cloning the returned
[`SharedSpillPoolWriter`].
+///
+/// In contrast to [`channel`], this implementation provides no guarantees
regarding
+/// the read order of the returned [`SendableRecordBatchStream`].
+///
+/// If you need strict end-to-end FIFO (a single writer whose batches are read
back in exact
+/// write order), use [`channel`] instead.
+///
+/// # File Management
+///
+/// The shared channel uses the same size-based rotation trigger as the
[single producer channel](channel).
+/// All writers share the same pool of write files and coordinate file
rotation. The number of open
+/// files is kept as small as possible. When more writes occur concurrently
than there are open write
+/// files an additional file will be opened to write to. This prevents
multiple writers from blocking
+/// each other.
+///
+/// When the last writer clone is dropped, it finalizes any remaining open
write files so that all
+/// written data can be accessed by the reader.
+///
+/// # Returns
+///
+/// A tuple of `(SharedSpillPoolWriter, SendableRecordBatchStream)` that share
the same
+/// underlying pool. The reader is returned as a stream for immediate use with
+/// async stream combinators. The writer can be cloned to create additional
writers.
+pub fn shared_channel(
Review Comment:
could this be `pub(crate)`?
--
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]
