tustvold commented on code in PR #4097:
URL: https://github.com/apache/arrow-datafusion/pull/4097#discussion_r1013330781
##########
datafusion/core/src/scheduler/task.rs:
##########
@@ -108,17 +108,24 @@ impl Task {
routable: &RoutablePipeline,
error: DataFusionError,
) {
- self.context.send_query_output(partition, Err(error));
- if let Some(link) = routable.output {
- trace!(
- "Closing pipeline: {:?}, partition: {}, due to error",
- link,
- self.waker.partition,
- );
-
- self.context.pipelines[link.pipeline]
- .pipeline
- .close(link.child, self.waker.partition);
+ match routable.output {
+ Some(link) => {
+ // The query output partition number may not equal current
pipeline's output partition number,
+ // but the query output has at least one partition，
+ // so send error to the first partition of the query output.
Review Comment:
```suggestion
// The query output partitioning may not match the current
pipeline's
// but the query output has at least one partition
// so send error to the first partition of the query output.
```
##########
datafusion/core/src/scheduler/mod.rs:
##########
@@ -353,8 +358,8 @@ mod tests {
info!("Plan: {}", displayable(plan.as_ref()).indent());
let stream = scheduler.schedule(plan, task).unwrap().stream();
- let scheduled: Vec<_> = stream.try_collect().await.unwrap();
- let expected = query.collect().await.unwrap();
+ let scheduled: Vec<_> =
stream.try_collect().await.unwrap_or(vec![]);
+ let expected = query.collect().await.unwrap_or(vec![]);
Review Comment:
```suggestion
let scheduled: Vec<_> =
stream.try_collect().await.unwrap_or_default();
let expected = query.collect().await.unwrap_or_default();
```
--
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]
