[
https://issues.apache.org/jira/browse/ARROW-9707?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17201435#comment-17201435
]
Jorge edited comment on ARROW-9707 at 9/24/20, 10:33 AM:
---------------------------------------------------------
I thought a bit about this, and I have an hypothesis: the core pattern in
DataFusion today is:
# {{ExecutionPlan}} is an iterator of {{RecordBatchReader}} via the function
{{build}}
# {{RecordBatchReaders}} is an iterator of {{RecordBatch}} via {{next_batch}}
# {{ExecutionPlan}} 's {{size_hint}} is given by {{output_partitioning}}
# {{RecordBatchReaders}}'s {{size_hint}} is unknown, as it typically comes
from scaning through a file
If this hypothesis holds, IMO we could convert {{ExecutionPlan}} to allow an
{{IntoIter<Item=IntoIter<Item=RecordBatch>>}}
which would allow to easily flatten the iterator over a set of threads. Threads
switch context during I/O, e.g. when a thread finishes reading a block of
batches in a CSV, it can switch to execute whatever it is happening in another
place.
This will also give more visibility to a scheduler, as it has all the necessary
information it needs to schedule tasks.
[this answer in SO|https://stackoverflow.com/a/53176418/931303] uses IntoIter
of IntoIter to implement a generic (single-threaded) {{merge_sort}}. This would
also allow other architectures, as they would offload whole partitions to
different processes. In our context, something like this (all static for now
{code:java}
fn merge_sorted<IterT: 'static + Send, IterIterT, T: 'static + Ord + Clone +
fmt::Debug + Send>(arrays: IterIterT) -> Vec<T>
where
IterT: IntoIterator<Item = T>,
IterIterT: IntoIterator<Item = IterT>,
{
let all_values = Arc::new(Mutex::new(vec![]));
let threads: Vec<JoinHandle<()>> = arrays.into_iter().map(|array| {
let mutex_clone = Arc::clone(&all_values);
thread::spawn(move || {
let mut values: Vec<T> = array.into_iter().collect();
values.sort(); // this is wrong, but all information is available
to do it
mutex_clone.lock().unwrap().extend_from_slice(&values);
})
}).collect();
for thread in threads {
thread.join().unwrap()
}
let result = all_values.lock().unwrap().clone();
result
}
{code}
was (Author: jorgecarleitao):
I thought a bit about this, and I have an hypothesis: the core pattern in
DataFusion today is:
# {{ExecutionPlan}} is an iterator of {{RecordBatchReader}} via the function
{{build}}
# {{RecordBatchReaders}} is an iterator of {{RecordBatch}} via {{next_batch}}
# {{ExecutionPlan}} 's {{size_hint}} is given by {{output_partitioning}}
# {{RecordBatchReaders}}{{}} {{size_hint}} is unknown, as it typically comes
from scaning through a file
If this hypothesis holds, IMO we could convert {{ExecutionPlan}} to allow an
{{IntoIter<Item=IntoIter<Item=RecordBatch>>}}
which would allow to easily flatten the iterator over a set of threads. Threads
switch context during I/O, e.g. when a thread finishes reading a block of
batches in a CSV, it can switch to execute whatever it is happening in another
place.
This will also give more visibility to a scheduler, as it has all the necessary
information it needs to schedule tasks.
[this answer in SO|https://stackoverflow.com/a/53176418/931303] uses IntoIter
of IntoIter to implement a generic (single-threaded) {{merge_sort}}. This would
also allow other architectures, as they would offload whole partitions to
different processes. In our context, something like this (all static for now
{code:java}
fn merge_sorted<IterT: 'static + Send, IterIterT, T: 'static + Ord + Clone +
fmt::Debug + Send>(arrays: IterIterT) -> Vec<T>
where
IterT: IntoIterator<Item = T>,
IterIterT: IntoIterator<Item = IterT>,
{
let all_values = Arc::new(Mutex::new(vec![]));
let threads: Vec<JoinHandle<()>> = arrays.into_iter().map(|array| {
let mutex_clone = Arc::clone(&all_values);
thread::spawn(move || {
let mut values: Vec<T> = array.into_iter().collect();
values.sort(); // this is wrong, but all information is available
to do it
mutex_clone.lock().unwrap().extend_from_slice(&values);
})
}).collect();
for thread in threads {
thread.join().unwrap()
}
let result = all_values.lock().unwrap().clone();
result
}
{code}
> [Rust] [DataFusion] Re-implement threading model
> ------------------------------------------------
>
> Key: ARROW-9707
> URL: https://issues.apache.org/jira/browse/ARROW-9707
> Project: Apache Arrow
> Issue Type: Sub-task
> Components: Rust, Rust - DataFusion
> Reporter: Andy Grove
> Assignee: Andy Grove
> Priority: Major
> Fix For: 2.0.0
>
>
> The current threading model is very simple and does not scale. We currently
> use 1-2 dedicated threads per partition and they all run simultaneously,
> which is a huge problem if you have more partitions than logical or physical
> cores.
> This task is to re-implement the threading model so that query execution uses
> a fixed (configurable) number of threads. Work will be broken down into
> stages and tasks and each in-process executor (running on a dedicated thread)
> will process its queue of tasks.
> This process will be driven by a scheduler.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
