Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
alamb merged PR #15692: URL: https://github.com/apache/datafusion/pull/15692 -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2043010696
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -759,12 +761,51 @@ impl ExternalSorter {
if self.runtime.disk_manager.tmp_files_enabled() {
let size = self.sort_spill_reservation_bytes;
if self.merge_reservation.size() != size {
-self.merge_reservation.try_resize(size)?;
+self.merge_reservation
+.try_resize(size)
+.map_err(Self::err_with_oom_context)?;
}
}
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch_and_maybe_spill(
+&mut self,
+input: &RecordBatch,
+) -> Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+match self.reservation.try_grow(size) {
+Ok(_) => Ok(()),
+Err(e) => {
+if self.in_mem_batches.is_empty() {
+return Err(Self::err_with_oom_context(e));
+}
+
+// Spill and try again.
+self.sort_and_spill_in_mem_batches().await?;
+self.reservation
+.try_grow(size)
+.map_err(Self::err_with_oom_context)
+}
+}
+}
+
+/// Wraps the error with a context message suggesting settings to tweak.
+/// This is meant to be used with DataFusionError::ResourcesExhausted only.
+fn err_with_oom_context(e: DataFusionError) -> DataFusionError {
+match e {
+DataFusionError::ResourcesExhausted(_) => e.context(format!(
+"Not enough memory to continue external sort. \
+Consider increasing the memory limit, or decreasing
sort_spill_reservation_bytes"
+)),
Review Comment:
Created this: https://github.com/apache/datafusion/issues/15713
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041911140
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -1552,6 +1593,62 @@ mod tests {
Ok(())
}
+#[tokio::test]
+async fn test_batch_reservation_error() -> Result<()> {
+// Pick a memory limit and sort_spill_reservation that make the first
batch reservation fail.
+// These values assume that the ExternalSorter will reserve 800 bytes
for the first batch.
+let expected_batch_reservation = 800;
+let merge_reservation: usize = 0; // Set to 0 for simplicity
+let memory_limit: usize = expected_batch_reservation +
merge_reservation - 1; // Just short of what we need
+
+let session_config =
+
SessionConfig::new().with_sort_spill_reservation_bytes(merge_reservation);
+let runtime = RuntimeEnvBuilder::new()
+.with_memory_limit(memory_limit, 1.0)
+.build_arc()?;
+let task_ctx = Arc::new(
+TaskContext::default()
+.with_session_config(session_config)
+.with_runtime(runtime),
+);
+
+let plan = test::scan_partitioned(1);
+
+// Read the first record batch to assert that our memory limit and
sort_spill_reservation
+// settings trigger the test scenario.
+{
+let mut stream = plan.execute(0, Arc::clone(&task_ctx))?;
+let first_batch = stream.next().await.unwrap()?;
+let batch_reservation =
get_reserved_byte_for_record_batch(&first_batch);
+
+assert_eq!(batch_reservation, expected_batch_reservation);
+assert!(memory_limit < (merge_reservation + batch_reservation));
+}
+
+let sort_exec = Arc::new(SortExec::new(
+LexOrdering::new(vec![PhysicalSortExpr {
+expr: col("i", &plan.schema())?,
+options: SortOptions::default(),
+}]),
+plan,
+));
+
+let result = collect(
+Arc::clone(&sort_exec) as Arc,
+Arc::clone(&task_ctx),
+)
+.await;
+
+let err = result.unwrap_err();
+assert!(
+matches!(err, DataFusionError::ResourcesExhausted(_)),
Review Comment:
Made it pass by doing the unnesting in-line for now
https://github.com/apache/datafusion/pull/15692/commits/d2a076533f5aa80a63d153eee7d92b136f82fe02.
I'll create a separate issue to discuss a potential API extension of
`DataFusionError`.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041912243
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -759,12 +761,51 @@ impl ExternalSorter {
if self.runtime.disk_manager.tmp_files_enabled() {
let size = self.sort_spill_reservation_bytes;
if self.merge_reservation.size() != size {
-self.merge_reservation.try_resize(size)?;
+self.merge_reservation
+.try_resize(size)
+.map_err(Self::err_with_oom_context)?;
}
}
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch_and_maybe_spill(
+&mut self,
+input: &RecordBatch,
+) -> Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+match self.reservation.try_grow(size) {
+Ok(_) => Ok(()),
+Err(e) => {
+if self.in_mem_batches.is_empty() {
+return Err(Self::err_with_oom_context(e));
+}
+
+// Spill and try again.
+self.sort_and_spill_in_mem_batches().await?;
+self.reservation
+.try_grow(size)
+.map_err(Self::err_with_oom_context)
+}
+}
+}
+
+/// Wraps the error with a context message suggesting settings to tweak.
+/// This is meant to be used with DataFusionError::ResourcesExhausted only.
+fn err_with_oom_context(e: DataFusionError) -> DataFusionError {
+match e {
+DataFusionError::ResourcesExhausted(_) => e.context(format!(
+"Not enough memory to continue external sort. \
+Consider increasing the memory limit, or decreasing
sort_spill_reservation_bytes"
+)),
Review Comment:
Works for me :) CI is passing again and I'l create a separate issue 👍
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041254566
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -759,12 +761,51 @@ impl ExternalSorter {
if self.runtime.disk_manager.tmp_files_enabled() {
let size = self.sort_spill_reservation_bytes;
if self.merge_reservation.size() != size {
-self.merge_reservation.try_resize(size)?;
+self.merge_reservation
+.try_resize(size)
+.map_err(Self::err_with_oom_context)?;
}
}
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch_and_maybe_spill(
+&mut self,
+input: &RecordBatch,
+) -> Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+match self.reservation.try_grow(size) {
+Ok(_) => Ok(()),
+Err(e) => {
+if self.in_mem_batches.is_empty() {
+return Err(Self::err_with_oom_context(e));
+}
+
+// Spill and try again.
+self.sort_and_spill_in_mem_batches().await?;
+self.reservation
+.try_grow(size)
+.map_err(Self::err_with_oom_context)
+}
+}
+}
+
+/// Wraps the error with a context message suggesting settings to tweak.
+/// This is meant to be used with DataFusionError::ResourcesExhausted only.
+fn err_with_oom_context(e: DataFusionError) -> DataFusionError {
+match e {
+DataFusionError::ResourcesExhausted(_) => e.context(format!(
+"Not enough memory to continue external sort. \
+Consider increasing the memory limit, or decreasing
sort_spill_reservation_bytes"
+)),
Review Comment:
This would actually be a good argument for adding a function to
`DataFusionError` that allows you to safely compare against any of its nested
errors (I'm thinking something similar to Go's
[`errors.Is`](https://pkg.go.dev/errors#Is)).
Without such an API, it would never be safe to just wrap an error with
context 🤔
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041251862
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -759,12 +761,51 @@ impl ExternalSorter {
if self.runtime.disk_manager.tmp_files_enabled() {
let size = self.sort_spill_reservation_bytes;
if self.merge_reservation.size() != size {
-self.merge_reservation.try_resize(size)?;
+self.merge_reservation
+.try_resize(size)
+.map_err(Self::err_with_oom_context)?;
}
}
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch_and_maybe_spill(
+&mut self,
+input: &RecordBatch,
+) -> Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+match self.reservation.try_grow(size) {
+Ok(_) => Ok(()),
+Err(e) => {
+if self.in_mem_batches.is_empty() {
+return Err(Self::err_with_oom_context(e));
+}
+
+// Spill and try again.
+self.sort_and_spill_in_mem_batches().await?;
+self.reservation
+.try_grow(size)
+.map_err(Self::err_with_oom_context)
+}
+}
+}
+
+/// Wraps the error with a context message suggesting settings to tweak.
+/// This is meant to be used with DataFusionError::ResourcesExhausted only.
+fn err_with_oom_context(e: DataFusionError) -> DataFusionError {
+match e {
+DataFusionError::ResourcesExhausted(_) => e.context(format!(
+"Not enough memory to continue external sort. \
+Consider increasing the memory limit, or decreasing
sort_spill_reservation_bytes"
+)),
Review Comment:
That's a good point! It will be a different type, that's also why the test
is currently failing (see
https://github.com/apache/datafusion/pull/15692/files#r2041251481)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
2010YOUY01 commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041385875
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -765,6 +765,25 @@ impl ExternalSorter {
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch(&mut self, input: &RecordBatch) ->
Result<()> {
Review Comment:
Nice!
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
2010YOUY01 commented on PR #15692: URL: https://github.com/apache/datafusion/pull/15692#issuecomment-2800424077 Thank you, it looks good to me 🙏🏼 Let's make the CI pass, I think we can change the assertion type for simplicity here, and do a separate PR for this utility `DataFusionError::is(&self, target: &DataFusionError) -> bool` if you prefer. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
2010YOUY01 commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041381039
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -759,12 +761,51 @@ impl ExternalSorter {
if self.runtime.disk_manager.tmp_files_enabled() {
let size = self.sort_spill_reservation_bytes;
if self.merge_reservation.size() != size {
-self.merge_reservation.try_resize(size)?;
+self.merge_reservation
+.try_resize(size)
+.map_err(Self::err_with_oom_context)?;
}
}
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch_and_maybe_spill(
+&mut self,
+input: &RecordBatch,
+) -> Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+match self.reservation.try_grow(size) {
+Ok(_) => Ok(()),
+Err(e) => {
+if self.in_mem_batches.is_empty() {
+return Err(Self::err_with_oom_context(e));
+}
+
+// Spill and try again.
+self.sort_and_spill_in_mem_batches().await?;
+self.reservation
+.try_grow(size)
+.map_err(Self::err_with_oom_context)
+}
+}
+}
+
+/// Wraps the error with a context message suggesting settings to tweak.
+/// This is meant to be used with DataFusionError::ResourcesExhausted only.
+fn err_with_oom_context(e: DataFusionError) -> DataFusionError {
+match e {
+DataFusionError::ResourcesExhausted(_) => e.context(format!(
+"Not enough memory to continue external sort. \
+Consider increasing the memory limit, or decreasing
sort_spill_reservation_bytes"
+)),
Review Comment:
I agree it's better to have a utility to assert an error's root cause, like
`Context(Context(ResourceExhausted, ...`'s root cause is
`ResourceExhanusted`, however, I think this is not required for 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041234887
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -765,6 +765,25 @@ impl ExternalSorter {
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch(&mut self, input: &RecordBatch) ->
Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+let result = self.reservation.try_grow(size);
+if result.is_err() {
+if self.in_mem_batches.is_empty() {
+return result;
Review Comment:
This is great! Thanks a lot for the reference 🙏
I added some context with
https://github.com/apache/datafusion/pull/15692/commits/d4e654c176b4eb0ce084029d181f91735d83f375
I dropped the detail about buffered batches because I figured that it's
probably not relevant to the user. Regardless of whether in-memory batches
could be spilled or not, the actionable info is to increase the memory limit.
This also allowed me to reuse the same context in
https://github.com/apache/datafusion/pull/15692/commits/a14e585c490372167814714dfcf6cf20b3d1726b
(and
https://github.com/apache/datafusion/pull/15692/commits/0ee1e8e7543fce169086f78d3ef3186c87bdf810
for completeness). This means all failing memory reservations in the external
sorter are wrapped with context. Let me know if you don't think these are
necessary.
I also added a suggestion to decrease `sort_spill_reservation_bytes`
instead. But I'm not convinced this is good advice to the user. I can leave it
as a comment instead 🤷
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041251862
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -759,12 +761,51 @@ impl ExternalSorter {
if self.runtime.disk_manager.tmp_files_enabled() {
let size = self.sort_spill_reservation_bytes;
if self.merge_reservation.size() != size {
-self.merge_reservation.try_resize(size)?;
+self.merge_reservation
+.try_resize(size)
+.map_err(Self::err_with_oom_context)?;
}
}
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch_and_maybe_spill(
+&mut self,
+input: &RecordBatch,
+) -> Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+match self.reservation.try_grow(size) {
+Ok(_) => Ok(()),
+Err(e) => {
+if self.in_mem_batches.is_empty() {
+return Err(Self::err_with_oom_context(e));
+}
+
+// Spill and try again.
+self.sort_and_spill_in_mem_batches().await?;
+self.reservation
+.try_grow(size)
+.map_err(Self::err_with_oom_context)
+}
+}
+}
+
+/// Wraps the error with a context message suggesting settings to tweak.
+/// This is meant to be used with DataFusionError::ResourcesExhausted only.
+fn err_with_oom_context(e: DataFusionError) -> DataFusionError {
+match e {
+DataFusionError::ResourcesExhausted(_) => e.context(format!(
+"Not enough memory to continue external sort. \
+Consider increasing the memory limit, or decreasing
sort_spill_reservation_bytes"
+)),
Review Comment:
That's a good point! It will be a different type (or enum variant), that's
also why the test is currently failing (see
https://github.com/apache/datafusion/pull/15692/files#r2041251481)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041254566
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -759,12 +761,51 @@ impl ExternalSorter {
if self.runtime.disk_manager.tmp_files_enabled() {
let size = self.sort_spill_reservation_bytes;
if self.merge_reservation.size() != size {
-self.merge_reservation.try_resize(size)?;
+self.merge_reservation
+.try_resize(size)
+.map_err(Self::err_with_oom_context)?;
}
}
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch_and_maybe_spill(
+&mut self,
+input: &RecordBatch,
+) -> Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+match self.reservation.try_grow(size) {
+Ok(_) => Ok(()),
+Err(e) => {
+if self.in_mem_batches.is_empty() {
+return Err(Self::err_with_oom_context(e));
+}
+
+// Spill and try again.
+self.sort_and_spill_in_mem_batches().await?;
+self.reservation
+.try_grow(size)
+.map_err(Self::err_with_oom_context)
+}
+}
+}
+
+/// Wraps the error with a context message suggesting settings to tweak.
+/// This is meant to be used with DataFusionError::ResourcesExhausted only.
+fn err_with_oom_context(e: DataFusionError) -> DataFusionError {
+match e {
+DataFusionError::ResourcesExhausted(_) => e.context(format!(
+"Not enough memory to continue external sort. \
+Consider increasing the memory limit, or decreasing
sort_spill_reservation_bytes"
+)),
Review Comment:
This would actually be a good argument for adding a function to
`DataFusionError` that allows you to safely compare against any of its nested
errors (I'm thinking something similar to Go's
[`errors.Is`](https://pkg.go.dev/errors#Is).
Without such an API, it would never be safe to just wrap an error with
context 🤔
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041251481
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -1552,6 +1593,62 @@ mod tests {
Ok(())
}
+#[tokio::test]
+async fn test_batch_reservation_error() -> Result<()> {
+// Pick a memory limit and sort_spill_reservation that make the first
batch reservation fail.
+// These values assume that the ExternalSorter will reserve 800 bytes
for the first batch.
+let expected_batch_reservation = 800;
+let merge_reservation: usize = 0; // Set to 0 for simplicity
+let memory_limit: usize = expected_batch_reservation +
merge_reservation - 1; // Just short of what we need
+
+let session_config =
+
SessionConfig::new().with_sort_spill_reservation_bytes(merge_reservation);
+let runtime = RuntimeEnvBuilder::new()
+.with_memory_limit(memory_limit, 1.0)
+.build_arc()?;
+let task_ctx = Arc::new(
+TaskContext::default()
+.with_session_config(session_config)
+.with_runtime(runtime),
+);
+
+let plan = test::scan_partitioned(1);
+
+// Read the first record batch to assert that our memory limit and
sort_spill_reservation
+// settings trigger the test scenario.
+{
+let mut stream = plan.execute(0, Arc::clone(&task_ctx))?;
+let first_batch = stream.next().await.unwrap()?;
+let batch_reservation =
get_reserved_byte_for_record_batch(&first_batch);
+
+assert_eq!(batch_reservation, expected_batch_reservation);
+assert!(memory_limit < (merge_reservation + batch_reservation));
+}
+
+let sort_exec = Arc::new(SortExec::new(
+LexOrdering::new(vec![PhysicalSortExpr {
+expr: col("i", &plan.schema())?,
+options: SortOptions::default(),
+}]),
+plan,
+));
+
+let result = collect(
+Arc::clone(&sort_exec) as Arc,
+Arc::clone(&task_ctx),
+)
+.await;
+
+let err = result.unwrap_err();
+assert!(
+matches!(err, DataFusionError::ResourcesExhausted(_)),
Review Comment:
Note, this is now failing because it gets a `DataFusionError::Context`
instead. I'm still pondering whether to ~unwrap~ unnest the error here, or add
something like an `DataFusionError::is(&self, target: &DataFusionError) ->
bool` to match the nested error.
I don't think matching against `Context` instead would be interesting here.
I'd rather test the cause of the error, less so whether it has added context or
not.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041251481
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -1552,6 +1593,62 @@ mod tests {
Ok(())
}
+#[tokio::test]
+async fn test_batch_reservation_error() -> Result<()> {
+// Pick a memory limit and sort_spill_reservation that make the first
batch reservation fail.
+// These values assume that the ExternalSorter will reserve 800 bytes
for the first batch.
+let expected_batch_reservation = 800;
+let merge_reservation: usize = 0; // Set to 0 for simplicity
+let memory_limit: usize = expected_batch_reservation +
merge_reservation - 1; // Just short of what we need
+
+let session_config =
+
SessionConfig::new().with_sort_spill_reservation_bytes(merge_reservation);
+let runtime = RuntimeEnvBuilder::new()
+.with_memory_limit(memory_limit, 1.0)
+.build_arc()?;
+let task_ctx = Arc::new(
+TaskContext::default()
+.with_session_config(session_config)
+.with_runtime(runtime),
+);
+
+let plan = test::scan_partitioned(1);
+
+// Read the first record batch to assert that our memory limit and
sort_spill_reservation
+// settings trigger the test scenario.
+{
+let mut stream = plan.execute(0, Arc::clone(&task_ctx))?;
+let first_batch = stream.next().await.unwrap()?;
+let batch_reservation =
get_reserved_byte_for_record_batch(&first_batch);
+
+assert_eq!(batch_reservation, expected_batch_reservation);
+assert!(memory_limit < (merge_reservation + batch_reservation));
+}
+
+let sort_exec = Arc::new(SortExec::new(
+LexOrdering::new(vec![PhysicalSortExpr {
+expr: col("i", &plan.schema())?,
+options: SortOptions::default(),
+}]),
+plan,
+));
+
+let result = collect(
+Arc::clone(&sort_exec) as Arc,
+Arc::clone(&task_ctx),
+)
+.await;
+
+let err = result.unwrap_err();
+assert!(
+matches!(err, DataFusionError::ResourcesExhausted(_)),
Review Comment:
Note, this is now failing because it gets a `DataFusionError::Context`
instead. I'm still pondering whether to unwrap the error here, or add something
like an `DataFusionError::is(&self, target: &DataFusionError) -> bool` to match
the nested error.
I don't think matching against `Context` instead would be interesting here.
I'd rather test the cause of the error, less so whether it has added context or
not.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
rluvaton commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041239380
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -759,12 +761,51 @@ impl ExternalSorter {
if self.runtime.disk_manager.tmp_files_enabled() {
let size = self.sort_spill_reservation_bytes;
if self.merge_reservation.size() != size {
-self.merge_reservation.try_resize(size)?;
+self.merge_reservation
+.try_resize(size)
+.map_err(Self::err_with_oom_context)?;
}
}
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch_and_maybe_spill(
+&mut self,
+input: &RecordBatch,
+) -> Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+match self.reservation.try_grow(size) {
+Ok(_) => Ok(()),
+Err(e) => {
+if self.in_mem_batches.is_empty() {
+return Err(Self::err_with_oom_context(e));
+}
+
+// Spill and try again.
+self.sort_and_spill_in_mem_batches().await?;
+self.reservation
+.try_grow(size)
+.map_err(Self::err_with_oom_context)
+}
+}
+}
+
+/// Wraps the error with a context message suggesting settings to tweak.
+/// This is meant to be used with DataFusionError::ResourcesExhausted only.
+fn err_with_oom_context(e: DataFusionError) -> DataFusionError {
+match e {
+DataFusionError::ResourcesExhausted(_) => e.context(format!(
+"Not enough memory to continue external sort. \
+Consider increasing the memory limit, or decreasing
sort_spill_reservation_bytes"
+)),
Review Comment:
If I match on the result of this function will it still be resources
exhausted or different type, if different type this can break usages that
depend on that, no?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041234887
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -765,6 +765,25 @@ impl ExternalSorter {
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch(&mut self, input: &RecordBatch) ->
Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+let result = self.reservation.try_grow(size);
+if result.is_err() {
+if self.in_mem_batches.is_empty() {
+return result;
Review Comment:
This is great! Thanks a lot for the reference 🙏
I added some context with
https://github.com/apache/datafusion/pull/15692/commits/d4e654c176b4eb0ce084029d181f91735d83f375
I dropped the detail about buffered batches because I figured that it's
probably not relevant to the user. Regardless of whether in-memory batches
could be spilled or not, the actionable info is to increase the memory limit.
This also allowed me to reuse the same context in
https://github.com/apache/datafusion/pull/15692/commits/a14e585c490372167814714dfcf6cf20b3d1726b
(let me know if you don't think this is necessary).
I also added a suggestion to decrease `sort_spill_reservation_bytes`
instead. But I'm not convinced this is good advice to the user. I can leave it
as a comment instead 🤷
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041222385
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -765,6 +765,25 @@ impl ExternalSorter {
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch(&mut self, input: &RecordBatch) ->
Result<()> {
Review Comment:
I renamed it to `reserve_memory_for_batch_and_maybe_spill` in
https://github.com/apache/datafusion/pull/15692/commits/ba247814b6c12f3e0c8dd19bc45a2e635510aa81
because I think `reserve_memory_for_batch_or_spill` suggests that it only does
one of those things, wdyt?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041181467
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -529,6 +523,12 @@ impl ExternalSorter {
/// Sorts the in-memory batches and merges them into a single sorted run,
then writes
/// the result to spill files.
async fn sort_and_spill_in_mem_batches(&mut self) -> Result<()> {
+if self.in_mem_batches.is_empty() {
Review Comment:
Makes a lot of sense to me! Its now returning an error with
https://github.com/apache/datafusion/pull/15692/commits/c3e5388910916f6093fedffa460bd91c07f410f7
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2041181467
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -529,6 +523,12 @@ impl ExternalSorter {
/// Sorts the in-memory batches and merges them into a single sorted run,
then writes
/// the result to spill files.
async fn sort_and_spill_in_mem_batches(&mut self) -> Result<()> {
+if self.in_mem_batches.is_empty() {
Review Comment:
Makes a lot of sense to me! Its now returning an error with
https://github.com/apache/datafusion/pull/15692/commits/5c3502d6fbe3136f0435afb90fa99b5af742d914
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
2010YOUY01 commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2040878250
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -529,6 +523,12 @@ impl ExternalSorter {
/// Sorts the in-memory batches and merges them into a single sorted run,
then writes
/// the result to spill files.
async fn sort_and_spill_in_mem_batches(&mut self) -> Result<()> {
+if self.in_mem_batches.is_empty() {
Review Comment:
> Now we handle this case in two places. I can remove the handling here
unless a more defensive approach is desired.
I think it's better to be more defensive, I'd prefer to throw an internal
error saying that we must assume the buffer is not empty when calling this
function.
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -1552,6 +1571,62 @@ mod tests {
Ok(())
}
+#[tokio::test]
+async fn test_batch_reservation_error() -> Result<()> {
Review Comment:
👍🏼
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -765,6 +765,25 @@ impl ExternalSorter {
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch(&mut self, input: &RecordBatch) ->
Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+let result = self.reservation.try_grow(size);
+if result.is_err() {
+if self.in_mem_batches.is_empty() {
+return result;
Review Comment:
It makes sense that we should preserve the raw resource error. In this case,
`ContextError` comes in handy:
https://github.com/apache/datafusion/blob/0b01fdf7f02f9097c319204058576f420b9790b4/datafusion/common/src/error.rs#L132
It can add extra notes to the original error message, to make it more
understandable.
Here, the extra note can be like "When the memory limit is reached and there
are no buffered batches to spill, external sort cannot continue. Consider
setting a higher memory limit."
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -765,6 +765,25 @@ impl ExternalSorter {
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch(&mut self, input: &RecordBatch) ->
Result<()> {
Review Comment:
nit: I think a more obvious name for this function would be
`reserve_memory_for_batch_or_spill`
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2040700866
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -765,6 +765,25 @@ impl ExternalSorter {
Ok(())
}
+
+/// Reserves memory to be able to accommodate the given batch.
+/// If memory is scarce, tries to spill current in-memory batches to disk
first.
+async fn reserve_memory_for_batch(&mut self, input: &RecordBatch) ->
Result<()> {
+let size = get_reserved_byte_for_record_batch(input);
+
+let result = self.reservation.try_grow(size);
+if result.is_err() {
+if self.in_mem_batches.is_empty() {
+return result;
Review Comment:
@2010YOUY01 since you commented on the issue
https://github.com/apache/datafusion/issues/15675#issuecomment-2796177594 with
suggestions:
> I think the fix (for a more user-friendly error message) would be
>
> 1. When sort_spill_reservation_byte reservation failed, the error message
should include considering set the memory limit larger than this reservation
memory size
> 2. For the code path triggered this internal error, return an
ExecutionError instead with a similar error message
This code now emits a
[`DataFusionError::ResourcesExhausted`](https://github.com/apache/datafusion/blob/0b01fdf7f02f9097c319204058576f420b9790b4/datafusion/common/src/error.rs#L126)
instead of the suggested
[`DataFusionError::Execution`](https://github.com/apache/datafusion/blob/0b01fdf7f02f9097c319204058576f420b9790b4/datafusion/common/src/error.rs#L117).
I think this makes sense, given the descriptions of the errors.
E.g. Execution error doesn't seem to apply:
> This error is returned when an error happens during execution due to a
malformed input.
but ResourcesExhausted does:
> This error is thrown when a consumer cannot acquire additional memory
---
I still like the idea of suggesting to bump the memory limit in the error
message.
However, since the code simply bubbles up the error emitted by the memory
pool's
[`insufficient_capacity_err`](https://github.com/apache/datafusion/blob/0b01fdf7f02f9097c319204058576f420b9790b4/datafusion/execution/src/memory_pool/pool.rs#L249)
macro, we would either need to map it, or change the macro's message.
I don't think mapping makes sense here because the error is generally used
for reservation-related issues - I don't think this is a special case.
Changing the error message would probably require a separate discussion.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Fix internal error in sort when hitting memory limit [datafusion]
DerGut commented on code in PR #15692:
URL: https://github.com/apache/datafusion/pull/15692#discussion_r2040698187
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -529,6 +523,12 @@ impl ExternalSorter {
/// Sorts the in-memory batches and merges them into a single sorted run,
then writes
/// the result to spill files.
async fn sort_and_spill_in_mem_batches(&mut self) -> Result<()> {
+if self.in_mem_batches.is_empty() {
Review Comment:
This was my first version to fix the internal error. Later I decided to move
the logic to the caller (`ExternalSorter::insert_batch`) because the name
`ExternalSorter::sort_and_spill_in_mem_batches` kind of implies that it only
makes sense to call when there are in-mem batches.
Now we handle this case in two places. I can remove the handling here unless
a more defensive approach is desired.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
