You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2020/08/13 00:21:06 UTC
[GitHub] [arrow] andygrove opened a new pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads
andygrove opened a new pull request #7951:
URL: https://github.com/apache/arrow/pull/7951
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#issuecomment-673173406
@alamb This is a quick version of what we were discussing
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] jorgecarleitao commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
jorgecarleitao commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r470421892
##########
File path: rust/datafusion/src/execution/context.rs
##########
@@ -62,10 +62,31 @@ use crate::sql::{
};
use crate::table::Table;
+/// Configuration options for execution context
+#[derive(Copy, Clone)]
+pub struct ExecutionConfig {
+ /// Maximum number of concurrent threads for query execution
+ max_concurrency: usize,
Review comment:
Isn't it simpler to make this value public instead of having `with_max_concurrency` as a setter?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] alamb commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r470019311
##########
File path: rust/datafusion/src/execution/physical_plan/merge.rs
##########
@@ -64,33 +75,51 @@ struct MergePartition {
schema: SchemaRef,
/// Input partitions
partitions: Vec<Arc<dyn Partition>>,
+ /// Maximum number of concurrent threads
+ max_concurrency: usize,
+}
+
+fn collect_from_thread(
+ thread: JoinHandle<Result<Vec<RecordBatch>>>,
+ combined_results: &mut Vec<Arc<RecordBatch>>,
+) -> Result<()> {
+ match thread.join() {
+ Ok(join) => {
+ join?
+ .iter()
+ .for_each(|batch| combined_results.push(Arc::new(batch.clone())));
+ Ok(())
+ }
+ Err(e) => Err(ExecutionError::General(format!(
+ "Error collecting batches from thread: {:?}",
+ e
+ ))),
+ }
}
impl Partition for MergePartition {
fn execute(&self) -> Result<Arc<Mutex<dyn RecordBatchReader + Send + Sync>>> {
- let threads: Vec<JoinHandle<Result<Vec<RecordBatch>>>> = self
- .partitions
- .iter()
- .map(|p| {
- let p = p.clone();
- thread::spawn(move || {
- let it = p.execute()?;
- common::collect(it)
- })
- })
- .collect();
+ let mut combined_results: Vec<Arc<RecordBatch>> = vec![];
+ let mut threads: Vec<JoinHandle<Result<Vec<RecordBatch>>>> = vec![];
+
+ for partition in &self.partitions {
+ // limit number of concurrent threads
+ if threads.len() == self.max_concurrency {
Review comment:
This approach (to wait for an arbitrary thread (thread 0) to be done) can lead to cases where the thread you are waiting on happens to have a large amount of work but other threads are waiting on the next partitions. Thus you can block here with only one thread working.
Another approach that would avoid that starvation would be to put the partitions into a `Arc<Mutex<Channel>>` and then have the threads each read and process partitions from that channel.
I can try and whip that up if you would like a specific code example of what I am talking about
##########
File path: rust/datafusion/src/execution/context.rs
##########
@@ -62,10 +62,31 @@ use crate::sql::{
};
use crate::table::Table;
+/// Configuration options for execution context
+#[derive(Copy, Clone)]
+pub struct ExecutionConfig {
+ /// Maximum number of concurrent threads for query execution
+ max_concurrency: usize,
+}
+
+impl ExecutionConfig {
+ /// Create an execution config with default settings
+ pub fn new() -> Self {
+ Self { max_concurrency: 2 }
Review comment:
I think the canonical choice for default concurrency is "number of CPU cores on the machine", which is the approach that Rayon takes and I think you can use this Rust library: https://docs.rs/num_cpus/1.13.0/num_cpus/
##########
File path: rust/datafusion/src/execution/physical_plan/merge.rs
##########
@@ -64,33 +75,51 @@ struct MergePartition {
schema: SchemaRef,
/// Input partitions
partitions: Vec<Arc<dyn Partition>>,
+ /// Maximum number of concurrent threads
+ max_concurrency: usize,
+}
+
+fn collect_from_thread(
Review comment:
👍 Nice refactor
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] alamb commented on pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#issuecomment-674181449
> @alamb How would you feel about us merging this, and then you follow up with an improved approach? I did have a quick attempt at using Rayon but you would probably be able to do this faster and more correctly than I could right now.
I think it is a good idea (for the record) :)
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] alamb commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r470759208
##########
File path: rust/datafusion/src/execution/physical_plan/merge.rs
##########
@@ -64,39 +76,70 @@ struct MergePartition {
schema: SchemaRef,
/// Input partitions
partitions: Vec<Arc<dyn Partition>>,
+ /// Maximum number of concurrent threads
+ concurrency: usize,
+}
+
+fn collect_from_thread(
+ thread: JoinHandle<Result<Vec<RecordBatch>>>,
+ combined_results: &mut Vec<Arc<RecordBatch>>,
+) -> Result<()> {
+ match thread.join() {
+ Ok(join) => {
+ join?
+ .iter()
+ .for_each(|batch| combined_results.push(Arc::new(batch.clone())));
+ Ok(())
+ }
+ Err(e) => Err(ExecutionError::General(format!(
+ "Error collecting batches from thread: {:?}",
+ e
+ ))),
+ }
}
impl Partition for MergePartition {
fn execute(&self) -> Result<Arc<Mutex<dyn RecordBatchReader + Send + Sync>>> {
- let threads: Vec<JoinHandle<Result<Vec<RecordBatch>>>> = self
- .partitions
- .iter()
- .map(|p| {
- let p = p.clone();
- thread::spawn(move || {
- let it = p.execute()?;
- common::collect(it)
- })
- })
- .collect();
-
- // combine the results from each thread
- let mut combined_results: Vec<Arc<RecordBatch>> = vec![];
- for thread in threads {
- match thread.join() {
- Ok(join) => {
- join?
- .iter()
- .for_each(|batch| combined_results.push(Arc::new(batch.clone())));
+ match self.partitions.len() {
+ 0 => Err(ExecutionError::General(
+ "MergeExec requires at least one input partition".to_owned(),
+ )),
+ 1 => {
+ // bypass any threading if there is a single partition
+ self.partitions[0].execute()
+ }
+ _ => {
+ let partitions_per_thread =
+ (self.partitions.len() / self.concurrency).max(1);
+ let chunks = self.partitions.chunks(partitions_per_thread);
Review comment:
yeah, again this approach will work well as long as the work per partition (and thus chunks) is relatively even and well balanced. If we have imbalanced partitions then some threads may be idle when there is still work to do in chunks being processed by other threads
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove edited a comment on pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove edited a comment on pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#issuecomment-673495888
Actually this needs to happen in multiple operators, not just MergeExec, so I am exploring adding a thread pool crate now to avoid boilerplate code everywhere.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r470603772
##########
File path: rust/datafusion/src/execution/context.rs
##########
@@ -62,10 +62,31 @@ use crate::sql::{
};
use crate::table::Table;
+/// Configuration options for execution context
+#[derive(Copy, Clone)]
+pub struct ExecutionConfig {
+ /// Maximum number of concurrent threads for query execution
+ max_concurrency: usize,
Review comment:
I was following the builder pattern that @houqp introduced with `CsvReadOptions`. Once we have more configs we can chain them together, like this:
```
ExecutionConfig::new()
.with_concurrency(48)
.with_foo(...)
.with_bar(...)
```
We can also add validation logic in these methods. For example, I have now added an assertion in `with_concurrency` to assert that the argument is > 0.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#issuecomment-673495888
Actually this needs to happen in multipe operators, not just MergeExec, so I am exploring adding a thread pool crate now to avoid boilerplate code everywhere.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] jorgecarleitao commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
jorgecarleitao commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r470725832
##########
File path: rust/datafusion/src/execution/context.rs
##########
@@ -62,10 +62,31 @@ use crate::sql::{
};
use crate::table::Table;
+/// Configuration options for execution context
+#[derive(Copy, Clone)]
+pub struct ExecutionConfig {
+ /// Maximum number of concurrent threads for query execution
+ max_concurrency: usize,
Review comment:
Thanks for the explanation. I agree.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#issuecomment-673768514
@alamb How would you feel about us merging this, and then you follow up with an improved approach? I did have a quick attempt at using Rayon but you would probably be able to do this faster and more correctly than I could right now.
With this and the other pending PRs, I can run TPC-H query 1 against a 100 GB data set with 240 partitions with reasonable performance (about the same as Apache Spark).
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#issuecomment-673183472
https://issues.apache.org/jira/browse/ARROW-9716
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r470602850
##########
File path: rust/datafusion/src/execution/context.rs
##########
@@ -62,10 +62,31 @@ use crate::sql::{
};
use crate::table::Table;
+/// Configuration options for execution context
+#[derive(Copy, Clone)]
+pub struct ExecutionConfig {
+ /// Maximum number of concurrent threads for query execution
+ max_concurrency: usize,
+}
+
+impl ExecutionConfig {
+ /// Create an execution config with default settings
+ pub fn new() -> Self {
+ Self { max_concurrency: 2 }
Review comment:
Thanks. I have made this change.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r470036542
##########
File path: rust/datafusion/src/execution/physical_plan/merge.rs
##########
@@ -64,33 +75,51 @@ struct MergePartition {
schema: SchemaRef,
/// Input partitions
partitions: Vec<Arc<dyn Partition>>,
+ /// Maximum number of concurrent threads
+ max_concurrency: usize,
+}
+
+fn collect_from_thread(
+ thread: JoinHandle<Result<Vec<RecordBatch>>>,
+ combined_results: &mut Vec<Arc<RecordBatch>>,
+) -> Result<()> {
+ match thread.join() {
+ Ok(join) => {
+ join?
+ .iter()
+ .for_each(|batch| combined_results.push(Arc::new(batch.clone())));
+ Ok(())
+ }
+ Err(e) => Err(ExecutionError::General(format!(
+ "Error collecting batches from thread: {:?}",
+ e
+ ))),
+ }
}
impl Partition for MergePartition {
fn execute(&self) -> Result<Arc<Mutex<dyn RecordBatchReader + Send + Sync>>> {
- let threads: Vec<JoinHandle<Result<Vec<RecordBatch>>>> = self
- .partitions
- .iter()
- .map(|p| {
- let p = p.clone();
- thread::spawn(move || {
- let it = p.execute()?;
- common::collect(it)
- })
- })
- .collect();
+ let mut combined_results: Vec<Arc<RecordBatch>> = vec![];
+ let mut threads: Vec<JoinHandle<Result<Vec<RecordBatch>>>> = vec![];
+
+ for partition in &self.partitions {
+ // limit number of concurrent threads
+ if threads.len() == self.max_concurrency {
Review comment:
Sure, I use that approach already in Ballista. I can make that change here as well.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove closed pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove closed pull request #7951:
URL: https://github.com/apache/arrow/pull/7951
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#issuecomment-674065115
@alamb @jorgecarleitao This is ready for re-review.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r470602921
##########
File path: rust/datafusion/src/execution/physical_plan/merge.rs
##########
@@ -64,33 +75,51 @@ struct MergePartition {
schema: SchemaRef,
/// Input partitions
partitions: Vec<Arc<dyn Partition>>,
+ /// Maximum number of concurrent threads
+ max_concurrency: usize,
+}
+
+fn collect_from_thread(
+ thread: JoinHandle<Result<Vec<RecordBatch>>>,
+ combined_results: &mut Vec<Arc<RecordBatch>>,
+) -> Result<()> {
+ match thread.join() {
+ Ok(join) => {
+ join?
+ .iter()
+ .for_each(|batch| combined_results.push(Arc::new(batch.clone())));
+ Ok(())
+ }
+ Err(e) => Err(ExecutionError::General(format!(
+ "Error collecting batches from thread: {:?}",
+ e
+ ))),
+ }
}
impl Partition for MergePartition {
fn execute(&self) -> Result<Arc<Mutex<dyn RecordBatchReader + Send + Sync>>> {
- let threads: Vec<JoinHandle<Result<Vec<RecordBatch>>>> = self
- .partitions
- .iter()
- .map(|p| {
- let p = p.clone();
- thread::spawn(move || {
- let it = p.execute()?;
- common::collect(it)
- })
- })
- .collect();
+ let mut combined_results: Vec<Arc<RecordBatch>> = vec![];
+ let mut threads: Vec<JoinHandle<Result<Vec<RecordBatch>>>> = vec![];
+
+ for partition in &self.partitions {
+ // limit number of concurrent threads
+ if threads.len() == self.max_concurrency {
Review comment:
This is now implemented.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] andygrove commented on a change in pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
andygrove commented on a change in pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#discussion_r469629508
##########
File path: rust/datafusion/src/execution/context.rs
##########
@@ -80,9 +82,16 @@ fn tuple_err<T, R>(value: (Result<T>, Result<R>)) -> Result<(T, R)> {
impl ExecutionContext {
/// Create a new execution context for in-memory queries
pub fn new() -> Self {
+ Self::with_max_concurrency(2)
+ }
+
+ /// Create a new execution context for in-memory queries with the specified maximum
+ /// concurrency
+ pub fn with_max_concurrency(max_concurrency: usize) -> Self {
Review comment:
This is lame. It is probably time we had a config struct to pass in.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [arrow] alamb commented on pull request #7951: ARROW-9716: [Rust] [DataFusion] Implement limit on concurrent threads in MergeExec
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #7951:
URL: https://github.com/apache/arrow/pull/7951#issuecomment-673532846
> so I am exploring adding a thread pool crate now to avoid boilerplate code everywhere.
@andygrove -- FWIW I had great luck with Rayon (and its parallel iterators in particular) -- it took care of a lot of the mess I was talking about in https://github.com/apache/arrow/pull/7951#discussion_r470019311 -- in other words, using a rayon parallel iterator would behave well if you are willing to add a rayon dependency
----------------------------------------------------------------
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:
users@infra.apache.org