JonSnow1807 commented on code in PR #7607:
URL: https://github.com/apache/datafusion-comet/pull/7607#discussion_r4555669486
##########
docs/source/contributor-guide/memory_management.md:
##########
@@ -332,6 +332,13 @@ JNI, which goes through Spark's ordinary
`TaskMemoryManager`. That means:
granted. While overcommit is outstanding, `try_grow` asks Spark for it on
top of the request and
is refused unless Spark can cover both, so operators spill until the debt is
repaid. Both pools'
`Display` output and their `try_grow` errors report the current overcommit.
+- When Spark throws while acquiring memory, for instance because a spill of
another consumer
+ failed, every later `try_grow` of the task that asks for memory fails,
before the fair pool
+ checks its own limits. A `grow` still records the full amount as overcommit,
and when it was the
+ first to fail, the first `try_grow` after it is handed the exception itself.
The others fail with
+ its message. That error is not `ResourcesExhausted`, so an operator that
checks for a refusal,
+ such as the shuffle repartitioner, fails the task rather than spilling, and
one that takes any
+ error for a refusal and asks again fails when it asks again.
Review Comment:
Agreed. The doc now says a failed `grow` with no later `try_grow` is still
only logged, and the PR is now `Part of #7244`. The end-of-stream check will be
a follow-up.
##########
native/core/src/execution/memory_pools/spark_memory.rs:
##########
@@ -163,6 +195,29 @@ impl SparkMemory {
self.overcommit.load(Relaxed)
}
+ /// Fails once a request to Spark has failed: the first time after a
failed [`Self::acquire`]
+ /// with its error, and otherwise with the message of the first request
that failed.
+ pub(super) fn check_failure(&self) -> CometResult<()> {
Review Comment:
Done: the throwable is now kept (in an `Arc`), so every later request fails
with the same exception. A new Spark test shows the task now ends with
`SparkOutOfMemoryError` instead of `CometNativeException`.
##########
native/shuffle/src/partitioners/multi_partition.rs:
##########
@@ -176,6 +176,15 @@ fn count_new_buffers(batch: &RecordBatch, seen: &mut
HashSet<usize>) -> usize {
total
}
+/// Whether `e`, from growing the spill reservation, is the pool refusing the
memory, which every
+/// pool reports as `ResourcesExhausted` and the repartitioner answers by
spilling. A pool backed
+/// by Spark can also fail with an exception Spark threw, such as a failed
spill of another
+/// consumer or the interrupt of a killed task. That comes back as another
kind of error and has
+/// to fail the task. Spilling would hide it, and the task would carry on as
if Spark had refused.
+fn is_refusal(e: &DataFusionError) -> bool {
Review Comment:
Done: one `pub fn is_refusal` in `datafusion-comet-common`, used by both.
##########
native/core/src/execution/memory_pools/mod.rs:
##########
@@ -180,6 +181,52 @@ mod tests {
}
}
+ /// Spark refusing memory is a refusal, which makes an operator spill.
Spark failing a `grow`
+ /// is not: every later `try_grow` of the task fails with that error,
through the wrappers of
+ /// each pool type, also where the pool's own limits or the spill replay
would have refused it
+ /// or recorded it past a refusal.
+ #[test]
+ fn a_failed_grow_fails_every_later_try_grow_of_the_task() {
Review Comment:
Done: removed the older replay test and the repeated overcommit and release
checks.
--
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]
