[feature](paimon) implemet INSERT INTO for Paimon external tables

be/src/pipeline/exec/operator.cpp

@@ -41,6 +41,7 @@
 #include "pipeline/exec/hashjoin_probe_operator.h"
 #include "pipeline/exec/hive_table_sink_operator.h"
 #include "pipeline/exec/iceberg_table_sink_operator.h"
+#include "pipeline/exec/paimon_table_sink_operator.h"
 #include "pipeline/exec/jdbc_scan_operator.h"
 #include "pipeline/exec/jdbc_table_sink_operator.h"
 #include "pipeline/exec/local_merge_sort_source_operator.h"
@@ -820,6 +821,7 @@ DECLARE_OPERATOR(ResultFileSinkLocalState)
 DECLARE_OPERATOR(OlapTableSinkLocalState)
 DECLARE_OPERATOR(OlapTableSinkV2LocalState)
 DECLARE_OPERATOR(HiveTableSinkLocalState)
+DECLARE_OPERATOR(PaimonTableSinkLocalState)
 DECLARE_OPERATOR(TVFTableSinkLocalState)
 DECLARE_OPERATOR(IcebergTableSinkLocalState)
 DECLARE_OPERATOR(SpillIcebergTableSinkLocalState)
@@ -941,6 +943,7 @@ template class AsyncWriterSink<doris::vectorized::VJdbcTableWriter, JdbcTableSin
 template class AsyncWriterSink<doris::vectorized::VTabletWriter, OlapTableSinkOperatorX>;
 template class AsyncWriterSink<doris::vectorized::VTabletWriterV2, OlapTableSinkV2OperatorX>;
 template class AsyncWriterSink<doris::vectorized::VHiveTableWriter, HiveTableSinkOperatorX>;
+template class AsyncWriterSink<doris::vectorized::VPaimonTableWriter, PaimonTableSinkOperatorX>;
 template class AsyncWriterSink<doris::vectorized::VIcebergTableWriter, IcebergTableSinkOperatorX>;
 template class AsyncWriterSink<doris::vectorized::VIcebergTableWriter,
                                SpillIcebergTableSinkOperatorX>;

be/src/pipeline/exec/paimon_table_sink_operator.cpp

@@ -0,0 +1,34 @@
+// 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 "paimon_table_sink_operator.h"
+
+#include "common/status.h"
+
+namespace doris::pipeline {
+#include "common/compile_check_begin.h"
+
+Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) {
+    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));
+    return Status::OK();
+}
+
+} // namespace doris::pipeline

be/src/pipeline/exec/paimon_table_sink_operator.h

@@ -0,0 +1,86 @@
+// 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.
+
+#pragma once
+
+#include "operator.h"
+#include "vec/sink/writer/paimon/vpaimon_table_writer.h"
+
+namespace doris::pipeline {
+#include "common/compile_check_begin.h"
+
+class PaimonTableSinkOperatorX;
+
+class PaimonTableSinkLocalState final
+        : public AsyncWriterSink<vectorized::VPaimonTableWriter, PaimonTableSinkOperatorX> {
+public:
+    using Base = AsyncWriterSink<vectorized::VPaimonTableWriter, PaimonTableSinkOperatorX>;
+    using Parent = PaimonTableSinkOperatorX;
+    ENABLE_FACTORY_CREATOR(PaimonTableSinkLocalState);
+    PaimonTableSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
+            : Base(parent, state) {}
+    Status init(RuntimeState* state, LocalSinkStateInfo& info) override;
+    Status open(RuntimeState* state) override {
+        SCOPED_TIMER(exec_time_counter());
+        SCOPED_TIMER(_open_timer);
+        return Base::open(state);
+    }
+    friend class PaimonTableSinkOperatorX;
+};
+
+class PaimonTableSinkOperatorX final : public DataSinkOperatorX<PaimonTableSinkLocalState> {
+public:
+    using Base = DataSinkOperatorX<PaimonTableSinkLocalState>;
+    PaimonTableSinkOperatorX(ObjectPool* pool, int operator_id, const RowDescriptor& row_desc,
+                             const std::vector<TExpr>& t_output_expr)
+            : Base(operator_id, 0, 0),
+              _row_desc(row_desc),
+              _t_output_expr(t_output_expr),
+              _pool(pool) {}
+
+    Status init(const TDataSink& thrift_sink) override {
+        RETURN_IF_ERROR(Base::init(thrift_sink));
+        RETURN_IF_ERROR(vectorized::VExpr::create_expr_trees(_t_output_expr, _output_vexpr_ctxs));
+        return Status::OK();
+    }
+
+    Status prepare(RuntimeState* state) override {
+        RETURN_IF_ERROR(Base::prepare(state));
+        RETURN_IF_ERROR(vectorized::VExpr::prepare(_output_vexpr_ctxs, state, _row_desc));
+        return vectorized::VExpr::open(_output_vexpr_ctxs, state);
+    }
+
+    Status sink(RuntimeState* state, vectorized::Block* in_block, bool eos) override {
+        auto& local_state = get_local_state(state);
+        SCOPED_TIMER(local_state.exec_time_counter());
+        COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());
+        return local_state.sink(state, in_block, eos);
+    }
+
+private:
+    friend class PaimonTableSinkLocalState;
+    template <typename Writer, typename Parent>
+        requires(std::is_base_of_v<vectorized::AsyncResultWriter, Writer>)
+    friend class AsyncWriterSink;
+    const RowDescriptor& _row_desc;
+    vectorized::VExprContextSPtrs _output_vexpr_ctxs;
+    const std::vector<TExpr>& _t_output_expr;
+    ObjectPool* _pool = nullptr;
+};
+
+#include "common/compile_check_end.h"
+} // namespace doris::pipeline

be/src/pipeline/pipeline_fragment_context.cpp

@@ -63,6 +63,7 @@
 #include "pipeline/exec/hashjoin_probe_operator.h"
 #include "pipeline/exec/hive_table_sink_operator.h"
 #include "pipeline/exec/iceberg_table_sink_operator.h"
+#include "pipeline/exec/paimon_table_sink_operator.h"
 #include "pipeline/exec/jdbc_scan_operator.h"
 #include "pipeline/exec/jdbc_table_sink_operator.h"
 #include "pipeline/exec/local_merge_sort_source_operator.h"
@@ -1094,6 +1095,14 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS
         }
         break;
     }
+    case TDataSinkType::PAIMON_TABLE_SINK: {
+        if (!thrift_sink.__isset.paimon_table_sink) {
+            return Status::InternalError("Missing paimon table sink.");
+        }
+        _sink = std::make_shared<PaimonTableSinkOperatorX>(pool, next_sink_operator_id(), row_desc,
+                                                           output_exprs);
+        break;
+    }
     case TDataSinkType::MAXCOMPUTE_TABLE_SINK: {
         if (!thrift_sink.__isset.max_compute_table_sink) {
             return Status::InternalError("Missing max compute table sink.");

be/src/runtime/fragment_mgr.cpp

@@ -566,6 +566,27 @@ void FragmentMgr::coordinator_callback(const ReportStatusRequest& req) {
         }
     }
 
+    LOG(INFO) << "coordinator_callback paimon: done=" << req.done
+              << ", fragment_state_pcd_size=" << req.runtime_state->paimon_commit_datas().size()
+              << ", runtime_states_count=" << req.runtime_states.size();
+    if (auto pcd = req.runtime_state->paimon_commit_datas(); !pcd.empty()) {
+        params.__isset.paimon_commit_datas = true;
+        params.paimon_commit_datas.insert(params.paimon_commit_datas.end(), pcd.begin(),
+                                          pcd.end());
+    } else if (!req.runtime_states.empty()) {
+        for (auto* rs : req.runtime_states) {
+            auto rs_pcd = rs->paimon_commit_datas();
+            LOG(INFO) << "coordinator_callback paimon task_state: pcd_size=" << rs_pcd.size();
+            if (!rs_pcd.empty()) {
+                params.__isset.paimon_commit_datas = true;
+                params.paimon_commit_datas.insert(params.paimon_commit_datas.end(),
+                                                  rs_pcd.begin(), rs_pcd.end());
+            }
+        }
+    }
+    LOG(INFO) << "coordinator_callback paimon result: isset=" << params.__isset.paimon_commit_datas
+              << ", count=" << params.paimon_commit_datas.size();
+
     // Send new errors to coordinator
     req.runtime_state->get_unreported_errors(&(params.error_log));
     params.__isset.error_log = (!params.error_log.empty());

be/src/runtime/runtime_state.h

@@ -521,6 +521,16 @@ class RuntimeState {
         _mc_commit_datas.emplace_back(mc_commit_data);
     }
 
+    std::vector<TPaimonCommitData> paimon_commit_datas() const {
+        std::lock_guard<std::mutex> lock(_paimon_commit_datas_mutex);
+        return _paimon_commit_datas;
+    }
+
+    void add_paimon_commit_datas(const TPaimonCommitData& paimon_commit_data) {
+        std::lock_guard<std::mutex> lock(_paimon_commit_datas_mutex);
+        _paimon_commit_datas.emplace_back(paimon_commit_data);
+    }
+
     // local runtime filter mgr, the runtime filter do not have remote target or
     // not need local merge should regist here. the instance exec finish, the local
     // runtime filter mgr can release the memory of local runtime filter
@@ -881,6 +891,9 @@ class RuntimeState {
     mutable std::mutex _mc_commit_datas_mutex;
     std::vector<TMCCommitData> _mc_commit_datas;
 
+    mutable std::mutex _paimon_commit_datas_mutex;
+    std::vector<TPaimonCommitData> _paimon_commit_datas;
+
     std::vector<std::unique_ptr<doris::pipeline::PipelineXLocalStateBase>> _op_id_to_local_state;
 
     std::unique_ptr<doris::pipeline::PipelineXSinkLocalStateBase> _sink_local_state;

be/src/vec/sink/writer/async_result_writer.cpp

@@ -204,7 +204,11 @@ void AsyncResultWriter::process_block(RuntimeState* state, RuntimeProfile* opera
     Status st = Status::OK();
     { st = _writer_status.status(); }
 
+    LOG(INFO) << "AsyncResultWriter::process_block - before close: st.ok()=" << st.ok()
+              << ", st=" << st.to_string();
     Status close_st = close(st);
+    LOG(INFO) << "AsyncResultWriter::process_block - after close: close_st.ok()=" << close_st.ok()
+              << ", close_st=" << close_st.to_string();
     {
         // If it is already failed before, then not update the write status so that we could get
         // the real reason.