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 2022/09/22 22:02:28 UTC
[GitHub] [arrow-datafusion] isidentical opened a new pull request, #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
isidentical opened a new pull request, #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593
# 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 #3579.
# 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.
-->
During sorting, when we receive a new record batch we try to allocate space for it. This is done with the assumption that the result of this sort will still be around, and we don't want to accidentally overflow the memory so we have to keep track of it. But after the #3510, this assumption might not hold for all cases (particularly when you have a `fetch` limit set on your sorting operation) so we might be over-allocating memory and constantly spilling for no good reason.
# 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.
-->
This PR adds the logic for avoiding over-allocations by instructing the memory manager to shrink after each partial sort with a limit.
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
No, this should be an optimization.
<!--
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] Dandandan merged pull request #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
Dandandan merged PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593
--
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 #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
Dandandan commented on code in PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#discussion_r978746836
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -124,6 +124,21 @@ impl ExternalSorter {
// calls to `timer.done()` below.
let _timer = tracking_metrics.elapsed_compute().timer();
let partial = sort_batch(input, self.schema.clone(), &self.expr, self.fetch)?;
+ // The resulting batch might be smaller than the input batch if there
+ // is an propagated limit.
+
+ if self.fetch.is_some() {
+ let new_size = batch_byte_size(&partial.sorted_batch);
+ let size_delta = size.checked_sub(new_size).ok_or_else(|| {
+ DataFusionError::Internal(format!(
+ "The size of the sorted batch is larger than the size of the input batch: {} > {}",
+ size,
+ new_size
+ ))
+ })?;
+ self.shrink(size_delta);
+ self.metrics.mem_used().sub(size_delta);
Review Comment:
Change looks good - maybe we can add a unit test for the memory to be reduced when a `fetch` is provided.
--
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 #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#issuecomment-1257010500
Benchmark runs are scheduled for baseline = 8bcc96532ad4272bebff09796b6dda87b9030a9a and contender = 696a0b5146842978962cc7e3b94a6a993e2ad460. 696a0b5146842978962cc7e3b94a6a993e2ad460 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/a23c811eebf0482fb6a56adc85725122...74fcfe896b5e49d892ababb1099fdfe9/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] [test-mac-arm](https://conbench.ursa.dev/compare/runs/f670afcc01bc4cbebc9e289cae21d4d0...d39111fa30f5437994b40e9feae4a4c2/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/96865eb9398043219c6063970abe727f...94cae110c3bd44edb59bfc42f4c94a82/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/38e7a47c71f64a8b9dce5eec66346649...3cf3eadf10f3489a931ba15c2253b61e/)
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] isidentical commented on a diff in pull request #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
isidentical commented on code in PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#discussion_r978935000
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -124,6 +124,21 @@ impl ExternalSorter {
// calls to `timer.done()` below.
let _timer = tracking_metrics.elapsed_compute().timer();
let partial = sort_batch(input, self.schema.clone(), &self.expr, self.fetch)?;
+ // The resulting batch might be smaller than the input batch if there
+ // is an propagated limit.
+
+ if self.fetch.is_some() {
+ let new_size = batch_byte_size(&partial.sorted_batch);
+ let size_delta = size.checked_sub(new_size).ok_or_else(|| {
+ DataFusionError::Internal(format!(
+ "The size of the sorted batch is larger than the size of the input batch: {} > {}",
+ size,
+ new_size
+ ))
+ })?;
+ self.shrink(size_delta);
+ self.metrics.mem_used().sub(size_delta);
Review Comment:
@Dandandan I was trying a bunch of methods to write a reliable test, but it seems like without accessing to the internal implementation details (creating a stream, feeding it one by one, comparing the size of internal memory metrics with the total size of in memory batches) it seems to be very tricky to do so (with only using the public stuff, like the `SortExec`).
I can also try to write a test case to ensure a specific usage does not spill (which would normally spill without a limit), but that also seems like would test a factor of conditions (the initial TopK implementation, if implemented #3579, this PR etc. they all try to prevent spills) and not this fix as is.
If the spilling/not-spilling one sounds good, I can submit it (or if you have any other suggestions, would happy to also give them a try).
--
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 #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#issuecomment-1256935762
Thank you @isidentical
--
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 #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
Dandandan commented on code in PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#discussion_r979252095
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -1062,6 +1077,65 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn test_sort_fetch_memory_calculation() -> Result<()> {
+ // This test mirrors down the size from the example above.
+ let avg_batch_size = 5336;
+ let partitions = 4;
+
+ // A tuple of (fetch, expect_spillage)
+ let test_options = vec![
+ // Since we don't have a limit (and the memory is less than the total size of
+ // all the batches we are processing, we expect it to spill.
+ (None, true),
+ // When we have a limit however, the buffered size of batches should fit in memory
+ // since it is much lover than the total size of the input batch.
+ (Some(1), false),
+ ];
+
+ for (fetch, expect_spillage) in test_options {
+ let config = RuntimeConfig::new()
+ .with_memory_limit(avg_batch_size * (partitions - 1), 1.0);
Review Comment:
👍
--
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 #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
Dandandan commented on code in PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#discussion_r979229766
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -124,6 +124,21 @@ impl ExternalSorter {
// calls to `timer.done()` below.
let _timer = tracking_metrics.elapsed_compute().timer();
let partial = sort_batch(input, self.schema.clone(), &self.expr, self.fetch)?;
+ // The resulting batch might be smaller than the input batch if there
+ // is an propagated limit.
+
+ if self.fetch.is_some() {
+ let new_size = batch_byte_size(&partial.sorted_batch);
+ let size_delta = size.checked_sub(new_size).ok_or_else(|| {
+ DataFusionError::Internal(format!(
+ "The size of the sorted batch is larger than the size of the input batch: {} > {}",
+ size,
+ new_size
+ ))
+ })?;
+ self.shrink(size_delta);
+ self.metrics.mem_used().sub(size_delta);
Review Comment:
A test for spilling/not spilling sounds good!
--
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] isidentical commented on a diff in pull request #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
isidentical commented on code in PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#discussion_r979251679
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -124,6 +124,21 @@ impl ExternalSorter {
// calls to `timer.done()` below.
let _timer = tracking_metrics.elapsed_compute().timer();
let partial = sort_batch(input, self.schema.clone(), &self.expr, self.fetch)?;
+ // The resulting batch might be smaller than the input batch if there
+ // is an propagated limit.
+
+ if self.fetch.is_some() {
+ let new_size = batch_byte_size(&partial.sorted_batch);
+ let size_delta = size.checked_sub(new_size).ok_or_else(|| {
+ DataFusionError::Internal(format!(
+ "The size of the sorted batch is larger than the size of the input batch: {} > {}",
+ size,
+ new_size
+ ))
+ })?;
+ self.shrink(size_delta);
+ self.metrics.mem_used().sub(size_delta);
Review Comment:
@Dandandan, thanks I've added the spill/not-spill one!
--
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] isidentical commented on a diff in pull request #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
isidentical commented on code in PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#discussion_r978356394
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -124,6 +124,20 @@ impl ExternalSorter {
// calls to `timer.done()` below.
let _timer = tracking_metrics.elapsed_compute().timer();
let partial = sort_batch(input, self.schema.clone(), &self.expr, self.fetch)?;
+ // The resulting batch might be smaller than the input batch if there
+ // is an propagated limit.
+ if self.fetch.is_some() {
+ let new_size = batch_byte_size(&partial.sorted_batch);
+ let size_delta = size.checked_sub(new_size).unwrap_or_else(|| {
+ panic!(
Review Comment:
Makes sense, should be fixed now!
--
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 #3593: Prevent over-allocations (and spills) on sorts with a fixed limit
Posted by GitBox <gi...@apache.org>.
Dandandan commented on code in PR #3593:
URL: https://github.com/apache/arrow-datafusion/pull/3593#discussion_r978305031
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -124,6 +124,20 @@ impl ExternalSorter {
// calls to `timer.done()` below.
let _timer = tracking_metrics.elapsed_compute().timer();
let partial = sort_batch(input, self.schema.clone(), &self.expr, self.fetch)?;
+ // The resulting batch might be smaller than the input batch if there
+ // is an propagated limit.
+ if self.fetch.is_some() {
+ let new_size = batch_byte_size(&partial.sorted_batch);
+ let size_delta = size.checked_sub(new_size).unwrap_or_else(|| {
+ panic!(
Review Comment:
Would be better to return an internal error here instead of panicking.
--
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