rluvaton commented on code in PR #23407:
URL: https://github.com/apache/datafusion/pull/23407#discussion_r3564855160
##########
datafusion/physical-plan/src/sorts/merge.rs:
##########
@@ -171,24 +152,40 @@ impl<C: CursorValues> SortPreservingMergeStream<C> {
in_progress: BatchBuilder::new(schema, stream_count, batch_size,
reservation),
streams,
metrics,
- done: false,
- drain_in_progress_on_done: false,
cursors: (0..stream_count).map(|_| None).collect(),
prev_cursors: (0..stream_count).map(|_| None).collect(),
round_robin_tie_breaker_mode: false,
num_of_polled_with_same_value: vec![0; stream_count],
current_reset_epoch: 0,
poll_reset_epochs: vec![0; stream_count],
loser_tree: vec![],
- loser_tree_adjusted: false,
batch_size,
fetch,
produced: 0,
- uninitiated_partitions: (0..stream_count).collect(),
enable_round_robin_tie_breaker,
}
}
+ pub(crate) fn into_stream(self) -> SendableRecordBatchStream
+ where
+ C: 'static,
+ {
+ let schema_clone = Arc::clone(self.in_progress.schema());
+
+ let cloned_metrics = self.metrics.clone();
+
+ let mut this = self;
+ let stream = Gen::new(|co| async move {
Review Comment:
> [...] across is that this looks much nicer imo at the call site in
SortPreservingMergeStream
The only difference I see is no need to do the `if let Err(...` the rest
looks the same
>
> The main downside I see is that without a proc macro it's really hard to
make this entirely abuse proof. I started with the thread local approach used
by the `async-stream` macro, but that's unsound when the emitter is exposed. I
went for the `Arc<Mutex>` then rather than the `SmallVec` that
`async-fn-stream` uses.
>
> Doesn't have to be this implementation, but perhaps something like this
could be a "have your cake and eat it" solution?
This is possible, but aside from not adding a dependency I don't see much
value, we can always start with the dep and move to in house impl (or vice
versa) but we can first see if the dependency is enough for us and not needing
to add more maintenance burden
--
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]
