mbutrovich commented on code in PR #6205:
URL: https://github.com/apache/datafusion-comet/pull/6205#discussion_r4097161359
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -154,16 +155,26 @@ impl MemoryPool for CometFairMemoryPool {
.pool_size
.checked_div(num)
.expect("overflow in checked_div");
- // We use state.used instead of reservation.size() because
DataFusion 53+
- // calls pool.try_grow() before incrementing the reservation's
atomic size,
- // so reservation.size() would not include prior grows.
- let used = state.used;
- if limit < used + additional {
+ // DataFusion calls pool.try_grow() before adding to the
reservation's size, so this
+ // is what the reservation held before the request. shrink() is
the other way round.
+ let size = reservation.size();
+ if limit < size.saturating_add(additional) {
return resources_err!(
- "Failed to acquire {additional} bytes where {used} bytes
already reserved ({} bytes overcommitted) and the fair limit is {limit} bytes,
{num} registered",
+ "Failed to acquire {additional} bytes where this
reservation already holds {size} bytes and the fair limit is {limit} bytes,
{num} registered ({} bytes overcommitted)",
self.spark.overcommit()
);
}
Review Comment:
What happens to other consumers when one operator grows through several
sibling reservations? Each sibling is checked against the full share on its
own, so the share check becomes a per-allocation cap for those operators, and
only the pool-total check bounds them. The streaming merge in sort creates a
fresh `new_empty()` reservation for every cursor batch ([`stream.rs`
L198-L199](https://github.com/apache/datafusion/blob/7d3835c71f30cbd3c3ae4041732267f1f453097a/datafusion/physical-plan/src/sorts/stream.rs#L198-L199)),
and the hash aggregate's merge path does the same ([`hash_stream.rs`
L402](https://github.com/apache/datafusion/blob/7d3835c71f30cbd3c3ae4041732267f1f453097a/datafusion/physical-plan/src/aggregates/hash_stream.rs#L402)).
I checked this with a scratch test at the head commit: a 100-byte pool
wrapped in `TrackConsumersPool`, consumer `a` with a sibling from
`new_empty()`, and consumer `b`. `a` and its sibling each reserve 50, then
`b.try_grow(1)` fails with "Failed to acquire 1 bytes where 100 bytes already
reserved ... and the pool limit is 100 bytes". On main, `a`'s second 50 would
have been refused, so `b` could not be starved this way. The PR description
says this never refuses a request that main accepts. That holds for a given
pool state, but this PR makes pool states reachable that main never reached,
and in them `b` is refused where main would have accepted.
Tracking usage per consumer instead of per reservation would close this and
would also stop the check from depending on when DataFusion updates
`reservation.size()`, which is the ordering that already changed once in
DataFusion 53. `reservation.consumer().id()` is available in `try_grow`, `grow`
and `shrink`, and #5847 has an implementation of that. Could this PR take the
per-consumer accounting from there, or, if you want to keep it per reservation
for the backport, add a test that pins down the sibling behavior and correct
the tuning guide sentence ("prevents operators from using more than an even
fraction") that it contradicts?
--
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]
