github-actions[bot] commented on code in PR #69355:
URL: https://github.com/apache/doris/pull/69355#discussion_r4331442885
##########
be/src/exec/operator/iceberg_table_sink_operator.cpp:
##########
@@ -24,9 +24,64 @@ Status IcebergTableSinkLocalState::init(RuntimeState* state,
LocalSinkStateInfo&
RETURN_IF_ERROR(Base::init(state, info));
SCOPED_TIMER(exec_time_counter());
SCOPED_TIMER(_init_timer);
- auto& p = _parent->cast<Parent>();
- RETURN_IF_ERROR(_writer->init_properties(p._pool, p._row_desc));
+ _writer = std::make_unique<VIcebergTableWriter>(info.tsink,
_output_vexpr_ctxs);
+ _writer->defer_file_cleanup_until_outer_close();
+ auto& parent = _parent->cast<Parent>();
+ RETURN_IF_ERROR(_writer->init_properties(parent._pool, parent._row_desc));
return Status::OK();
}
+Status IcebergTableSinkLocalState::open(RuntimeState* state) {
+ SCOPED_TIMER(exec_time_counter());
+ SCOPED_TIMER(_open_timer);
+ RETURN_IF_ERROR(Base::open(state));
+
+ 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 IcebergTableSinkLocalState::sink(RuntimeState* state, Block* block,
bool /*eos*/) {
+ if (block->rows() == 0) {
+ return Status::OK();
+ }
+ DCHECK(_writer);
+ return _writer->write(state, *block);
+}
+
+Status IcebergTableSinkLocalState::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;
+ // Observe cancellation before close so a known-cancelled write is not
finalized as successful.
+ if (final_status.ok() && state->is_cancelled()) {
+ final_status = state->cancel_reason();
+ }
+ Status writer_status = _writer->close(final_status);
Review Comment:
[P2] Keep unsorted file finalization under the EOS reservation
With reserve-memory admission enabled, the normal sink ignores EOS and
reaches this close only from `TaskScheduler::close_task`, after `PipelineTask`
has released the sink call's reservation. The previous async writer retained
its EOS token through `close()`. This also matters without a sorter:
`VParquetWriter` buffers a row group, and [Arrow 24's final-page
flush](https://github.com/apache/arrow/blob/apache-arrow-24.0.0/cpp/src/parquet/column_writer.cc#L1024-L1062)
grows the uncompressed/compression buffers during close. For an INSERT ending
with an empty EOS and a partial Parquet page, concurrent queries can claim that
released headroom before finalization, so the flush bypasses admission and can
wait for allocator GC or allocate beyond the process limit. Please finalize the
normal writer inside `sink(EOS)` while retaining cancellation-aware cleanup for
outer close, as the spill sink does, or retain a reservation through this 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]
