morningman commented on code in PR #69355:
URL: https://github.com/apache/doris/pull/69355#discussion_r4333438097
##########
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:
Fixed in fc54f991f97. The finding is right. `PipelineTask::execute` releases
the sink reservation when the scope around `_sink->sink()` ends
(`DEFER_RELEASE_RESERVED`), and `close_task` closes the sink only after that.
So with the first commit, the unsorted writer's close, which flushes what every
partition file still buffers (such as a half-filled Parquet row group), ran
with no reservation, while master's async writer adopts the EOS token and holds
it through `finish()` and `close()` (#66348).
The normal sink now closes its writer in the EOS `sink()` call, the same way
the spill sink does:
- `_close_writer()` closes the writer once and keeps its status. `close()`
calls it as well, so it closes the writer itself only when the task ends before
EOS (an error or a cancellation).
- `close()` keeps both cancellation checks and passes the final status to
`finish_deferred_file_cleanup()`, so a cancellation that arrives after EOS
still deletes the closed files instead of reporting them.
- The EOS call is covered by the sink's default reservation,
`minimum_operator_memory_required_bytes()`, which is the amount master's async
path carried to `close()`.
Tests in `IcebergTableSinkOperatorTest`: the case ported from branch-4.1
that expected the writer to stay open after an empty EOS now expects the EOS
call to close it, and five new cases cover a non-empty EOS, a cancellation seen
at EOS, a write error, a close error, and a cancellation after EOS. Removing
the close from the EOS `sink()` or closing the writer a second time in
`close()` makes 5 of these cases fail; dropping the cancellation check at EOS
makes 1 fail.
For reference, branch-4.1 and branch-4.2 also close this writer in
`close()`, but their async queue never carried reservations, so this was a
regression only against master's model.
--
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]
