jorgecarleitao commented on a change in pull request #8503:
URL: https://github.com/apache/arrow/pull/8503#discussion_r512384366
##########
File path: rust/datafusion/src/physical_plan/merge.rs
##########
@@ -103,37 +105,56 @@ impl ExecutionPlan for MergeExec {
self.input.execute(0).await
}
_ => {
- let tasks = (0..input_partitions).map(|part_i| {
+ let (sender, receiver) =
mpsc::unbounded::<ArrowResult<RecordBatch>>();
Review comment:
I it that this is by design: there is no guarantee that a task spawn by
Tokio finishes unless we `.join()` it. A `tokio::spawn` is a future like any
other future: we need to await for it.
At least the problem is well defined: we have a vector (over `parts`) of
futures (`spawn`) whose result is a stream (of `record batches`) and we would
like to convert this into a stream of record batches.
The solution IMO is to have a new adapter for this:
```rust
type Streams = Vec<Item=Future<Stream<Item=Result<RecordBatch>>>>;
struct Adapter {}
impl Adapter {
pub fn new(it: Streams);
}
impl Stream for Adapter {
item = Result<RecordBatch>
fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
// poll any of the tasks, if it is done, store the stream in `self`
and start pulling
// from the stream together with the other tasks
}
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
