github-actions[bot] commented on code in PR #67087:
URL: https://github.com/apache/doris/pull/67087#discussion_r3852313611
##########
be/src/exec/sink/writer/iceberg/viceberg_sort_writer.cpp:
##########
@@ -82,7 +83,21 @@ size_t VIcebergSortWriter::data_size() const {
size_t VIcebergSortWriter::get_reserve_mem_size(RuntimeState* state, bool eos)
const {
std::lock_guard<std::mutex> lock(_sorter_mutex);
- return _sorter == nullptr ? 0 : _sorter->get_reserve_mem_size(state, eos);
+ size_t reserve_size = _sorter == nullptr ? 0 :
_sorter->get_reserve_mem_size(state, eos);
+ if (eos && !_sorted_spill_files.empty()) {
+ const size_t buffer_size =
static_cast<size_t>(state->spill_buffer_size_bytes());
+ const size_t merge_limit =
static_cast<size_t>(state->spill_sort_merge_mem_limit_bytes());
+ const size_t max_fan_in = std::max<size_t>(2, merge_limit /
buffer_size);
+ // Reservation is computed before sink(), so a non-empty EOS can add
one final spill run.
+ const size_t selected_streams = std::min(_sorted_spill_files.size(),
max_fan_in - 1) + 1;
+ // Every selected cursor eagerly owns one input block. A multiway
merge also builds a
+ // separate output block while those inputs remain live.
+ const size_t merge_buffer_count = selected_streams + (selected_streams
> 1 ? 1 : 0);
Review Comment:
**[P1] Include merger workspace in the EOS reservation**
With the default 8 MiB spill buffer and one-byte rows, the intermediate
merger receives a batch of 8,388,608 rows. `VSortedRunMerger::get_next()`
reserves both its index and block-address vectors to that count, so those two
vectors alone allocate about 128 MiB while this formula requests only 72 MiB
for the default eight-way merge. Every eagerly prepared cursor also retains its
reader's serialized buffer and a deserialized input block. Under memory
pressure the synchronous EOS close can therefore exceed a reservation that was
successfully granted. Please bound/count the row-selection and retained
reader/input/output workspace, and cover a real intermediate merge rather than
only the arithmetic helper.
##########
be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp:
##########
@@ -221,6 +219,34 @@ Status VIcebergTableWriter::write_prepared_block(Block&
block) {
return _write_prepared_block(block);
}
+size_t VIcebergTableWriter::get_reserve_mem_size(RuntimeState* state, bool
eos) const {
+ size_t reserve_size = state->minimum_operator_memory_required_bytes();
+ if (!eos) {
+ auto current_writer = _current_writer.load();
+ if (!current_writer) {
+ return reserve_size;
+ }
+ auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(current_writer.get());
+ DORIS_CHECK(sort_writer != nullptr);
+ return std::max(reserve_size, sort_writer->get_reserve_mem_size(state,
false));
+ }
+
+ auto partition_writer_snapshot =
std::make_shared<PartitionWriterSnapshot>();
+ partition_writer_snapshot->reserve(_partitions_to_writers.size());
+ // close() finalizes partition writers sequentially, so reserve the
largest close peak rather
+ // than the sum. Publish the same owning set for workload
accounting/revoke if reservation
+ // fails. The admission floor covers a first non-empty EOS before its
first writer exists.
Review Comment:
**[P1] Reserve for partitions created by the pending block**
This estimate only visits writers that already exist, but it runs before
`sink()` consumes the pending block. On the first non-empty EOS, or when the
block introduces a new partition, the reservation is therefore only the fixed
admission floor; that same call then materializes/filters the block, creates a
sorter, appends the rows, and immediately finalizes it. The production 32,000
KiB floor is not tied to a legal adaptive block (which can target up to 512
MiB) or its sort workspace. Please account for the pending block and possible
new destinations, or split ingestion from finalization and recompute after
those writers exist.
##########
be/src/exec/sink/writer/iceberg/viceberg_sort_writer.cpp:
##########
@@ -58,9 +58,10 @@ Status VIcebergSortWriter::open(RuntimeState* state,
RuntimeProfile* profile,
Status VIcebergSortWriter::write(Block& block) {
std::lock_guard<std::mutex> lock(_sorter_mutex);
+ // FullSorter consumes the input block, so derive the spill batch size
before append_block().
+ _update_spill_block_batch_row_count(block);
Review Comment:
**[P1] Keep spill batches byte-bounded as row widths change**
This one-shot sample permanently fixes the spill row count from the first
non-empty block. If later variable-length rows are wider, `_do_spill()` still
reads that many rows and has no byte stop, so a narrow first block can turn the
configured 8 MiB spill buffer into a block hundreds of MiB large during memory
revocation. The final-merger clamp does not affect this spill or
intermediate-merge path, and the uniform-width helper test cannot expose the
skew. Please make these reads byte-bounded (or continually lower the row limit
from observed/max widths) and test a narrow-first, wide-later actual spill.
--
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]
