Using local shuffle reader avoid flight rpc call. #347
Conversation
| @@ -498,36 +639,37 @@ mod tests { | |||
| Schema::new(vec![Field::new("id", DataType::Int32, false)]) | |||
| } | |||
|
|
|||
| #[derive(Clone)] | |||
| struct MockPartitionReader {} | |||
There was a problem hiding this comment.
this will replace by PartitionReaderEnum::local
|
@mingmwang @andygrove @yahoNanJing PTAL😊 |
| } | ||
|
|
||
| AbortableReceiverStream::create(response_receiver, join_handles) | ||
| } | ||
|
|
||
| fn check_is_local_location(host: &str) -> bool { |
There was a problem hiding this comment.
An alternate approach here would be to just look on disk and see if the shuffle files exist, then this can work in containerized environments, such as k8s.
There was a problem hiding this comment.
Filed #356 so we can discuss this as a follow on
There was a problem hiding this comment.
Sounds like a better approach🤔
|
Thanks @Ted-Jiang! I will test this over the weekend. |
There was a problem hiding this comment.
This is awesome! Thanks @Ted-Jiang. I see a nice 30% speedup when running queries locally. 🚀
@andygrove Cool! We can care the locality of shuffle in the future!👍 |
|
@Ted-Jiang Could you please add some log info to indicate how many shuffle partitions are read from local and how many are ready from remote? |
|
Please pay attention to the error handling, both remote shuffle read and local shuffle read should return the same error when encounter issues, so that the Scheduler can reschedule tasks/stages.
|
Sure! |
Which issue does this PR close?
Closes #346.
There is a situation when shuffler reader and writer are on the same machine. So we can read from directly from local disk without calling RPC and arrow-flight decoding.
Rationale for this change
What changes are included in this PR?
Are there any user-facing changes?