andygrove commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338746703
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -851,8 +851,7 @@ impl ExecutionPlan for IcebergWriteExec {
context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
let metrics = WriteMetrics::new(&self.metrics, partition);
- // One consumer for the whole task, however many files it opens: a
consumer per file
- // would shrink every other consumer's share of the fair pool as a
fanout write widened.
+ // One consumer for the whole task, however many files it opens.
Review Comment:
Agreed, and thanks for the catch. The fanout writer now registers with
`with_can_spill(true)` in 6a1c4dffca, through a small `write_consumer` helper,
so it is held to a share and closes files rather than growing until Spark
refuses. The unpartitioned and clustered writers stay unspillable. A new test,
`only_a_fanout_write_is_held_to_a_share_of_the_pool`, checks all three modes
against a sort in the same task, and the guide now describes the two cases
separately, which removes the contradiction.
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -153,22 +209,32 @@ impl MemoryPool for CometFairMemoryPool {
) -> Result<(), DataFusionError> {
if additional > 0 {
let mut state = self.state.lock();
- let num = state.consumers.len();
- let limit = self
- .pool_size
- .checked_div(num)
- .expect("overflow in checked_div");
- let consumer_used = *state.consumer_used(reservation);
- if limit < consumer_used.saturating_add(additional) {
- return resources_err!(
- "Failed to acquire {additional} bytes where this consumer
already holds {consumer_used} bytes and the fair limit is {limit} bytes, {num}
registered ({} bytes overcommitted)",
- self.spark.overcommit()
- );
- }
- // The shares alone do not bound the pool's total, because a
consumer keeps what it
- // reserved before another consumer registered.
+ let can_spill = reservation.consumer().can_spill();
let used = state.used;
- if self.pool_size < used.saturating_add(additional) {
+ if can_spill {
+ let (task_share, tasks) = self.task_share();
+ // `num_spill` counts this consumer.
+ let limit = task_share.saturating_sub(state.unspillable) /
state.num_spill.max(1);
Review Comment:
This is a trade-off I accepted rather than weighed carefully enough, so
thank you for spelling out the paths. I kept the policy as benchmarked and
documented it instead: the 1.2.0 migration note, the tuning guide and the
contributor guide now say a sort can fail with "Not enough memory to continue
external sort" when one batch is larger than its share, and point to more
off-heap memory, a smaller `spark.comet.batchSize` or `greedy_unified`. I did
not add an exemption for a consumer that holds nothing, since that changes
behavior the benchmarks did not measure. Would you want it in this PR, or as a
follow-up once there is a repro?
--
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]
