Gabriel39 commented on code in PR #67087:
URL: https://github.com/apache/doris/pull/67087#discussion_r3849261551
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -36,17 +36,47 @@ Status SpillIcebergTableSinkLocalState::init(RuntimeState*
state, LocalSinkState
SCOPED_TIMER(_init_timer);
_init_spill_counters();
+ _writer = std::make_unique<VIcebergTableWriter>(info.tsink,
_output_vexpr_ctxs);
- auto& p = _parent->cast<Parent>();
- RETURN_IF_ERROR(_writer->init_properties(p._pool, p._row_desc));
+ auto& parent = _parent->cast<Parent>();
+ RETURN_IF_ERROR(_writer->init_properties(parent._pool, parent._row_desc));
return Status::OK();
}
Status SpillIcebergTableSinkLocalState::open(RuntimeState* state) {
SCOPED_TIMER(Base::exec_time_counter());
SCOPED_TIMER(Base::_open_timer);
RETURN_IF_ERROR(Base::open(state));
- return Status::OK();
+
+ auto& parent = _parent->cast<Parent>();
+ _output_vexpr_ctxs.resize(parent._output_vexpr_ctxs.size());
+ for (size_t i = 0; i < _output_vexpr_ctxs.size(); ++i) {
+ RETURN_IF_ERROR(parent._output_vexpr_ctxs[i]->clone(state,
_output_vexpr_ctxs[i]));
+ }
+ return _writer->open(state, operator_profile());
+}
+
+Status SpillIcebergTableSinkLocalState::close(RuntimeState* state, Status
exec_status) {
+ if (_closed) {
+ return Status::OK();
+ }
+
+ SCOPED_TIMER(exec_time_counter());
+ SCOPED_TIMER(_close_timer);
+
+ DCHECK(_writer);
+ Status final_status = exec_status;
+ Status writer_status = _writer->close(exec_status);
Review Comment:
[P1] Please keep the EOS reservation alive while finalizing the spill
writer. `PipelineTask::execute()` reserves `get_reserve_mem_size(state, eos)`
only around the `sink()` call, but this PR makes `sink_impl()` ignore `eos` and
return immediately for the empty EOS block. That reservation is therefore
released before this `close()` runs. For a sink with `sort_info`,
`VIcebergSortWriter::close()` can still run `do_sort()`, spill the remaining
block, and perform the final multi-way merge. Under memory pressure those
allocations can hit `QUERY_MEMORY_EXCEEDED` instead of letting the task
pause/revoke memory, so a large sorted Iceberg insert may fail during
finalization. Please finalize the writer from the EOS sink path while its
reservation is held, or add equivalent reserve/retry handling around close.
--
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]
