hubgeter commented on code in PR #67087:
URL: https://github.com/apache/doris/pull/67087#discussion_r3849780430
##########
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:
Fixed in 95c86ded656. The spill sink now finalizes VIcebergTableWriter from
sink() on EOS, so the final sort, spill, and merge run before PipelineTask
releases the sink reservation. Successful inner close defers file cleanup until
local-state close, preserving cancellation and error cleanup. Covered by
non-empty and empty EOS tests; the focused ASAN BE UT passed 7/7.
##########
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);
+ if (final_status.ok() && !writer_status.ok()) {
+ final_status = writer_status;
+ }
+ _writer.reset();
Review Comment:
Fixed in 95c86ded656. Writer pointer snapshot and reset are serialized by
_writer_mutex. Memory accounting and revoke copy the child shared_ptr while
holding that lock, then do sorter work after releasing it; final writer close
and cleanup also stay outside the outer lock, and reset uses the same lock.
This prevents the null-check/reset UAF without holding a lifecycle lock across
sort or merge I/O.
##########
be/src/exec/operator/iceberg_table_sink_operator.cpp:
##########
@@ -25,9 +25,46 @@ 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);
+ 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::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:
Fixed in 95c86ded656. Both normal and spill close paths replace an incoming
OK status with state->cancel_reason() before closing. Spill EOS also observes
cancellation, and a successful early EOS close retains deferred file cleanup
until outer close so a later cancellation deletes the files. Tests cover normal
cancellation and spill cancellation after EOS.
--
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]
