alamb commented on code in PR #23407:
URL: https://github.com/apache/datafusion/pull/23407#discussion_r3570132228
##########
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:
I left a comment below - I don't think adding this new depedency is a good
idea. We already striggle with the transitive dependency thing in downstream
various projects
##########
Cargo.toml:
##########
@@ -159,6 +159,7 @@ datafusion-session = { path = "datafusion/session", version
= "54.0.0" }
datafusion-spark = { path = "datafusion/spark", version = "54.0.0" }
datafusion-sql = { path = "datafusion/sql", version = "54.0.0" }
datafusion-substrait = { path = "datafusion/substrait", version = "54.0.0" }
+genawaiter = { version = "0.99.1", default-features = false }
Review Comment:
I have found that the readme of a crate always puts the most positive spin
on it
https://crates.io/crates/genawaiter
My hesitation for adding new dependencies is that we will now basically have
one more thing we have to deal with (e.g. RUSTSEC, bugs, etc)
From a little research, `genawaiter` seems like an unmaintained crate (one
author, last release was 6 years ago)
<img width="1392" height="793" alt="Image"
src="https://github.com/user-attachments/assets/90b0e2e0-1ce4-4713-98f5-f1e3e3d37b3b";
/>
<img width="1221" height="962" alt="Image"
src="https://github.com/user-attachments/assets/2835dbd2-5446-4253-adca-8d1d6d99dc6c";
/>
Thus from a practical perspective if there is a problem, we (the DataFusion
community) will likely be responsible for fixing anything that comes up.
Therefore I think we would probably hve to vendor this crate, but the repo
doesn't have a LICENSE file https://github.com/whatisaphone/genawaiter (though
the crates.io page says "MIT Licensed")
##########
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:
BTW this looks pretty similar to what
[futures::stream::try_unfold](https://docs.rs/futures/latest/futures/stream/fn.try_unfold.html)
does (and that is already used elsewhere in DataFusion)
--
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]
