Gabriel39 commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3698114241
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -55,47 +64,57 @@ size_t
SpillIcebergTableSinkLocalState::get_reserve_mem_size(RuntimeState* state
if (!_writer) {
return 0;
}
- auto current_writer = _writer->current_writer();
- auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(current_writer.get());
- if (!sort_writer) {
- return 0;
+ std::vector<size_t> per_partition_reservations;
+ auto active_writers = _writer->active_writers();
+ per_partition_reservations.reserve(active_writers->size());
+ for (const auto& writer : *active_writers) {
+ if (auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(writer.get())) {
+
per_partition_reservations.push_back(sort_writer->get_reserve_mem_size(state,
eos));
+ }
}
-
- return sort_writer->get_reserve_mem_size(state, eos);
+ // One input block is partitioned among writers and consumed serially, so
their full-batch estimates overlap.
+ return bounded_iceberg_reserve_size(per_partition_reservations);
Review Comment:
Fixed in 481123549a and reinforced by 399b6fe356. The reservation now sums
retained growth across all touched sorters, uses only the maximum transient
sorting workspace, and reserves the incoming block before cold writers are
published. Regression coverage includes both accumulated growth and the
no-published-writer case.
--
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]
