github-actions[bot] commented on code in PR #67087:
URL: https://github.com/apache/doris/pull/67087#discussion_r3849310092
##########
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:
**[P1] Keep the writer alive while revocable accounting can snapshot it**
Workload-group accounting calls the sink from outside the pipeline scheduler
after only sampling `PipelineTask::_running`. A scheduler thread can
subsequently set the task running and reach terminal close, so this reset can
occur between `get_revocable_mem_size()`'s `_writer` null check and its
`_writer->current_writer()` call (for example when cancellation races
paused-query accounting). That is a data race/use-after-free on the owning
`unique_ptr`; the atomic child snapshot only helps after the parent dereference
succeeds. Please keep the writer alive through local-state teardown, or fence
accounting and close with a shared owning snapshot/lock.
##########
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:
**[P1] Preserve cancellation status before finalizing files**
Cancellation can arrive during the final `PipelineTask::execute()` after the
scheduler sampled the fragment as live. Its exit defer then marks the task done
because the fragment is cancelled, but the scheduler can still call this close
with an OK execution status. Forwarding that OK value makes partition close add
commit data and lets the table writer clear its cleanup list instead of
deleting the cancelled query's files. The removed async close normalized this
exact case through `state->cancel_reason()`. Please convert an OK `exec_status`
to the runtime cancellation reason before closing the writer here and in the
spill sink, while preserving any existing non-OK status.
##########
be/test/exec/operator/iceberg_table_sink_operator_test.cpp:
##########
@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "exec/operator/iceberg_table_sink_operator.h"
+
+#include <gtest/gtest.h>
+
+#include <type_traits>
+
+#include "exec/operator/spill_iceberg_table_sink_operator.h"
+#include "exec/sink/writer/async_result_writer.h"
+
+namespace doris {
+
+static_assert(!std::is_base_of_v<AsyncResultWriter, VIcebergTableWriter>);
+
+TEST(IcebergTableSinkOperatorTest,
SyncWritersUseBlockingSchedulerWithoutDependencies) {
Review Comment:
**[P2] Exercise the synchronous lifecycle in this test**
This test only constructs uninitialized local states and checks their
inheritance/scheduler-dependency shape, so it still passes if the new
`init`/`open`/inline `write`/EOS `close` sequence loses buffered rows or
mishandles write/close failures and cancellation cleanup. Those are the
behavioral changes in this PR, including the spill writer's final sort/merge.
Please add coverage that drives at least a successful non-empty/empty EOS and
the error/cleanup paths through the converted operator lifecycle.
--
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]
