github-actions[bot] commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3751668430
##########
be/src/exec/sort/sorter.cpp:
##########
@@ -202,35 +217,61 @@ bool FullSorter::has_enough_capacity(Block* input_block,
Block* unsorted_block)
}
size_t FullSorter::get_reserve_mem_size(RuntimeState* state, bool eos) const {
- size_t size_to_reserve = 0;
+ return get_reserve_mem_size_components(state, eos).total();
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state,
+ bool eos)
const {
+ const auto rows = _state->unsorted_block()->rows();
+ const auto bytes = _state->unsorted_block()->bytes();
+ const auto bytes_per_row = rows == 0 ? 0 : bytes / rows;
+ return get_reserve_mem_size_components(
+ state, eos, state->batch_size(),
+ saturating_multiply_size(bytes_per_row, state->batch_size()));
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state, bool eos,
+ size_t
incoming_rows,
+ size_t
incoming_bytes) const {
+ SorterReserveMemory reserve;
const auto rows = _state->unsorted_block()->rows();
if (rows != 0) {
const auto bytes = _state->unsorted_block()->bytes();
const auto allocated_bytes =
_state->unsorted_block()->allocated_bytes();
- const auto bytes_per_row = bytes / rows;
- const auto estimated_size_of_next_block = bytes_per_row *
state->batch_size();
- auto new_block_bytes = estimated_size_of_next_block + bytes;
- auto new_rows = rows + state->batch_size();
+ auto new_block_bytes = saturating_add_size(bytes, incoming_bytes);
+ auto new_rows = saturating_add_size(rows, incoming_rows);
// If the new size is greater than 85% of allocalted bytes, it maybe
need to realloc.
- if ((new_block_bytes * 100 / allocated_bytes) >= 85) {
- size_to_reserve += (size_t)(allocated_bytes * 1.15);
+ const auto growth_threshold = static_cast<size_t>(
+ (static_cast<unsigned __int128>(allocated_bytes) * 85 + 99) /
100);
+ const size_t growth_trigger_bytes = growth_threshold > bytes ?
growth_threshold - bytes : 0;
+ if (incoming_rows > 0 && growth_trigger_bytes <= incoming_bytes) {
+ reserve.retained_growth = static_cast<size_t>(std::min<unsigned
__int128>(
+ (static_cast<unsigned __int128>(allocated_bytes) * 115 +
99) / 100,
+ std::numeric_limits<size_t>::max()));
+ reserve.retained_growth_trigger_bytes = growth_trigger_bytes;
}
auto sort = new_rows > _buffered_block_size || new_block_bytes >
_buffered_block_bytes;
Review Comment:
[P1] Include the forced EOS sort in this reservation. This method never uses
`eos` and adds sort workspace only after the generic 4M-row/256 MiB append
thresholds, but Iceberg close calls `do_sort()` for every nonempty sorter (or
`_do_spill()`, which first sorts a pending run). A buffered sorter of tens of
MiB below those thresholds can therefore reserve only the operator floor while
`sort_block` keeps the source live and allocates the permutation plus a full
permuted destination. The new fan-in test uses an empty sorter, so it misses
both the no-spill and pending-run cases. Please size the actual EOS
sort/serialization workspace and cover a buffered sorter below the normal
thresholds.
##########
be/src/exec/sink/writer/async_result_writer.cpp:
##########
@@ -199,6 +243,9 @@ void AsyncResultWriter::process_block(RuntimeState* state,
RuntimeProfile* opera
Status st = finish(state);
_writer_status.update(st);
}
+ if (reservation_held_for_finish) {
+ thread_context()->thread_mem_tracker_mgr->shrink_reserved();
Review Comment:
[P1] Keep the EOS reservation through the real Iceberg finalization.
`VIcebergTableWriter` does not override `finish()` (the base implementation is
a no-op); its partition writers run `do_sort()`, `_do_spill()`, and
`_combine_files_output()` from `close()`, which is called only after this
`shrink_reserved()`. Consequently the newly sized fan-in reservation is already
gone when the merger allocates its cursors and blocks. Please release the token
after `close(st)` on every path (ideally with an RAII guard), and make the
async test observe the reservation inside `close`, not only an overridden
`finish`.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorTransaction.java:
##########
@@ -288,13 +288,23 @@ private ScheduledFuture<?> startCommitLockHeartbeat(long
lockId) {
}
private void commitWhileTableLocked() {
- validateWriteMetadataBeforePublication();
- // The classification (finishInsertTable) ran from the executor in the
legacy class; the unified SPI
- // exposes only commit(), so it runs here (before the committer) to
populate the action maps. If it
- // throws, the committer was never created and the engine's subsequent
rollback() cleans up.
- finishInsertTable(nameMapping);
- // Classification can perform metastore reads, so close that interval
before any file or HMS mutation.
- validateWriteMetadataBeforePublication();
+ try {
Review Comment:
[P1] Extend self-cleanup to lock acquisition failures. The rollback boundary
still starts too late: `PluginDrivenTransactionManager.commit()` has already
removed the transaction, so if `acquireExclusiveTableLock` times out or its RPC
fails, this catch is never entered, the wrapper closes the connector, and the
executor's later manager rollback finds no entry. Any valid S3 MPUs already fed
into this transaction are then left open even though no Hive metadata was
published. Please include lock acquisition in the connector's self-cleanup
failure boundary (while releasing a lock only if it was acquired), and add a
failure-injection test that verifies reported uploads are aborted on an HMS
lock timeout/RPC failure.
--
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]
