You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "korowa (via GitHub)" <gi...@apache.org> on 2023/03/06 19:22:19 UTC
[GitHub] [arrow-datafusion] korowa opened a new pull request, #5490: Memory limited hash join
korowa opened a new pull request, #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Part of #5339
Part of #5220
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
Control over memory allocations made by `HashJoinExec`
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
Major change is `MemoryReservation.try_grow` calls added to 3 places where allocations for build-side data and related structures are performed:
1. build-side record batches collection -- similar to `CrossJoinExec` memory reserved for each `RecordBatch` fetched from build-side, reservation size calculated using already fetched batch, which could lead to insignificant (no more then single batch size) overallocations
2. build-side hashtable -- there is only one allocation happens during `RawTable::with_capacity()` and it could be massive, so it seems reasonable to preemptively estimate the size required for `RawTable` and reserve it in pool -- the calculations of the size are described in comment
3. visited_left_side bitmap in each stream (if required by join type) -- like in case with hashtable, there is an attempt to estimate required size and reserve memory before `ArrayBuilder` creation.
Another change is stream/operator-level reservation (also affects `CrossJoinExec`) -- storing `MemoryReservation`, and freeing it after `...JoinStream` (how it was done in #5339) doesn't seem to be correct for `CrossJoinExec` and `HashJoinExec` with `PartitionMode::CollectLeft` for a couple of reasons:
1. only one stream out of multiple partitions executes `collect_left_data` function and reserves memory -- in case of partition processing time being skewed, the only stream with reservation could be dropped before Join execution is completed
2. LeftData stored in `...JoinExec` rather than in `...ExecStream`, and the object could potentially outlive reservation
These issues are fixed by `OperatorMemoryReservation` added to both `CrossJoinExec` and `HashJoinExec` -- for hash joins this reservation used only for `PartitionedMode::CollectLeft` cases, in case of parittioned joins -- stream-level memory reservations are used.
# Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
5. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
Test cases for overallocation has been added for all `JoinType`s, for both `CollectLeft` and `Partitioned` modes.
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
`HashJoinExec` should now respect runtime memory limitations.
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] korowa commented on a diff in pull request #5490: Memory limited hash join
Posted by "korowa (via GitHub)" <gi...@apache.org>.
korowa commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1129895159
##########
datafusion/core/src/physical_plan/joins/hash_join.rs:
##########
@@ -1173,6 +1149,18 @@ impl HashJoinStream {
};
build_timer.done();
+ // Reserving memory for visited_left_side bitmap in case it hasn't been initialied yet
+ // and join_type requires to store it
+ if self.visited_left_side.is_none()
Review Comment:
It was my first intention, but then I've realized that `Option` doesn't have fallible versions of `get_or_insert`-like methods (i.e.`try_get_or_insert_with` in this case would be just right), and, the only alternative I saw, was storing `visited_left_side` as a `Result<_, DataFusionError> which, I guess, would also be not really clean.
##########
datafusion/core/src/physical_plan/common.rs:
##########
@@ -39,8 +39,14 @@ use std::task::{Context, Poll};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
+/// [`MemoryReservation`] used across query execution streams
pub(crate) type SharedMemoryReservation = Arc<Mutex<MemoryReservation>>;
+/// [`MemoryReservation`] used at query operator level
+/// `Option` wrapper allows to initialize empty reservation in operator constructor,
+/// and set it to actual reservation at stream level.
+pub(crate) type OperatorMemoryReservation = Arc<Mutex<Option<SharedMemoryReservation>>>;
Review Comment:
Unfortunately I haven't found better solution yet, but I'll check if there is better way to wrap `MemoryReservation` into something allowing attempts of creation (or initialization) reservation from multiple streams.
Another solution, probably, is to allow to allow creation of MemoryReservation without providing context (in `HashJoinExec::new`), and registering it in `MemoryPool` later (in `HashJoinExec.execute`, when we have context available).
##########
datafusion/core/src/physical_plan/joins/cross_join.rs:
##########
@@ -452,10 +464,7 @@ impl CrossJoinStream {
Some(result)
}
- other => {
- self.reservation.lock().free();
- other
- }
+ other => other,
Review Comment:
True, now it seems more reliable to store reservations along with left-data they are targeted at, and utilize [this feature](https://github.com/apache/arrow-datafusion/blob/deeaa5632ed99a58b91767261570756db736d158/datafusion/execution/src/memory_pool/mod.rs#L159)
##########
datafusion/core/src/physical_plan/joins/hash_join.rs:
##########
@@ -493,90 +491,73 @@ impl ExecutionPlan for HashJoinExec {
}
async fn collect_left_input(
+ partition: Option<usize>,
random_state: RandomState,
left: Arc<dyn ExecutionPlan>,
on_left: Vec<Column>,
context: Arc<TaskContext>,
+ metrics: BuildProbeJoinMetrics,
+ reservation: SharedMemoryReservation,
) -> Result<JoinLeftData> {
let schema = left.schema();
- let start = Instant::now();
- // merge all left parts into a single stream
- let merge = {
- if left.output_partitioning().partition_count() != 1 {
- Arc::new(CoalescePartitionsExec::new(left))
- } else {
- left
- }
- };
- let stream = merge.execute(0, context)?;
Review Comment:
Correct, these two functions shared same logic, except for left-side stream sourcing.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on pull request #5490: Memory limited hash join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1462395616
> I've also created https://github.com/apache/arrow-datafusion/pull/5531 to support passing boolean by values in one of the tests added in this PR π
Thank you for your diligence here π
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #5490: Memory limited hash join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1129788629
##########
datafusion/core/src/physical_plan/joins/cross_join.rs:
##########
@@ -452,10 +464,7 @@ impl CrossJoinStream {
Some(result)
}
- other => {
- self.reservation.lock().free();
- other
- }
+ other => other,
Review Comment:
My understanding is that the memory is freed on drop of the `SharedReservation`, is that correct?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on pull request #5490: Memory limited hash join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1458940848
Thank you @korowa -- I plan to review this PR carefully tomorrow morning
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] korowa commented on a diff in pull request #5490: Memory limited hash join
Posted by "korowa (via GitHub)" <gi...@apache.org>.
korowa commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1130636974
##########
datafusion/core/src/physical_plan/common.rs:
##########
@@ -39,8 +39,14 @@ use std::task::{Context, Poll};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
+/// [`MemoryReservation`] used across query execution streams
pub(crate) type SharedMemoryReservation = Arc<Mutex<MemoryReservation>>;
+/// [`MemoryReservation`] used at query operator level
+/// `Option` wrapper allows to initialize empty reservation in operator constructor,
+/// and set it to actual reservation at stream level.
+pub(crate) type OperatorMemoryReservation = Arc<Mutex<Option<SharedMemoryReservation>>>;
Review Comment:
So, if there is not objections, I'd prefer to figure out proper way to handle operator-level reservations as a separate follow-up PR, or one of remaining joins (at least NL will have the same issue 100%)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #5490: Memory limited hash join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1129654905
##########
datafusion/core/src/physical_plan/common.rs:
##########
@@ -39,8 +39,14 @@ use std::task::{Context, Poll};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
+/// [`MemoryReservation`] used across query execution streams
pub(crate) type SharedMemoryReservation = Arc<Mutex<MemoryReservation>>;
+/// [`MemoryReservation`] used at query operator level
+/// `Option` wrapper allows to initialize empty reservation in operator constructor,
+/// and set it to actual reservation at stream level.
+pub(crate) type OperatorMemoryReservation = Arc<Mutex<Option<SharedMemoryReservation>>>;
Review Comment:
~I wonder if the double layer of Mutex is necessary (Mutex over an Option of a Mutex)~
Update after reading the code I see why it is necessary.
Perhaps it would make the code clearer (as a follow on PR, perhaps) if we had a proper `SharedMemoryReservation` that could be cloned and handled mutability internally. π€
##########
datafusion/core/src/physical_plan/joins/hash_join.rs:
##########
@@ -1173,6 +1149,18 @@ impl HashJoinStream {
};
build_timer.done();
+ // Reserving memory for visited_left_side bitmap in case it hasn't been initialied yet
+ // and join_type requires to store it
+ if self.visited_left_side.is_none()
Review Comment:
You might be able to make this less verbose by moving the initialization into the initialization of `visited_left_size`
```
let visited_left_side = self.visited_left_side.get_or_insert_with(|| {
let visited_bitmap_size = bit_util::ceil(left_data.1.num_rows(), 8);
self.reservation.lock().try_grow(visited_bitmap_size)?;
self.join_metrics.build_mem_used.add(visited_bitmap_size);
...
```
##########
datafusion/core/src/physical_plan/joins/cross_join.rs:
##########
@@ -683,6 +692,7 @@ mod tests {
err.to_string(),
"External error: Resources exhausted: Failed to allocate additional"
);
+ assert_contains!(err.to_string(), "CrossJoinExec");
Review Comment:
π
##########
datafusion/core/src/physical_plan/joins/hash_join.rs:
##########
@@ -146,50 +150,6 @@ pub struct HashJoinExec {
pub(crate) null_equals_null: bool,
}
-/// Metrics for HashJoinExec
Review Comment:
it is a nice cleanup to use `BuildProbeMetrics` π
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] Dandandan commented on pull request #5490: Memory limited hash join
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1459854702
> Thanks @Dandandan!
>
> > I think it would be great if we could run some benchmarks to show that we're not regressing too much (e.g. running tpch benchmark queries with joins).
>
> I've done a couple of iterations for
>
> ```
> cargo run --release --bin tpch -- benchmark datafusion --iterations 5 --path ./parquet --format parquet
> ```
>
> With the result (a bit floating but I can't see any noticeable regressions -- ofc I may be wrong):
>
> ```
> Query branch main
> ----------------------------------------------
> Query 1 avg time: 1047.93 ms 1135.36 ms
> Query 2 avg time: 280.91 ms 286.69 ms
> Query 3 avg time: 323.87 ms 351.31 ms
> Query 4 avg time: 146.87 ms 146.58 ms
> Query 5 avg time: 482.85 ms 463.07 ms
> Query 6 avg time: 274.73 ms 342.29 ms
> Query 7 avg time: 750.73 ms 762.43 ms
> Query 8 avg time: 443.34 ms 426.89 ms
> Query 9 avg time: 821.48 ms 775.03 ms
> Query 10 avg time: 585.21 ms 584.16 ms
> Query 11 avg time: 247.56 ms 232.90 ms
> Query 12 avg time: 258.51 ms 231.19 ms
> Query 13 avg time: 899.16 ms 885.56 ms
> Query 14 avg time: 300.63 ms 282.56 ms
> Query 15 avg time: 346.36 ms 318.97 ms
> Query 16 avg time: 198.33 ms 184.26 ms
> Query 17 avg time: 4197.54 ms 4101.92 ms
> Query 18 avg time: 2726.41 ms 2548.96 ms
> Query 19 avg time: 566.67 ms 535.74 ms
> Query 20 avg time: 1193.82 ms 1319.49 ms
> Query 21 avg time: 1027.00 ms 1050.08 ms
> Query 22 avg time: 120.03 ms 111.32 ms
> ```
>
> Also
>
> ```
> cargo run --release --bin tpch -- benchmark datafusion --query 17 --iterations 5 --path ./parquet --format parquet --debug | grep -e "build_time" -e "iteration"
> ```
>
> is not showing any difference in join build / probe times. Full log is in attachment. [mem_mgmt_benches.txt](https://github.com/apache/arrow-datafusion/files/10918984/mem_mgmt_benches.txt)
>
> As for now, I tend to think that these changes do not affect performance significantly (due to the fact that reservations and memory estimations executed only once per partition/query), but I'll be happy to perform any additional checks.
>
> > Some reasons I defaulted to initializing the hashmap using the size of the left side is as following
>
> All of them look pretty reasonable and I agree with this approach, I was just trying to explain why I've decided to duplicate calculations from some functions subsequently called by `RawTable.with_capacity`.
Awesome, thanks for checking. It might be worth it to repeat the benchmark run with `-m` (loading to in-memory table) to get the Parquet scan out of the equation.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] Dandandan commented on pull request #5490: Memory limited hash join
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1460730390
It looks like the changes don't affect performance, which is perfect π
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] korowa commented on pull request #5490: Memory limited hash join
Posted by "korowa (via GitHub)" <gi...@apache.org>.
korowa commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1460721845
I've also benchmarked vs main using `-m` flag -- and despite it should be cleaner due to excluding file scanning from execution time, there still are fluctuations in both sides
```
Query branch main
----------------------------------------------
Query 1 avg time: 668.26 ms 652.55 ms
Query 2 avg time: 212.92 ms 216.46 ms
Query 3 avg time: 113.40 ms 116.00 ms
Query 4 avg time: 73.98 ms 65.35 ms
Query 5 avg time: 333.21 ms 318.39 ms
Query 6 avg time: 26.49 ms 28.68 ms
Query 7 avg time: 690.75 ms 732.79 ms
Query 8 avg time: 151.35 ms 170.01 ms
Query 9 avg time: 378.67 ms 415.37 ms
Query 10 avg time: 228.83 ms 252.53 ms
Query 11 avg time: 240.78 ms 216.22 ms
Query 12 avg time: 98.05 ms 96.31 ms
Query 13 avg time: 657.82 ms 638.33 ms
Query 14 avg time: 38.87 ms 35.11 ms
Query 15 avg time: 101.51 ms 109.23 ms
Query 16 avg time: 149.93 ms 149.01 ms
Query 17 avg time: 3713.62 ms 3952.60 ms
Query 18 avg time: 2312.22 ms 2463.61 ms
Query 19 avg time: 106.72 ms 107.72 ms
Query 20 avg time: 1047.35 ms 1022.07 ms
Query 21 avg time: 892.56 ms 855.86 ms
Query 22 avg time: 87.27 ms 73.53 ms
```
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] korowa commented on a diff in pull request #5490: Memory limited hash join
Posted by "korowa (via GitHub)" <gi...@apache.org>.
korowa commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1130636974
##########
datafusion/core/src/physical_plan/common.rs:
##########
@@ -39,8 +39,14 @@ use std::task::{Context, Poll};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
+/// [`MemoryReservation`] used across query execution streams
pub(crate) type SharedMemoryReservation = Arc<Mutex<MemoryReservation>>;
+/// [`MemoryReservation`] used at query operator level
+/// `Option` wrapper allows to initialize empty reservation in operator constructor,
+/// and set it to actual reservation at stream level.
+pub(crate) type OperatorMemoryReservation = Arc<Mutex<Option<SharedMemoryReservation>>>;
Review Comment:
So, if there are no objections, I'd prefer to figure out proper way to handle operator-level reservations as a separate follow-up PR, or one of remaining joins (at least NL will have the same issue 100%)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on pull request #5490: Memory limited hash join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1461976868
Thanks again @korowa !
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] Dandandan commented on a diff in pull request #5490: Memory limited hash join
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1129916540
##########
datafusion/core/src/physical_plan/joins/hash_join.rs:
##########
@@ -493,90 +491,73 @@ impl ExecutionPlan for HashJoinExec {
}
async fn collect_left_input(
+ partition: Option<usize>,
random_state: RandomState,
left: Arc<dyn ExecutionPlan>,
on_left: Vec<Column>,
context: Arc<TaskContext>,
+ metrics: BuildProbeJoinMetrics,
+ reservation: SharedMemoryReservation,
) -> Result<JoinLeftData> {
let schema = left.schema();
- let start = Instant::now();
- // merge all left parts into a single stream
- let merge = {
- if left.output_partitioning().partition_count() != 1 {
- Arc::new(CoalescePartitionsExec::new(left))
- } else {
- left
- }
- };
- let stream = merge.execute(0, context)?;
-
- // This operation performs 2 steps at once:
- // 1. creates a [JoinHashMap] of all batches from the stream
- // 2. stores the batches in a vector.
- let initial = (0, Vec::new());
- let (num_rows, batches) = stream
- .try_fold(initial, |mut acc, batch| async {
- acc.0 += batch.num_rows();
- acc.1.push(batch);
- Ok(acc)
- })
- .await?;
- let mut hashmap = JoinHashMap(RawTable::with_capacity(num_rows));
- let mut hashes_buffer = Vec::new();
- let mut offset = 0;
- for batch in batches.iter() {
- hashes_buffer.clear();
- hashes_buffer.resize(batch.num_rows(), 0);
- update_hash(
- &on_left,
- batch,
- &mut hashmap,
- offset,
- &random_state,
- &mut hashes_buffer,
- )?;
- offset += batch.num_rows();
- }
- // Merge all batches into a single batch, so we
- // can directly index into the arrays
- let single_batch = concat_batches(&schema, &batches, num_rows)?;
-
- debug!(
- "Built build-side of hash join containing {} rows in {} ms",
- num_rows,
- start.elapsed().as_millis()
- );
-
- Ok((hashmap, single_batch))
-}
-
-async fn partitioned_left_input(
- partition: usize,
- random_state: RandomState,
- left: Arc<dyn ExecutionPlan>,
- on_left: Vec<Column>,
- context: Arc<TaskContext>,
-) -> Result<JoinLeftData> {
- let schema = left.schema();
+ let (left_input, left_input_partition) = if let Some(partition) = partition {
+ (left, partition)
+ } else {
+ let merge = {
+ if left.output_partitioning().partition_count() != 1 {
+ Arc::new(CoalescePartitionsExec::new(left))
+ } else {
+ left
+ }
+ };
- let start = Instant::now();
+ (merge, 0)
+ };
- // Load 1 partition of left side in memory
- let stream = left.execute(partition, context.clone())?;
+ // Depending on partition argument load single partition or whole left side in memory
+ let stream = left_input.execute(left_input_partition, context.clone())?;
// This operation performs 2 steps at once:
// 1. creates a [JoinHashMap] of all batches from the stream
// 2. stores the batches in a vector.
- let initial = (0, Vec::new());
- let (num_rows, batches) = stream
+ let initial = (Vec::new(), 0, metrics, reservation);
+ let (batches, num_rows, metrics, reservation) = stream
.try_fold(initial, |mut acc, batch| async {
- acc.0 += batch.num_rows();
- acc.1.push(batch);
+ let batch_size = batch.get_array_memory_size();
+ // Reserve memory for incoming batch
+ acc.3.lock().try_grow(batch_size)?;
+ // Update metrics
+ acc.2.build_mem_used.add(batch_size);
+ acc.2.build_input_batches.add(1);
+ acc.2.build_input_rows.add(batch.num_rows());
+ // Update rowcount
+ acc.1 += batch.num_rows();
+ // Push batch to output
+ acc.0.push(batch);
Ok(acc)
})
.await?;
+ // Estimation of memory size, required for hashtable, prior to allocation.
+ // Final result can be verified using `RawTable.allocation_info()`
+ //
+ // For majority of cases hashbrown overestimates buckets qty to keep ~1/8 of them empty.
+ // This formula leads to overallocation for small tables (< 8 elements) but fine overall.
+ let estimated_buckets = (num_rows.checked_mul(8).ok_or_else(|| {
+ DataFusionError::Execution(
+ "usize overflow while estimating number of hasmap buckets".to_string(),
+ )
+ })? / 7)
+ .next_power_of_two();
+ // 32 bytes per `(u64, SmallVec<[u64; 1]>)`
+ // + 1 byte for each bucket
+ // + 16 bytes fixed
+ let estimated_hastable_size = 32 * estimated_buckets + estimated_buckets + 16;
+
+ reservation.lock().try_grow(estimated_hastable_size)?;
+ metrics.build_mem_used.add(estimated_hastable_size);
+
let mut hashmap = JoinHashMap(RawTable::with_capacity(num_rows));
Review Comment:
Ah I didn't realize the allocation is the same, only guarding against overallocationπ
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #5490: Memory limited hash join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1129798663
##########
datafusion/core/src/physical_plan/joins/hash_join.rs:
##########
@@ -493,90 +491,73 @@ impl ExecutionPlan for HashJoinExec {
}
async fn collect_left_input(
+ partition: Option<usize>,
random_state: RandomState,
left: Arc<dyn ExecutionPlan>,
on_left: Vec<Column>,
context: Arc<TaskContext>,
+ metrics: BuildProbeJoinMetrics,
+ reservation: SharedMemoryReservation,
) -> Result<JoinLeftData> {
let schema = left.schema();
- let start = Instant::now();
- // merge all left parts into a single stream
- let merge = {
- if left.output_partitioning().partition_count() != 1 {
- Arc::new(CoalescePartitionsExec::new(left))
- } else {
- left
- }
- };
- let stream = merge.execute(0, context)?;
Review Comment:
this change is removing a copy/paste of the code, right? It is very π
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] korowa commented on a diff in pull request #5490: Memory limited hash join
Posted by "korowa (via GitHub)" <gi...@apache.org>.
korowa commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1130636974
##########
datafusion/core/src/physical_plan/common.rs:
##########
@@ -39,8 +39,14 @@ use std::task::{Context, Poll};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
+/// [`MemoryReservation`] used across query execution streams
pub(crate) type SharedMemoryReservation = Arc<Mutex<MemoryReservation>>;
+/// [`MemoryReservation`] used at query operator level
+/// `Option` wrapper allows to initialize empty reservation in operator constructor,
+/// and set it to actual reservation at stream level.
+pub(crate) type OperatorMemoryReservation = Arc<Mutex<Option<SharedMemoryReservation>>>;
Review Comment:
So, if there are no objections, I'd prefer to figure out proper way to handle operator-level reservations as a separate follow-up PR, or during adding memory limitations to one of remaining joins (at least NL will have the same issue 100%)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] korowa commented on pull request #5490: Memory limited hash join
Posted by "korowa (via GitHub)" <gi...@apache.org>.
korowa commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1462049196
Thank you @alamb @Dandandan !
I've alse created #5531 to support passing boolean by values in one of the tests added in this PR π
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] Dandandan commented on pull request #5490: Memory limited hash join
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1462417287
Nice, thanks @korowa !
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ursabot commented on pull request #5490: Memory limited hash join
Posted by "ursabot (via GitHub)" <gi...@apache.org>.
ursabot commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1461989015
Benchmark runs are scheduled for baseline = ff013e24559850541c54cf5d254922649ef37d8c and contender = f5d23ff582f1393d5ce94cf977d11e59f90fe523. f5d23ff582f1393d5ce94cf977d11e59f90fe523 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/3bfa8872cc3040d9895ec0aacb26759f...626d7502b25d4ec1b6cd7be8bc631825/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] [test-mac-arm](https://conbench.ursa.dev/compare/runs/bfefdfdb489849ea85ba15860be8d59b...eb37f2c25fd84d08989dfc868b9ce301/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/bce6a57f53de497a8819d573189441e7...d2e0a44622fa448e984d4e28f60f7e28/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/54ce7426bdad437d9a22a07b893bb110...872cfbdc6a1b4c488cd381e3025f23bb/)
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb merged pull request #5490: Memory limited hash join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] Dandandan commented on pull request #5490: Memory limited hash join
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1458962589
Nice PR!
I think it would be great if we could run some benchmarks to show that we're not regressing too much (e.g. running tpch benchmark queries with joins). Some reasons I defaulted to initializing the hashmap using the size of the left side is as following:
* The build side (for the partition) already has to be loaded into memory, and usually will at least as much and often more memory than the hash table
* For many cases (e.g. unique identifiers) we need this capacity and the estimate is optimal
* Rebuilding the hash table can be slow (although some improvements were made in this area)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] korowa commented on pull request #5490: Memory limited hash join
Posted by "korowa (via GitHub)" <gi...@apache.org>.
korowa commented on PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#issuecomment-1459826565
Thanks @Dandandan!
> I think it would be great if we could run some benchmarks to show that we're not regressing too much (e.g. running tpch benchmark queries with joins).
I've done a couple of iterations for
```
cargo run --release --bin tpch -- benchmark datafusion --iterations 5 --path ./parquet --format parquet
```
With the result (a bit floating but I can't see any noticeable regressions -- ofc I may be wrong):
```
Query branch main
----------------------------------------------
Query 1 avg time: 1047.93 ms 1135.36 ms
Query 2 avg time: 280.91 ms 286.69 ms
Query 3 avg time: 323.87 ms 351.31 ms
Query 4 avg time: 146.87 ms 146.58 ms
Query 5 avg time: 482.85 ms 463.07 ms
Query 6 avg time: 274.73 ms 342.29 ms
Query 7 avg time: 750.73 ms 762.43 ms
Query 8 avg time: 443.34 ms 426.89 ms
Query 9 avg time: 821.48 ms 775.03 ms
Query 10 avg time: 585.21 ms 584.16 ms
Query 11 avg time: 247.56 ms 232.90 ms
Query 12 avg time: 258.51 ms 231.19 ms
Query 13 avg time: 899.16 ms 885.56 ms
Query 14 avg time: 300.63 ms 282.56 ms
Query 15 avg time: 346.36 ms 318.97 ms
Query 16 avg time: 198.33 ms 184.26 ms
Query 17 avg time: 4197.54 ms 4101.92 ms
Query 18 avg time: 2726.41 ms 2548.96 ms
Query 19 avg time: 566.67 ms 535.74 ms
Query 20 avg time: 1193.82 ms 1319.49 ms
Query 21 avg time: 1027.00 ms 1050.08 ms
Query 22 avg time: 120.03 ms 111.32 ms
```
Also
```
cargo run --release --bin tpch -- benchmark datafusion --query 17 --iterations 5 --path ./parquet --format parquet --debug | grep -e "build_time" -e "iteration"
```
is not showing any difference in join build / probe times. Full log is in attachment.
[mem_mgmt_benches.txt](https://github.com/apache/arrow-datafusion/files/10918984/mem_mgmt_benches.txt)
As for now, I tend to think that these changes do not affect performance significantly (due to the fact that reservations and memory estimations executed only once per partition/query), but I'll be happy to perform any additional checks.
> Some reasons I defaulted to initializing the hashmap using the size of the left side is as following
All of them look pretty reasonable and I agree with this approach, I was just trying to explain why I've decided to duplicate calculations from some functions subsequently called by `RawTable.with_capacity`.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] Dandandan commented on a diff in pull request #5490: Memory limited hash join
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #5490:
URL: https://github.com/apache/arrow-datafusion/pull/5490#discussion_r1129919195
##########
datafusion/core/src/physical_plan/joins/hash_join.rs:
##########
@@ -493,90 +491,73 @@ impl ExecutionPlan for HashJoinExec {
}
async fn collect_left_input(
+ partition: Option<usize>,
random_state: RandomState,
left: Arc<dyn ExecutionPlan>,
on_left: Vec<Column>,
context: Arc<TaskContext>,
+ metrics: BuildProbeJoinMetrics,
+ reservation: SharedMemoryReservation,
) -> Result<JoinLeftData> {
let schema = left.schema();
- let start = Instant::now();
- // merge all left parts into a single stream
- let merge = {
- if left.output_partitioning().partition_count() != 1 {
- Arc::new(CoalescePartitionsExec::new(left))
- } else {
- left
- }
- };
- let stream = merge.execute(0, context)?;
Review Comment:
Nice work!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org