branch-4.1: [feature](paimon) Support table writes via JNI and schema evolution

be/src/common/config.cpp

@@ -1552,6 +1552,11 @@ DEFINE_mInt64(hive_sink_max_file_size, "1073741824"); // 1GB
 /** Iceberg sink configurations **/
 DEFINE_mInt64(iceberg_sink_max_file_size, "1073741824"); // 1GB
 
+/** Paimon sink configurations **/
+DEFINE_mInt64(paimon_jni_writer_memory_pool_limit_bytes, "536870912"); // 512MB
+DEFINE_Validator(paimon_jni_writer_memory_pool_limit_bytes,
+                 [](int64_t bytes) -> bool { return bytes > 0; });
+
 // URI scheme to Doris file type mappings used by paimon-cpp DorisFileSystem.
 // Each entry uses the format "<scheme>=<file_type>", and file_type must be one of:
 // local, hdfs, s3, http, broker.

be/src/common/config.h

@@ -1641,6 +1641,10 @@ DECLARE_mInt64(hive_sink_max_file_size);
 /** Iceberg sink configurations **/
 DECLARE_mInt64(iceberg_sink_max_file_size);
 
+/** Paimon sink configurations **/
+// Hard upper bound for Doris-managed Paimon write-buffer memory per JNI writer.
+DECLARE_mInt64(paimon_jni_writer_memory_pool_limit_bytes);
+
 /** Paimon file system configurations **/
 DECLARE_Strings(paimon_file_system_scheme_mappings);
 

be/src/exec/operator/operator.cpp

@@ -63,6 +63,7 @@
 #include "exec/operator/olap_scan_operator.h"
 #include "exec/operator/olap_table_sink_operator.h"
 #include "exec/operator/olap_table_sink_v2_operator.h"
+#include "exec/operator/paimon_table_sink_operator.h"
 #include "exec/operator/partition_sort_sink_operator.h"
 #include "exec/operator/partition_sort_source_operator.h"
 #include "exec/operator/partitioned_aggregation_sink_operator.h"
@@ -802,6 +803,7 @@ DECLARE_OPERATOR(OlapTableSinkV2LocalState)
 DECLARE_OPERATOR(HiveTableSinkLocalState)
 DECLARE_OPERATOR(TVFTableSinkLocalState)
 DECLARE_OPERATOR(IcebergTableSinkLocalState)
+DECLARE_OPERATOR(PaimonTableSinkLocalState)
 DECLARE_OPERATOR(SpillIcebergTableSinkLocalState)
 DECLARE_OPERATOR(IcebergDeleteSinkLocalState)
 DECLARE_OPERATOR(IcebergMergeSinkLocalState)
@@ -929,6 +931,7 @@ template class AsyncWriterSink<doris::VIcebergDeleteSink, IcebergDeleteSinkOpera
 template class AsyncWriterSink<doris::VIcebergMergeSink, IcebergMergeSinkOperatorX>;
 template class AsyncWriterSink<doris::VMCTableWriter, MCTableSinkOperatorX>;
 template class AsyncWriterSink<doris::VTVFTableWriter, TVFTableSinkOperatorX>;
+template class AsyncWriterSink<doris::PaimonTableWriter, PaimonTableSinkOperatorX>;
 
 #ifdef BE_TEST
 template class OperatorX<DummyOperatorLocalState>;

be/src/exec/operator/paimon_table_sink_operator.cpp

@@ -0,0 +1,39 @@
+// 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/paimon_table_sink_operator.h"
+
+#include "common/logging.h"
+
+namespace doris {
+
+Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) {
+    return Base::init(state, info);
+}
+
+Status PaimonTableSinkOperatorX::sink_impl(RuntimeState* state, Block* in_block, bool eos) {
+    auto& local_state = get_local_state(state);
+    SCOPED_TIMER(local_state.exec_time_counter());
+    COUNTER_UPDATE(local_state.rows_input_counter(), static_cast<int64_t>(in_block->rows()));
+
+    // Delegate to AsyncWriterSink → PaimonTableWriter for this pipeline instance.
+    // Each pipeline instance has its own writer session; partition and bucket
+    // routing is handled internally by the Paimon SDK inside IPaimonWriter::write().
+    return local_state.sink(state, in_block, eos);
+}
+
+} // namespace doris

be/src/exec/operator/paimon_table_sink_operator.h

@@ -0,0 +1,104 @@
+// 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 <gen_cpp/DataSinks_types.h>
+
+#include <memory>
+#include <string>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "exec/operator/operator.h"
+#include "exec/sink/writer/paimon/paimon_table_writer.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+/// Paimon table sink operator — simple pass-through to AsyncWriterSink.
+///
+/// Each pipeline instance (LocalState) owns one PaimonTableWriter, which in
+/// turn owns one IPaimonWriteBackend + IPaimonWriter. Pipeline parallelism
+/// determines the number of concurrent Paimon writer sessions per table.
+///
+/// Partition and bucket routing is performed internally by the Paimon SDK
+/// (Java via JNI, or Rust via FFI). Doris does not compute partition values
+/// or bucket ids; it passes complete Blocks through the backend to the SDK,
+/// where each row is routed via getPartition(row) + getBucket(row).
+///
+/// This mirrors Iceberg's approach: IcebergTableSinkOperatorX delegates to
+/// AsyncWriterSink<VIcebergTableWriter>, with partition routing inside
+/// VIcebergTableWriter::write().
+class PaimonTableSinkOperatorX;
+
+class PaimonTableSinkLocalState final
+        : public AsyncWriterSink<PaimonTableWriter, PaimonTableSinkOperatorX> {
+public:
+    using Base = AsyncWriterSink<PaimonTableWriter, 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));
+        DCHECK(thrift_sink.__isset.paimon_table_sink);
+        RETURN_IF_ERROR(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(VExpr::prepare(_output_vexpr_ctxs, state, _row_desc));
+        return VExpr::open(_output_vexpr_ctxs, state);
+    }
+
+    Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override;
+
+private:
+    friend class PaimonTableSinkLocalState;
+    template <typename Writer, typename Parent>
+        requires(std::is_base_of_v<AsyncResultWriter, Writer>)
+    friend class AsyncWriterSink;
+
+    const RowDescriptor& _row_desc;
+    VExprContextSPtrs _output_vexpr_ctxs;
+    const std::vector<TExpr>& _t_output_expr;
+    ObjectPool* _pool = nullptr;
+};
+
+} // namespace doris

be/src/exec/pipeline/pipeline_fragment_context.cpp

@@ -86,6 +86,7 @@
 #include "exec/operator/olap_scan_operator.h"
 #include "exec/operator/olap_table_sink_operator.h"
 #include "exec/operator/olap_table_sink_v2_operator.h"
+#include "exec/operator/paimon_table_sink_operator.h"
 #include "exec/operator/partition_sort_sink_operator.h"
 #include "exec/operator/partition_sort_source_operator.h"
 #include "exec/operator/partitioned_aggregation_sink_operator.h"
@@ -1157,6 +1158,14 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS
                                                        output_exprs);
         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::JDBC_TABLE_SINK: {
         if (!thrift_sink.__isset.jdbc_table_sink) {
             return Status::InternalError("Missing data jdbc sink.");
@@ -2175,6 +2184,20 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
         }
     }
 
+    if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) {
+        params.__isset.paimon_commit_messages = true;
+        params.paimon_commit_messages.insert(params.paimon_commit_messages.end(), pcm.begin(),
+                                             pcm.end());
+    } else if (!req.runtime_states.empty()) {
+        for (auto* rs : req.runtime_states) {
+            if (auto rs_pcm = rs->paimon_commit_messages(); !rs_pcm.empty()) {
+                params.__isset.paimon_commit_messages = true;
+                params.paimon_commit_messages.insert(params.paimon_commit_messages.end(),
+                                                     rs_pcm.begin(), rs_pcm.end());
+            }
+        }
+    }
+
     req.runtime_state->get_unreported_errors(&(params.error_log));
     params.__isset.error_log = (!params.error_log.empty());
 

be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.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 "exec/sink/writer/paimon/ffi_paimon_write_backend.h"
+
+namespace doris {
+
+Status FfiPaimonWriteBackend::open(const TPaimonTableSink&, RuntimeState*, RuntimeProfile*) {
+    return Status::NotSupported("Paimon Rust FFI writer is not implemented");
+}
+
+Status FfiPaimonWriteBackend::create_writer(std::unique_ptr<IPaimonWriter>*) {
+    return Status::NotSupported("Paimon Rust FFI writer is not implemented");
+}
+
+Status FfiPaimonWriteBackend::close() {
+    return Status::OK();
+}
+
+} // namespace doris

be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.h

@@ -0,0 +1,36 @@
+// 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 "exec/sink/writer/paimon/paimon_write_backend.h"
+
+namespace doris {
+
+/// Placeholder for the future paimon-rust writer implementation. Keeping this
+/// backend in the factory makes the integration boundary explicit without
+/// introducing a BE commit contract that the Rust writer will not own.
+class FfiPaimonWriteBackend final : public IPaimonWriteBackend {
+public:
+    Status open(const TPaimonTableSink& sink, RuntimeState* state,
+                RuntimeProfile* profile) override;
+    Status create_writer(std::unique_ptr<IPaimonWriter>* writer) override;
+    Status close() override;
+    PaimonBackendType type() const override { return PaimonBackendType::FFI; }
+};
+
+} // namespace doris