You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by yt...@apache.org on 2025/05/16 10:04:51 UTC
(datafusion) branch main updated: Make error msg for oom human readable (#16050)
This is an automated email from the ASF dual-hosted git repository.
ytyou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 69dfe6c499 Make error msg for oom human readable (#16050)
69dfe6c499 is described below
commit 69dfe6c499d39563f4e6d9835fcdf3793f7d98c8
Author: ding-young <ls...@snu.ac.kr>
AuthorDate: Fri May 16 19:04:46 2025 +0900
Make error msg for oom human readable (#16050)
---
datafusion/core/tests/memory_limit/mod.rs | 24 +++++------
datafusion/execution/src/memory_pool/pool.rs | 49 ++++++++++++----------
datafusion/physical-plan/src/joins/cross_join.rs | 2 +-
datafusion/physical-plan/src/joins/hash_join.rs | 8 ++--
.../physical-plan/src/joins/nested_loop_join.rs | 2 +-
5 files changed, 46 insertions(+), 39 deletions(-)
diff --git a/datafusion/core/tests/memory_limit/mod.rs b/datafusion/core/tests/memory_limit/mod.rs
index 5957919824..cf835ac1a5 100644
--- a/datafusion/core/tests/memory_limit/mod.rs
+++ b/datafusion/core/tests/memory_limit/mod.rs
@@ -84,7 +84,7 @@ async fn group_by_none() {
TestCase::new()
.with_query("select median(request_bytes) from t")
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: AggregateStream"
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n AggregateStream"
])
.with_memory_limit(2_000)
.run()
@@ -96,7 +96,7 @@ async fn group_by_row_hash() {
TestCase::new()
.with_query("select count(*) from t GROUP BY response_bytes")
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: GroupedHashAggregateStream"
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n GroupedHashAggregateStream"
])
.with_memory_limit(2_000)
.run()
@@ -109,7 +109,7 @@ async fn group_by_hash() {
// group by dict column
.with_query("select count(*) from t GROUP BY service, host, pod, container")
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: GroupedHashAggregateStream"
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n GroupedHashAggregateStream"
])
.with_memory_limit(1_000)
.run()
@@ -122,7 +122,7 @@ async fn join_by_key_multiple_partitions() {
TestCase::new()
.with_query("select t1.* from t t1 JOIN t t2 ON t1.service = t2.service")
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: HashJoinInput",
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n HashJoinInput",
])
.with_memory_limit(1_000)
.with_config(config)
@@ -136,7 +136,7 @@ async fn join_by_key_single_partition() {
TestCase::new()
.with_query("select t1.* from t t1 JOIN t t2 ON t1.service = t2.service")
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: HashJoinInput",
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n HashJoinInput",
])
.with_memory_limit(1_000)
.with_config(config)
@@ -149,7 +149,7 @@ async fn join_by_expression() {
TestCase::new()
.with_query("select t1.* from t t1 JOIN t t2 ON t1.service != t2.service")
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: NestedLoopJoinLoad[0]",
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n NestedLoopJoinLoad[0]",
])
.with_memory_limit(1_000)
.run()
@@ -161,7 +161,7 @@ async fn cross_join() {
TestCase::new()
.with_query("select t1.*, t2.* from t t1 CROSS JOIN t t2")
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: CrossJoinExec",
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n CrossJoinExec",
])
.with_memory_limit(1_000)
.run()
@@ -217,7 +217,7 @@ async fn symmetric_hash_join() {
"select t1.* from t t1 JOIN t t2 ON t1.pod = t2.pod AND t1.time = t2.time",
)
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: SymmetricHashJoinStream",
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n SymmetricHashJoinStream",
])
.with_memory_limit(1_000)
.with_scenario(Scenario::AccessLogStreaming)
@@ -235,7 +235,7 @@ async fn sort_preserving_merge() {
// so only a merge is needed
.with_query("select * from t ORDER BY a ASC NULLS LAST, b ASC NULLS LAST LIMIT 10")
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: SortPreservingMergeExec",
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n SortPreservingMergeExec",
])
// provide insufficient memory to merge
.with_memory_limit(partition_size / 2)
@@ -315,7 +315,7 @@ async fn sort_spill_reservation() {
test.clone()
.with_expected_errors(vec![
"Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:",
- "bytes for ExternalSorterMerge",
+ "B for ExternalSorterMerge",
])
.with_config(config)
.run()
@@ -344,7 +344,7 @@ async fn oom_recursive_cte() {
SELECT * FROM nodes;",
)
.with_expected_errors(vec![
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: RecursiveQuery",
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n RecursiveQuery",
])
.with_memory_limit(2_000)
.run()
@@ -396,7 +396,7 @@ async fn oom_with_tracked_consumer_pool() {
.with_expected_errors(vec![
"Failed to allocate additional",
"for ParquetSink(ArrowColumnWriter)",
- "Additional allocation failed with top memory consumers (across reservations) as: ParquetSink(ArrowColumnWriter)"
+ "Additional allocation failed with top memory consumers (across reservations) as:\n ParquetSink(ArrowColumnWriter)"
])
.with_memory_pool(Arc::new(
TrackConsumersPool::new(
diff --git a/datafusion/execution/src/memory_pool/pool.rs b/datafusion/execution/src/memory_pool/pool.rs
index c975b69fd9..32ad3a5863 100644
--- a/datafusion/execution/src/memory_pool/pool.rs
+++ b/datafusion/execution/src/memory_pool/pool.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::memory_pool::{MemoryConsumer, MemoryLimit, MemoryPool, MemoryReservation};
+use crate::memory_pool::{
+ human_readable_size, MemoryConsumer, MemoryLimit, MemoryPool, MemoryReservation,
+};
use datafusion_common::HashMap;
use datafusion_common::{resources_datafusion_err, DataFusionError, Result};
use log::debug;
@@ -258,7 +260,8 @@ fn insufficient_capacity_err(
additional: usize,
available: usize,
) -> DataFusionError {
- resources_datafusion_err!("Failed to allocate additional {} bytes for {} with {} bytes already allocated for this reservation - {} bytes remain available for the total pool", additional, reservation.registration.consumer.name, reservation.size, available)
+ resources_datafusion_err!("Failed to allocate additional {} for {} with {} already allocated for this reservation - {} remain available for the total pool",
+ human_readable_size(additional), reservation.registration.consumer.name, human_readable_size(reservation.size), human_readable_size(available))
}
#[derive(Debug)]
@@ -340,10 +343,14 @@ impl<I: MemoryPool> TrackConsumersPool<I> {
consumers[0..std::cmp::min(top, consumers.len())]
.iter()
.map(|((id, name, can_spill), size)| {
- format!("{name}#{id}(can spill: {can_spill}) consumed {size} bytes")
+ format!(
+ " {name}#{id}(can spill: {can_spill}) consumed {}",
+ human_readable_size(*size)
+ )
})
.collect::<Vec<_>>()
- .join(", ")
+ .join(",\n")
+ + "."
}
}
@@ -430,7 +437,7 @@ fn provide_top_memory_consumers_to_error_msg(
error_msg: String,
top_consumers: String,
) -> String {
- format!("Additional allocation failed with top memory consumers (across reservations) as: {top_consumers}. Error: {error_msg}")
+ format!("Additional allocation failed with top memory consumers (across reservations) as:\n{top_consumers}\nError: {error_msg}")
}
#[cfg(test)]
@@ -456,10 +463,10 @@ mod tests {
assert_eq!(pool.reserved(), 4000);
let err = r2.try_grow(1).unwrap_err().strip_backtrace();
- assert_eq!(err, "Resources exhausted: Failed to allocate additional 1 bytes for r2 with 2000 bytes already allocated for this reservation - 0 bytes remain available for the total pool");
+ assert_eq!(err, "Resources exhausted: Failed to allocate additional 1.0 B for r2 with 2000.0 B already allocated for this reservation - 0.0 B remain available for the total pool");
let err = r2.try_grow(1).unwrap_err().strip_backtrace();
- assert_eq!(err, "Resources exhausted: Failed to allocate additional 1 bytes for r2 with 2000 bytes already allocated for this reservation - 0 bytes remain available for the total pool");
+ assert_eq!(err, "Resources exhausted: Failed to allocate additional 1.0 B for r2 with 2000.0 B already allocated for this reservation - 0.0 B remain available for the total pool");
r1.shrink(1990);
r2.shrink(2000);
@@ -484,12 +491,12 @@ mod tests {
.register(&pool);
let err = r3.try_grow(70).unwrap_err().strip_backtrace();
- assert_eq!(err, "Resources exhausted: Failed to allocate additional 70 bytes for r3 with 0 bytes already allocated for this reservation - 40 bytes remain available for the total pool");
+ assert_eq!(err, "Resources exhausted: Failed to allocate additional 70.0 B for r3 with 0.0 B already allocated for this reservation - 40.0 B remain available for the total pool");
//Shrinking r2 to zero doesn't allow a3 to allocate more than 45
r2.free();
let err = r3.try_grow(70).unwrap_err().strip_backtrace();
- assert_eq!(err, "Resources exhausted: Failed to allocate additional 70 bytes for r3 with 0 bytes already allocated for this reservation - 40 bytes remain available for the total pool");
+ assert_eq!(err, "Resources exhausted: Failed to allocate additional 70.0 B for r3 with 0.0 B already allocated for this reservation - 40.0 B remain available for the total pool");
// But dropping r2 does
drop(r2);
@@ -502,7 +509,7 @@ mod tests {
let mut r4 = MemoryConsumer::new("s4").register(&pool);
let err = r4.try_grow(30).unwrap_err().strip_backtrace();
- assert_eq!(err, "Resources exhausted: Failed to allocate additional 30 bytes for s4 with 0 bytes already allocated for this reservation - 20 bytes remain available for the total pool");
+ assert_eq!(err, "Resources exhausted: Failed to allocate additional 30.0 B for s4 with 0.0 B already allocated for this reservation - 20.0 B remain available for the total pool");
}
#[test]
@@ -539,7 +546,7 @@ mod tests {
// Test: reports if new reservation causes error
// using the previously set sizes for other consumers
let mut r5 = MemoryConsumer::new("r5").register(&pool);
- let expected = format!("Additional allocation failed with top memory consumers (across reservations) as: r1#{}(can spill: false) consumed 50 bytes, r3#{}(can spill: false) consumed 20 bytes, r2#{}(can spill: false) consumed 15 bytes. Error: Failed to allocate additional 150 bytes for r5 with 0 bytes already allocated for this reservation - 5 bytes remain available for the total pool", r1.consumer().id(), r3.consumer().id(), r2.consumer().id());
+ let expected = format!("Additional allocation failed with top memory consumers (across reservations) as:\n r1#{}(can spill: false) consumed 50.0 B,\n r3#{}(can spill: false) consumed 20.0 B,\n r2#{}(can spill: false) consumed 15.0 B.\nError: Failed to allocate additional 150.0 B for r5 with 0.0 B already allocated for this reservation - 5.0 B remain available for the total pool", r1.consumer().id(), r3.consumer().id(), r2.consumer().id());
let res = r5.try_grow(150);
assert!(
matches!(
@@ -561,7 +568,7 @@ mod tests {
// Test: see error message when no consumers recorded yet
let mut r0 = MemoryConsumer::new(same_name).register(&pool);
- let expected = format!("Additional allocation failed with top memory consumers (across reservations) as: foo#{}(can spill: false) consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated for this reservation - 100 bytes remain available for the total pool", r0.consumer().id());
+ let expected = format!("Additional allocation failed with top memory consumers (across reservations) as:\n foo#{}(can spill: false) consumed 0.0 B.\nError: Failed to allocate additional 150.0 B for foo with 0.0 B already allocated for this reservation - 100.0 B remain available for the total pool", r0.consumer().id());
let res = r0.try_grow(150);
assert!(
matches!(
@@ -579,7 +586,7 @@ mod tests {
let mut r1 = new_consumer_same_name.register(&pool);
// TODO: the insufficient_capacity_err() message is per reservation, not per consumer.
// a followup PR will clarify this message "0 bytes already allocated for this reservation"
- let expected = format!("Additional allocation failed with top memory consumers (across reservations) as: foo#{}(can spill: false) consumed 10 bytes, foo#{}(can spill: false) consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated for this reservation - 90 bytes remain available for the total pool", r0.consumer().id(), r1.consumer().id());
+ let expected = format!("Additional allocation failed with top memory consumers (across reservations) as:\n foo#{}(can spill: false) consumed 10.0 B,\n foo#{}(can spill: false) consumed 0.0 B.\nError: Failed to allocate additional 150.0 B for foo with 0.0 B already allocated for this reservation - 90.0 B remain available for the total pool", r0.consumer().id(), r1.consumer().id());
let res = r1.try_grow(150);
assert!(
matches!(
@@ -591,14 +598,14 @@ mod tests {
// Test: will accumulate size changes per consumer, not per reservation
r1.grow(20);
- let expected = format!("Additional allocation failed with top memory consumers (across reservations) as: foo#{}(can spill: false) consumed 20 bytes, foo#{}(can spill: false) consumed 10 bytes. Error: Failed to allocate additional 150 bytes for foo with 20 bytes already allocated for this reservation - 70 bytes remain available for the total pool", r1.consumer().id(), r0.consumer().id());
+ let expected = format!("Additional allocation failed with top memory consumers (across reservations) as:\n foo#{}(can spill: false) consumed 20.0 B,\n foo#{}(can spill: false) consumed 10.0 B.\nError: Failed to allocate additional 150.0 B for foo with 20.0 B already allocated for this reservation - 70.0 B remain available for the total pool", r1.consumer().id(), r0.consumer().id());
let res = r1.try_grow(150);
assert!(
matches!(
&res,
Err(DataFusionError::ResourcesExhausted(ref e)) if e.to_string().contains(&expected)
),
- "should provide proper error for 2 consumers(one foo=20 bytes, another foo=10 bytes, available=70), instead found {res:?}"
+ "should provide proper error for 2 consumers(one foo=20.0 B, another foo=10.0 B, available=70.0 B), instead found {res:?}"
);
// Test: different hashed consumer, (even with the same name),
@@ -606,7 +613,7 @@ mod tests {
let consumer_with_same_name_but_different_hash =
MemoryConsumer::new(same_name).with_can_spill(true);
let mut r2 = consumer_with_same_name_but_different_hash.register(&pool);
- let expected = format!("Additional allocation failed with top memory consumers (across reservations) as: foo#{}(can spill: false) consumed 20 bytes, foo#{}(can spill: false) consumed 10 bytes, foo#{}(can spill: true) consumed 0 bytes. Error: Failed to allocate additional 150 bytes for foo with 0 bytes already allocated for this reservation - 70 bytes remain available for the total pool", r1.consumer().id(), r0.consumer().id(), r2.consumer().id());
+ let expected = format!("Additional allocation failed with top memory consumers (across reservations) as:\n foo#{}(can spill: false) consumed 20.0 B,\n foo#{}(can spill: false) consumed 10.0 B,\n foo#{}(can spill: true) consumed 0.0 B.\nError: Failed to allocate additional 150.0 B for foo with 0.0 B already allocated for this reservation - 70.0 B remain available for the total pool", r1.consumer().id(), r0.consumer().id(), r2.consumer().id());
let res = r2.try_grow(150);
assert!(
matches!(
@@ -627,7 +634,7 @@ mod tests {
let mut r1 = r1_consumer.register(&pool);
r1.grow(20);
- let expected = format!("Additional allocation failed with top memory consumers (across reservations) as: r1#{}(can spill: false) consumed 20 bytes, r0#{}(can spill: false) consumed 10 bytes. Error: Failed to allocate additional 150 bytes for r0 with 10 bytes already allocated for this reservation - 70 bytes remain available for the total pool", r1.consumer().id(), r0.consumer().id());
+ let expected = format!("Additional allocation failed with top memory consumers (across reservations) as:\n r1#{}(can spill: false) consumed 20.0 B,\n r0#{}(can spill: false) consumed 10.0 B.\nError: Failed to allocate additional 150.0 B for r0 with 10.0 B already allocated for this reservation - 70.0 B remain available for the total pool", r1.consumer().id(), r0.consumer().id());
let res = r0.try_grow(150);
assert!(
matches!(
@@ -640,7 +647,7 @@ mod tests {
// Test: unregister one
// only the remaining one should be listed
drop(r1);
- let expected_consumers = format!("Additional allocation failed with top memory consumers (across reservations) as: r0#{}(can spill: false) consumed 10 bytes", r0.consumer().id());
+ let expected_consumers = format!("Additional allocation failed with top memory consumers (across reservations) as:\n r0#{}(can spill: false) consumed 10.0 B", r0.consumer().id());
let res = r0.try_grow(150);
assert!(
matches!(
@@ -652,7 +659,7 @@ mod tests {
// Test: actual message we see is the `available is 70`. When it should be `available is 90`.
// This is because the pool.shrink() does not automatically occur within the inner_pool.deregister().
- let expected_90_available = "Failed to allocate additional 150 bytes for r0 with 10 bytes already allocated for this reservation - 90 bytes remain available for the total pool";
+ let expected_90_available = "Failed to allocate additional 150.0 B for r0 with 10.0 B already allocated for this reservation - 90.0 B remain available for the total pool";
let res = r0.try_grow(150);
assert!(
matches!(
@@ -664,7 +671,7 @@ mod tests {
// Test: the registration needs to free itself (or be dropped),
// for the proper error message
- let expected_90_available = "Failed to allocate additional 150 bytes for r0 with 10 bytes already allocated for this reservation - 90 bytes remain available for the total pool";
+ let expected_90_available = "Failed to allocate additional 150.0 B for r0 with 10.0 B already allocated for this reservation - 90.0 B remain available for the total pool";
let res = r0.try_grow(150);
assert!(
matches!(
@@ -713,7 +720,7 @@ mod tests {
.unwrap();
// Test: can get runtime metrics, even without an error thrown
- let expected = format!("r3#{}(can spill: false) consumed 45 bytes, r1#{}(can spill: false) consumed 20 bytes", r3.consumer().id(), r1.consumer().id());
+ let expected = format!(" r3#{}(can spill: false) consumed 45.0 B,\n r1#{}(can spill: false) consumed 20.0 B.", r3.consumer().id(), r1.consumer().id());
let res = downcasted.report_top(2);
assert_eq!(
res, expected,
diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs
index cbc38f1149..4d8c48c659 100644
--- a/datafusion/physical-plan/src/joins/cross_join.rs
+++ b/datafusion/physical-plan/src/joins/cross_join.rs
@@ -874,7 +874,7 @@ mod tests {
assert_contains!(
err.to_string(),
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: CrossJoinExec"
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n CrossJoinExec"
);
Ok(())
diff --git a/datafusion/physical-plan/src/joins/hash_join.rs b/datafusion/physical-plan/src/joins/hash_join.rs
index 6eb435d558..c7719907e3 100644
--- a/datafusion/physical-plan/src/joins/hash_join.rs
+++ b/datafusion/physical-plan/src/joins/hash_join.rs
@@ -4066,12 +4066,12 @@ mod tests {
// Asserting that operator-level reservation attempting to overallocate
assert_contains!(
err.to_string(),
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: HashJoinInput"
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n HashJoinInput"
);
assert_contains!(
err.to_string(),
- "Failed to allocate additional 120 bytes for HashJoinInput"
+ "Failed to allocate additional 120.0 B for HashJoinInput"
);
}
@@ -4147,13 +4147,13 @@ mod tests {
// Asserting that stream-level reservation attempting to overallocate
assert_contains!(
err.to_string(),
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: HashJoinInput[1]"
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n HashJoinInput[1]"
);
assert_contains!(
err.to_string(),
- "Failed to allocate additional 120 bytes for HashJoinInput[1]"
+ "Failed to allocate additional 120.0 B for HashJoinInput[1]"
);
}
diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs
index c3178f0d1f..f87cf3d886 100644
--- a/datafusion/physical-plan/src/joins/nested_loop_join.rs
+++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs
@@ -1513,7 +1513,7 @@ pub(crate) mod tests {
assert_contains!(
err.to_string(),
- "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: NestedLoopJoinLoad[0]"
+ "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as:\n NestedLoopJoinLoad[0]"
);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org