comphead commented on code in PR #7607:
URL: https://github.com/apache/datafusion-comet/pull/7607#discussion_r4554846151
##########
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:
This test overlaps with
`a_failed_spark_call_during_the_replay_is_not_recorded` in `spill_replay.rs`.
Both run the two pool types, fail `merge.grow`, check that the replay's
`try_grow` fails without `ResourcesExhausted`, and check that `reserved()` does
not change. This one adds the fair pool's limits and the repeated calls. Could
the older test be removed, or extended with the new assertions, so the case is
tested in one place? In the same way,
`a_failed_acquire_fails_every_later_try_acquire` in `spark_memory.rs` repeats
the overcommit and release assertions of `failed_acquire_is_all_overcommit`.
##########
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:
The original exception goes only to the first `try_grow` after a failed
`grow`. If that caller is `ExternalSorter` with batches in memory, it takes the
error for a refusal, spills and asks again
(`reserve_memory_for_batch_and_maybe_spill` in DataFusion 55.2). The task then
fails with the `DataFusionError::Execution` message as a generic
`CometNativeException`, where a JVM consumer would get Spark's
`SparkOutOfMemoryError` from `TaskMemoryManager.spillConsumer`. A failed
`try_acquire` never stores the throwable, so its retries end the same way.
Could the throwable be kept and rethrown on every later failure? `Global` is
not `Clone` in jni 0.22, so that needs an env, and I have not tried it. If that
is too much for this PR, could the description's "the task fails with the
exception Spark threw" be limited to operators that pass the first error on?
##########
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:
This has the same body as `is_refusal` in
`native/core/src/execution/operators/iceberg_write.rs`. `datafusion-comet` and
`datafusion-comet-shuffle` both depend on `datafusion-comet-common`. One `pub
fn is_refusal` there would give both callers the same definition of a refusal,
and its doc comment could explain the Spark exception case once.
##########
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:
The PR description leaves one case open: a failed `grow` with no later
`try_grow` in the task is still only logged, which is the second half of the
expected behavior in #7244. Could this paragraph say so, or could the gap go
under **Open problems**? As written it reads as if every exception now reaches
the task. Also, `Closes #7244` will close the issue on merge. If the
end-of-stream check goes to a follow-up, `Part of #7244` would keep it open.
--
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]
