[feature](paimon) Paimon JNI write support

be/src/exec/operator/paimon_table_sink_operator.cpp

@@ -0,0 +1,161 @@
+// 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"
+#include "pipeline/exec/operator.h"
+
+namespace doris {
+
+Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) {
+    RETURN_IF_ERROR(Base::init(state, info));
+    // _writer is set lazily per (partition, bucket) in the operator,
+    // so we don't create a single _writer here.
+    return Status::OK();
+}
+
+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()));
+
+    if (in_block->rows() == 0) {
+        if (eos && !_eos_sent) {
+            _eos_sent = true;
+            for (auto& [key, writer] : _writers) {
+                RETURN_IF_ERROR(writer->sink(in_block, true));
+            }
+        }
+        return Status::OK();
+    }
+
+    // Project: apply output expressions
+    Block output_block;
+    {
+        SCOPED_TIMER(local_state.exec_time_counter());
+        RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(
+                _output_vexpr_ctxs, *in_block, &output_block, false));
+    }
+
+    // Compute (partition_bytes, bucket) for each row
+    std::vector<std::string> partition_bytes;
+    std::vector<int32_t> buckets;
+    RETURN_IF_ERROR(_compute_routing(output_block, partition_bytes, buckets));
+
+    // Split block by (partition_bytes, bucket)
+    auto grouped = _split_by_key(output_block, partition_bytes, buckets);
+
+    // Route each group to the corresponding writer
+    for (auto& [key, sub_block] : grouped) {
+        VPaimonTableWriter* writer = nullptr;
+        RETURN_IF_ERROR(_get_or_create_writer(state, key, &writer));
+        RETURN_IF_ERROR(writer->sink(sub_block.get(), false));
+    }
+
+    if (eos && !_eos_sent) {
+        _eos_sent = true;
+        Block empty_block;
+        for (auto& [key, writer] : _writers) {
+            RETURN_IF_ERROR(writer->sink(&empty_block, true));
+        }
+    }
+
+    return Status::OK();
+}
+
+Status PaimonTableSinkOperatorX::_get_or_create_writer(
+        RuntimeState* state, const PaimonPartitionBucketKey& key,
+        VPaimonTableWriter** writer) {
+    auto it = _writers.find(key);
+    if (it != _writers.end()) {
+        *writer = it->second.get();
+        return Status::OK();
+    }
+
+    const auto& paimon_sink = _t_output_expr; // need TDataSink reference
+    // Access TDataSink from the local state info
+    // We get it from the parent's init path
+    auto new_writer = std::make_shared<VPaimonTableWriter>(
+            _t_sink, _output_vexpr_ctxs, _dependency, _finish_dependency);
+    RETURN_IF_ERROR(new_writer->init_properties(_pool));
+    RETURN_IF_ERROR(new_writer->open(state, _operator_profile));
+
+    _writers[key] = new_writer;
+    *writer = new_writer.get();
+
+    LOG(INFO) << "Created writer for (partition=" << key.first.size() << "bytes"
+              << ", bucket=" << key.second << "), total writers: " << _writers.size();
+    return Status::OK();
+}
+
+Status PaimonTableSinkOperatorX::_compute_routing(
+        const Block& block,
+        std::vector<std::string>& partition_bytes,
+        std::vector<int32_t>& buckets) {
+    size_t rows = block.rows();
+    partition_bytes.resize(rows);
+    buckets.resize(rows);
+
+    // Compute bucket ids using BE-side MurmurHash
+    if (_routing_enabled && !_bucket_key_indices.empty()) {
+        buckets = paimon_compute_buckets(block, _bucket_key_indices, _total_buckets);
+    } else {
+        std::fill(buckets.begin(), buckets.end(), 0);
+    }
+
+    // Compute partition bytes: serialize partition col values to BinaryRow format.
+    // For now, use empty partition (Java SDK handles it internally).
+    // Full partition computation in BE will be added in a follow-up.
+    for (size_t i = 0; i < rows; ++i) {
+        partition_bytes[i].clear();
+    }
+
+    return Status::OK();
+}
+
+std::unordered_map<PaimonPartitionBucketKey, std::unique_ptr<Block>>
+PaimonTableSinkOperatorX::_split_by_key(
+        const Block& block,
+        const std::vector<std::string>& partition_bytes,
+        const std::vector<int32_t>& buckets) {
+    std::unordered_map<PaimonPartitionBucketKey, std::unique_ptr<Block>> result;
+
+    // Group row indices by (partition_bytes, bucket)
+    std::unordered_map<PaimonPartitionBucketKey, std::vector<size_t>> key_rows;
+    for (size_t i = 0; i < block.rows(); ++i) {
+        PaimonPartitionBucketKey key = {partition_bytes[i], buckets[i]};
+        key_rows[key].push_back(i);
+    }
+
+    // Build sub-blocks for each key
+    for (auto& [key, row_indices] : key_rows) {
+        auto sub_block = Block::create_unique(block.clone_empty());
+        auto columns = std::move(*sub_block).mutate_columns();
+        for (int col = 0; col < block.columns(); ++col) {
+            const auto& src_col = block.get_by_position(col).column;
+            for (size_t idx : row_indices) {
+                columns[col]->insert_from(*src_col, idx);
+            }
+        }
+        sub_block->set_columns(std::move(columns));
+        result[key] = std::move(sub_block);
+    }
+    return result;
+}
+
+} // namespace doris

be/src/exec/operator/paimon_table_sink_operator.h

@@ -0,0 +1,138 @@
+// 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 <unordered_map>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "exec/operator/operator.h"
+#include "exec/sink/writer/async_result_writer.h"
+#include "exec/sink/writer/paimon/paimon_bucket.h"
+#include "exec/sink/writer/paimon/vpaimon_table_writer.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+/// Key for per-(partition, bucket) writer grouping.
+/// partition_bytes is the serialized Paimon BinaryRow of partition values.
+/// bucket is the bucket id.
+using PaimonPartitionBucketKey = std::pair<std::string, int32_t>;
+
+/// Paimon table sink operator with per-(partition, bucket) routing.
+///
+/// Architecture:
+///   sink_impl(block):
+///     compute (partition_bytes, bucket) for each row
+///     split block by (p,b)
+///     for each (p,b): get_or_create_writer(p,b)->sink(sub_block)
+///
+/// Each (p,b) gets its own VPaimonTableWriter (and hence its own
+/// AsyncResultWriter + thread pool task + independent JNI/FFI object).
+class PaimonTableSinkOperatorX;
+
+class PaimonTableSinkLocalState final
+        : public AsyncWriterSink<VPaimonTableWriter, PaimonTableSinkOperatorX> {
+public:
+    using Base = AsyncWriterSink<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));
+        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;
+
+    /// Get or lazily create a writer for the given (partition, bucket).
+    Status _get_or_create_writer(RuntimeState* state,
+                                  const PaimonPartitionBucketKey& key,
+                                  VPaimonTableWriter** writer);
+
+    /// Compute partition bytes and bucket ids for all rows in the block.
+    /// partition computation serializes partition cols to Paimon BinaryRow format.
+    Status _compute_routing(const Block& block,
+                             std::vector<std::string>& partition_bytes,
+                             std::vector<int32_t>& buckets);
+
+    /// Split block by (partition_bytes, bucket) key.
+    std::unordered_map<PaimonPartitionBucketKey, std::unique_ptr<Block>> _split_by_key(
+            const Block& block,
+            const std::vector<std::string>& partition_bytes,
+            const std::vector<int32_t>& buckets);
+
+    const RowDescriptor& _row_desc;
+    VExprContextSPtrs _output_vexpr_ctxs;
+    const std::vector<TExpr>& _t_output_expr;
+    ObjectPool* _pool = nullptr;
+
+    // Per-(partition, bucket) writers. Lazily created.
+    std::unordered_map<PaimonPartitionBucketKey,
+                       std::shared_ptr<VPaimonTableWriter>> _writers;
+
+    // Bucket routing config from sink
+    std::vector<int> _bucket_key_indices;
+    std::vector<int> _partition_col_indices;
+    int32_t _total_buckets = 1;
+    bool _routing_enabled = false;
+
+    // EOS has been sent to all writers
+    bool _eos_sent = false;
+};
+
+} // namespace doris

be/src/exec/pipeline/pipeline_fragment_context.cpp

@@ -87,6 +87,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"
@@ -1371,6 +1372,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.");
@@ -2515,6 +2524,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,38 @@
+// 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& sink, RuntimeState* state) {
+    return Status::NotSupported(
+            "FFI (Rust) Paimon write backend is not yet implemented. "
+            "Currently only the JNI (Java) backend is available. "
+            "The FFI backend will be introduced in v2 for high-throughput append-only writes.");
+}
+
+Status FfiPaimonWriteBackend::create_writer(const std::string& partition_bytes, int32_t bucket,
+                                            std::unique_ptr<IPaimonWriter>* writer) {
+    return Status::NotSupported("FFI backend: create_writer not yet implemented");
+}
+
+Status FfiPaimonWriteBackend::create_committer(std::unique_ptr<IPaimonCommitter>* committer) {
+    return Status::NotSupported("FFI backend: create_committer not yet implemented");
+}
+
+} // namespace doris