This is an automated email from the ASF dual-hosted git repository.
suxiaogang223 pushed a commit to branch feature/paimon-jni-write-v1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/feature/paimon-jni-write-v1 by
this push:
new 1341369a91e [improvement](paimon) Optimize PaimonJniWriter and cleanup
dead code
1341369a91e is described below
commit 1341369a91ea790769182f9cd60622780596251d
Author: Socrates <[email protected]>
AuthorDate: Mon Jul 13 15:42:21 2026 +0800
[improvement](paimon) Optimize PaimonJniWriter and cleanup dead code
### What problem does this PR solve?
Problem Summary:
1. PaimonJniWriter.writeBatch() had per-cell instanceof dispatch via
readArrowValue(), causing ~819200 instanceof checks per batch
(4096 rows × 20 columns × ~10 instanceof per cell).
2. Dead code accumulated after the single-writer architecture refactor:
- paimon_bucket.h/.cpp (BE-side MurmurHash routing, now delegated to SDK)
- PhysicalPaimonTableSink bucked-shuffle logic (unused GATHER override)
- PaimonTableSink bucketNum/bucketKeyIndices Thrift payload
- Thrift TPaimonTableSink.bucket_num / bucket_key_indices fields
- Thrift TPaimonCommitMessage.partition / bucket fields
3. Missing DECLARE_OPERATOR(PaimonTableSinkLocalState) in operator.cpp
Root cause:
- writeBatch bottleneck: per-cell instanceof chain for every Arrow value
- cleanup: v0 per-bucket routing infrastructure no longer needed after
switching to SDK-internal routing
Fix:
- writeBatch: column-major typed extraction (1 instanceof per column),
group by SDK getPartition()/getBucket(), writeBundle() per group
- Remove paimon_bucket.h/.cpp (222+71 lines)
- Remove PhysicalPaimonTableSink bucket shuffle logic
- Remove PaimonTableSink bucketNum/bucketKeyIndices Thrift fields
- Simplify Thrift: TPaimonTableSink (remove fields 6,11),
TPaimonCommitMessage (remove fields 1,2, keep payload)
- Add DECLARE_OPERATOR(PaimonTableSinkLocalState)
- Stop tracking PAIMON_WRITE_DESIGN.md in git
### Release note
None
### Check List (For Author)
- Test: FE Maven compile + checkstyle verified
- Behavior changed: Yes (writeBatch optimized, dead code removed)
- Does this need documentation: No
---
be/src/exec/operator/operator.cpp | 1 +
be/src/exec/sink/writer/paimon/paimon_bucket.cpp | 222 --
be/src/exec/sink/writer/paimon/paimon_bucket.h | 71 -
.../exec/sink/writer/paimon/vpaimon_table_writer.h | 3 +-
.../paimon-scanner/PAIMON_WRITE_DESIGN.md | 2139 --------------------
.../org/apache/doris/paimon/PaimonJniWriter.java | 261 ++-
.../plans/physical/PhysicalPaimonTableSink.java | 31 +-
.../org/apache/doris/planner/PaimonTableSink.java | 29 +-
gensrc/thrift/DataSinks.thrift | 16 +-
9 files changed, 210 insertions(+), 2563 deletions(-)
diff --git a/be/src/exec/operator/operator.cpp
b/be/src/exec/operator/operator.cpp
index 1f255078b5d..d88f38b64fe 100644
--- a/be/src/exec/operator/operator.cpp
+++ b/be/src/exec/operator/operator.cpp
@@ -833,6 +833,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)
diff --git a/be/src/exec/sink/writer/paimon/paimon_bucket.cpp
b/be/src/exec/sink/writer/paimon/paimon_bucket.cpp
deleted file mode 100644
index ae07824aa53..00000000000
--- a/be/src/exec/sink/writer/paimon/paimon_bucket.cpp
+++ /dev/null
@@ -1,222 +0,0 @@
-// 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/paimon_bucket.h"
-
-#include <cstring>
-
-#include "core/column/column.h"
-#include "core/column/column_nullable.h"
-#include "core/column/column_vector.h"
-
-namespace doris {
-
-// ────────────────────────────────────────────────────────────
-// Murmur3-32 hash (seed=42, word-aligned input)
-// Matches Paimon's MurmurHashUtils.hashBytesByWords
-// ────────────────────────────────────────────────────────────
-
-static constexpr uint32_t MURMUR_C1 = 0xcc9e2d51;
-static constexpr uint32_t MURMUR_C2 = 0x1b873593;
-static constexpr uint32_t MURMUR_SEED = 42;
-
-static uint32_t murmur_mix_k1(uint32_t k1) {
- k1 *= MURMUR_C1;
- k1 = (k1 << 15) | (k1 >> 17); // rotate_left(15)
- k1 *= MURMUR_C2;
- return k1;
-}
-
-static uint32_t murmur_mix_h1(uint32_t h1, uint32_t k1) {
- h1 ^= k1;
- h1 = (h1 << 13) | (h1 >> 19); // rotate_left(13)
- h1 = h1 * 5 + 0xe6546b64;
- return h1;
-}
-
-static uint32_t murmur_fmix(uint32_t h) {
- h ^= h >> 16;
- h *= 0x85ebca6b;
- h ^= h >> 13;
- h *= 0xc2b2ae35;
- h ^= h >> 16;
- return h;
-}
-
-int32_t paimon_murmur_hash(const uint8_t* data, size_t len) {
- DCHECK(len % 4 == 0) << "Murmur hash requires word-aligned input, got " <<
len;
- uint32_t h1 = MURMUR_SEED;
- for (size_t i = 0; i < len; i += 4) {
- uint32_t word = static_cast<uint32_t>(data[i]) |
(static_cast<uint32_t>(data[i + 1]) << 8) |
- (static_cast<uint32_t>(data[i + 2]) << 16) |
- (static_cast<uint32_t>(data[i + 3]) << 24);
- uint32_t k1 = murmur_mix_k1(word);
- h1 = murmur_mix_h1(h1, k1);
- }
- return static_cast<int32_t>(murmur_fmix(h1 ^ static_cast<uint32_t>(len)));
-}
-
-// ────────────────────────────────────────────────────────────
-// Binary row construction for bucket key hashing
-// ────────────────────────────────────────────────────────────
-
-/// Compute Paimon BinaryRow null-bit-set width.
-/// Matches BinaryRow.calBitSetWidthInBytes(arity).
-static inline int32_t cal_bit_set_width(int32_t arity) {
- return ((arity + 63 + 8 /* HEADER_SIZE_IN_BYTES */) / 64) * 8;
-}
-
-/// Compute the fixed-size portion of a Paimon BinaryRow in bytes.
-static inline int32_t fixed_row_size(int32_t arity) {
- return cal_bit_set_width(arity) + arity * 8;
-}
-
-/// Write a null bit at the given position.
-/// Matches BinaryRowBuilder.setNullAt(pos).
-static void set_null_bit(uint8_t* data, int32_t pos, bool is_null) {
- int32_t bit_index = pos + 8; // HEADER_SIZE_IN_BYTES
- int32_t byte_idx = bit_index / 8;
- int32_t bit_off = bit_index % 8;
- if (is_null) {
- data[byte_idx] |= (1 << bit_off);
- }
-}
-
-// ────────────────────────────────────────────────────────────
-// Per-column value extraction helpers
-// ────────────────────────────────────────────────────────────
-
-namespace {
-
-/// Write a single column value at `row_idx` into the binary row at the
-/// correct fixed-field offset. Returns the number of bytes actually
-/// written (always writes into an 8-byte slot at field_offset).
-void write_col_to_row(uint8_t* row_data, int32_t null_bytes_size, int32_t
field_pos,
- const ColumnPtr& col, size_t row_idx) {
- int32_t field_offset = null_bytes_size + field_pos * 8;
-
- // Handle nullable column: unwrap the data column
- const IColumn* data_col = col.get();
- bool is_null = false;
- if (col->is_nullable()) {
- auto* nullable = assert_cast<const ColumnNullable*>(col.get());
- is_null = nullable->is_null_at(row_idx);
- data_col = &nullable->get_nested_column();
- }
-
- set_null_bit(row_data, field_pos, is_null);
- if (is_null) {
- return; // field already zeroed from initialization
- }
-
- // Write value in little-endian into the field slot
- PrimitiveType ptype = data_col->get_data_type();
-
- if (ptype == TYPE_TINYINT || ptype == TYPE_BOOLEAN) {
- auto val = assert_cast<const
ColumnVector<Int8>*>(data_col)->get_element(row_idx);
- row_data[field_offset] = static_cast<uint8_t>(val);
- } else if (ptype == TYPE_SMALLINT) {
- auto val = assert_cast<const
ColumnVector<Int16>*>(data_col)->get_element(row_idx);
- std::memcpy(row_data + field_offset, &val, sizeof(val));
- } else if (ptype == TYPE_INT || ptype == TYPE_DATE || ptype ==
TYPE_DATEV2) {
- auto val = assert_cast<const
ColumnVector<Int32>*>(data_col)->get_element(row_idx);
- std::memcpy(row_data + field_offset, &val, sizeof(val));
- } else if (ptype == TYPE_BIGINT || ptype == TYPE_DATETIME || ptype ==
TYPE_DATETIMEV2) {
- auto val = assert_cast<const
ColumnVector<Int64>*>(data_col)->get_element(row_idx);
- std::memcpy(row_data + field_offset, &val, sizeof(val));
- } else if (ptype == TYPE_FLOAT) {
- float f = assert_cast<const
ColumnVector<Float32>*>(data_col)->get_element(row_idx);
- int32_t bits;
- std::memcpy(&bits, &f, sizeof(bits));
- std::memcpy(row_data + field_offset, &bits, sizeof(bits));
- } else if (ptype == TYPE_DOUBLE) {
- double d = assert_cast<const
ColumnVector<Float64>*>(data_col)->get_element(row_idx);
- int64_t bits;
- std::memcpy(&bits, &d, sizeof(bits));
- std::memcpy(row_data + field_offset, &bits, sizeof(bits));
- } else if (ptype == TYPE_DECIMAL32) {
- auto val = assert_cast<const
ColumnVector<Int32>*>(data_col)->get_element(row_idx);
- std::memcpy(row_data + field_offset, &val, sizeof(val));
- } else if (ptype == TYPE_DECIMAL64) {
- auto val = assert_cast<const
ColumnVector<Int64>*>(data_col)->get_element(row_idx);
- std::memcpy(row_data + field_offset, &val, sizeof(val));
- } else if (ptype == TYPE_DECIMAL128 || ptype == TYPE_DECIMAL128I) {
- auto val = assert_cast<const
ColumnVector<Int128>*>(data_col)->get_element(row_idx);
- std::memcpy(row_data + field_offset, &val, sizeof(val));
- } else {
- // For complex types (STRING, ARRAY, MAP, etc.), we cannot compute
- // the correct binary row in BE. Fallback to single-bucket routing.
- // Callers should check this before calling.
- LOG(WARNING) << "Unsupported bucket key type for BE-side computation: "
- << static_cast<int>(ptype);
- }
-}
-
-} // anonymous namespace
-
-// ────────────────────────────────────────────────────────────
-// Public API
-// ────────────────────────────────────────────────────────────
-
-int32_t paimon_compute_bucket(const Block& block, size_t row_idx,
- const std::vector<int>& bucket_key_cols, int32_t
num_buckets) {
- if (num_buckets <= 1 || bucket_key_cols.empty()) {
- return 0;
- }
-
- int32_t arity = static_cast<int32_t>(bucket_key_cols.size());
- int32_t row_size = fixed_row_size(arity);
- std::vector<uint8_t> row_data(row_size, 0);
- int32_t null_bytes_size = cal_bit_set_width(arity);
-
- for (int32_t i = 0; i < arity; ++i) {
- const auto& col = block.get_by_position(bucket_key_cols[i]).column;
- write_col_to_row(row_data.data(), null_bytes_size, i, col, row_idx);
- }
-
- int32_t hash = paimon_murmur_hash(row_data.data(), row_size);
- int32_t bucket = hash % num_buckets;
- return bucket < 0 ? bucket + num_buckets : bucket;
-}
-
-std::vector<int32_t> paimon_compute_buckets(const Block& block,
- const std::vector<int>&
bucket_key_cols,
- int32_t num_buckets) {
- if (num_buckets <= 1 || bucket_key_cols.empty()) {
- return std::vector<int32_t>(block.rows(), 0);
- }
-
- std::vector<int32_t> buckets(block.rows());
- int32_t arity = static_cast<int32_t>(bucket_key_cols.size());
- int32_t row_size = fixed_row_size(arity);
- std::vector<uint8_t> row_data(row_size, 0);
- int32_t null_bytes_size = cal_bit_set_width(arity);
-
- for (size_t row = 0; row < block.rows(); ++row) {
- std::memset(row_data.data(), 0, row_size);
- for (int32_t i = 0; i < arity; ++i) {
- const auto& col = block.get_by_position(bucket_key_cols[i]).column;
- write_col_to_row(row_data.data(), null_bytes_size, i, col, row);
- }
- int32_t hash = paimon_murmur_hash(row_data.data(), row_size);
- int32_t bucket = hash % num_buckets;
- buckets[row] = bucket < 0 ? bucket + num_buckets : bucket;
- }
- return buckets;
-}
-
-} // namespace doris
diff --git a/be/src/exec/sink/writer/paimon/paimon_bucket.h
b/be/src/exec/sink/writer/paimon/paimon_bucket.h
deleted file mode 100644
index 3283b34c6f7..00000000000
--- a/be/src/exec/sink/writer/paimon/paimon_bucket.h
+++ /dev/null
@@ -1,71 +0,0 @@
-// 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 <cstdint>
-#include <string>
-#include <vector>
-
-#include "core/block/block.h"
-#include "core/column/column.h"
-
-namespace doris {
-
-/// Paimon-compatible bucket computation.
-///
-/// Implements the same bucket assignment logic as Paimon's Java SDK:
-/// Default: Math.abs(BinaryRow(murmur_hash) % numBuckets)
-/// Mod: Math.floorMod(bucket_key_value, numBuckets)
-///
-/// The binary row format matches Paimon's BinaryRow layout:
-/// - null_bits_size bytes of null bit set (always 8-byte aligned,
-/// includes conceptual 8-byte header)
-/// - arity * 8 bytes of fixed-length field slots
-///
-/// Each field occupies 8 bytes in the fixed part, with values
-/// written in little-endian order within their actual type width.
-
-/// Murmur3-32 hash over word-aligned data, matching Paimon's
-/// MurmurHashUtils.hashBytesByWords with seed=42.
-int32_t paimon_murmur_hash(const uint8_t* data, size_t len);
-
-/// Compute bucket id for a block row.
-///
-/// @param block input block containing all columns
-/// @param row_idx row index in the block
-/// @param bucket_key_cols indices of bucket key columns in the block
-/// @param num_buckets total number of buckets
-/// @return bucket id in [0, num_buckets)
-int32_t paimon_compute_bucket(const Block& block, size_t row_idx,
- const std::vector<int>& bucket_key_cols, int32_t
num_buckets);
-
-/// Compute bucket ids for all rows in a block.
-///
-/// @return vector of bucket ids, one per row
-std::vector<int32_t> paimon_compute_buckets(const Block& block,
- const std::vector<int>&
bucket_key_cols,
- int32_t num_buckets);
-
-/// Simple mod bucket for a single integral key.
-/// Matches Paimon's ModBucketFunction.
-inline int32_t paimon_bucket_mod(int64_t value, int32_t num_buckets) {
- int32_t r = static_cast<int32_t>(value % num_buckets);
- return r < 0 ? r + num_buckets : r;
-}
-
-} // namespace doris
diff --git a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h
b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h
index d4ce5b04bde..b54d254ac9a 100644
--- a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h
+++ b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h
@@ -85,8 +85,7 @@ private:
// Single writer managing all partitions and buckets (SDK-internal routing)
std::unique_ptr<IPaimonWriter> _writer;
- // Statistics
- int64_t _written_rows = 0;
+ // Statistics (note: _written_rows is inherited from ResultWriter)
int64_t _written_bytes = 0;
// Profile counters
diff --git a/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md
b/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md
deleted file mode 100644
index 324a38704be..00000000000
--- a/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md
+++ /dev/null
@@ -1,2139 +0,0 @@
-# Doris Paimon 写入功能架构设计
-
-> **设计原则**：从第一性原理出发，基于 Paimon Java SDK（功能完备）和 Paimon Rust（性能优先）的差异，设计一个统一的写入框架。
->
-> **核心策略**：Rust writer 和 Java writer 是**互相替代**的关系，Rust 是 Java 的功能子集。优先使用
Rust（性能），功能缺失时回退 Java。首个版本仅引入 Java JNI writer，但在架构上必须提前为 Rust FFI writer 预留接口。
-
-## 目录
-
-1. [背景与目标](#1-背景与目标)
-2. [Paimon 写入模型分析](#2-paimon-写入模型分析)
-3. [Java SDK vs Rust 差异矩阵](#3-java-sdk-vs-rust-差异矩阵)
-4. [Doris 现有写入框架分析](#4-doris-现有写入框架分析)
-5. [架构设计](#5-架构设计)
-6. [接口定义](#6-接口定义)
-7. [数据流与交互时序](#7-数据流与交互时序)
-8. [FE 端设计](#8-fe-端设计)
-9. [BE 端设计](#9-be-端设计)
-10. [功能路由矩阵](#10-功能路由矩阵)
-11. [实施路线图](#11-实施路线图)
-
----
-
-## 1. 背景与目标
-
-### 1.1 背景
-
-Doris 需要支持向 Paimon 表写入数据。目前有两个可用的 Paimon 客户端库：
-
-- **Paimon Java SDK** (`paimon-core`)：功能最完整的参考实现，支持全量的写入、提交、Compaction、Schema
Evolution、Lookup Changelog Producer、Deletion Vector 等功能。
-- **Paimon Rust** (`paimon-rust`)：高性能实现，使用 Arrow 原生格式，零拷贝，异步 I/O。但目前功能较 Java
版本有差距（缺少 Compaction、Lookup/Full-compaction Changelog Producer、部分 MergeEngine 与
Deletion Vector 的组合等）。
-
-### 1.2 目标
-
-1. **v1（当前）**：基于 Java SDK 通过 JNI 实现功能完备的 Paimon 写入。同时建立 `IPaimonWriteBackend`
抽象接口，为 Rust 后端预留设计空间。
-2. **v2（中期）**：引入 Paimon Rust 通过 FFI 加速写入。Rust 覆盖 Append-only +
固定桶等高吞吐场景，功能缺失时自动回退 Java。
-3. **长期**：随着 Paimon Rust 功能完善，逐步扩大 Rust 后端的覆盖范围，最终在大多数场景下优先使用 Rust。
-4. **架构目标**：`IPaimonWriteBackend` 是核心抽象——Java 和 Rust 是平等的后端实现，可以互相替代，上层（Doris
Sink）不感知差异。
-
-### 1.3 关键约束
-
-- Paimon C++ native 版本（paimon-cpp）未来会被删除，不考虑
-- Java SDK 通过 JNI 调用，Rust 通过 C FFI 调用
-- Doris BE 是 C++ 代码，需要通过 JNI（调用 Java）或 FFI（调用 Rust）来桥接
-- 写入流程包括：数据写入 → CommitMessage 生成 → Snapshot 提交
-- FE 端需要感知表类型（Paimon）来做查询规划和写入协调
-
----
-
-## 2. Paimon 写入模型分析
-
-### 2.1 核心写入模型（两阶段提交）
-
-Paimon 的写入模型是一个**两阶段提交**过程，无论 Java 还是 Rust 都遵循相同的抽象：
-
-```
-阶段1：Write（分布式）
- WriteBuilder → newWrite() → TableWrite
- → write(row/batch) × N // 每个 worker 写自己的数据
- → prepareCommit() // 关闭文件，生成 CommitMessage[]
- → CommitMessage[] // 序列化传递给 Coordinator
-
-阶段2：Commit（集中式）
- WriteBuilder → newCommit() → TableCommit
- → commit(CommitMessage[]) // 原子提交：写 Manifest → 创建 Snapshot
- → abort(CommitMessage[]) // 失败时清理数据文件
-```
-
-### 2.2 Java SDK 核心接口
-
-#### WriteBuilder（工厂）
-
-```java
-// 批写入构建器
-public interface BatchWriteBuilder extends WriteBuilder {
- BatchWriteBuilder withOverwrite();
- BatchWriteBuilder withOverwrite(Map<String, String> staticPartition);
- BatchTableWrite newWrite();
- BatchTableCommit newCommit();
-}
-
-// 流写入构建器
-public interface StreamWriteBuilder extends WriteBuilder {
- StreamWriteBuilder withCommitUser(String commitUser);
- StreamTableWrite newWrite();
- StreamTableCommit newCommit();
-}
-```
-
-**关键点**：`WriteBuilder` 同时创建 `TableWrite` 和 `TableCommit`。同一个 `WriteBuilder`
实例产出的 write 和 commit 共享 `commitUser`（用于幂等提交）。
-
-#### TableWrite（数据写入）
-
-```java
-public interface TableWrite extends AutoCloseable {
- TableWrite withIOManager(IOManager ioManager);
- TableWrite withWriteType(RowType writeType);
- TableWrite withMemoryPoolFactory(MemoryPoolFactory factory);
-
- // 分区/桶计算
- BinaryRow getPartition(InternalRow row);
- int getBucket(InternalRow row);
-
- // 逐行写入
- void write(InternalRow row) throws Exception;
- void write(InternalRow row, int bucket) throws Exception;
-
- // 批量写入（高效路径）
- void writeBundle(BinaryRow partition, int bucket, BundleRecords bundle)
throws Exception;
-
- // 压缩
- void compact(BinaryRow partition, int bucket, boolean fullCompaction)
throws Exception;
-
- // 生成提交消息（在 TableWriteImpl 上，非接口方法）
- // List<CommitMessage> prepareCommit(boolean waitCompaction, long
commitIdentifier)
-}
-```
-
-**关键设计点**：
-- `writeBundle()` 是高性能批量写入路径：调用者预先计算好 partition 和 bucket，将一批行打包为
`BundleRecords` 直接写入
-- `compact()` 是显式的 bucket 级压缩触发
-- `prepareCommit()` 返回 `List<CommitMessage>`，每个 `CommitMessage` 对应一个
(partition, bucket) 的新文件列表
-- `CommitMessage` 是 `Serializable` 的，可以跨网络传输
-
-#### CommitMessage
-
-```java
-public interface CommitMessage extends Serializable {
- BinaryRow partition(); // 分区键（二进制编码）
- int bucket(); // 桶编号
- Integer totalBuckets(); // 该分区的总桶数
-}
-```
-
-> **注意**：Java 的 `CommitMessage` 接口只暴露了 partition 和 bucket。实际的文件列表（newFiles,
changelogFiles, indexFiles 等）在实现类 `CommitMessageImpl` 中，通过
`CommitMessageSerializer` 序列化。这是一种封装设计——调用者不需要知道内部文件细节。
-
-#### BatchTableCommit
-
-```java
-public interface BatchTableCommit extends TableCommit {
- void commit(List<CommitMessage> commitMessages);
- void truncateTable();
- void truncatePartitions(List<Map<String, String>> partitionSpecs);
- void updateStatistics(Statistics statistics);
- void compactManifests();
-}
-```
-
-### 2.3 Rust 核心接口
-
-Rust 采用更直接的 API 设计：
-
-```rust
-// 工厂
-pub struct WriteBuilder<'a> { table, commit_user, overwrite }
-impl WriteBuilder {
- pub fn new_write(&self) -> Result<TableWrite>;
- pub fn new_commit(&self) -> TableCommit;
- pub fn with_overwrite(self) -> Self;
- pub fn with_commit_user(self, user) -> Result<Self>;
-}
-
-// 写入
-pub struct TableWrite { ... }
-impl TableWrite {
- pub fn with_overwrite(self) -> Self;
- pub async fn write_arrow_batch(&mut self, batch: &RecordBatch) ->
Result<()>;
- pub async fn write_arrow(&mut self, batches: &[RecordBatch]) -> Result<()>;
- pub async fn prepare_commit(&mut self) -> Result<Vec<CommitMessage>>;
-}
-
-// 提交
-pub struct TableCommit { ... }
-impl TableCommit {
- pub async fn commit(&self, messages: Vec<CommitMessage>) -> Result<()>;
- pub async fn overwrite(&self, messages: Vec<CommitMessage>,
- static_partitions: Option<HashMap<String,
Option<Datum>>>) -> Result<()>;
- pub async fn truncate_table(&self) -> Result<()>;
- pub async fn truncate_partitions(&self, partitions: Vec<HashMap<String,
Option<Datum>>>) -> Result<()>;
- pub async fn abort(&self, messages: &[CommitMessage]) -> Result<()>;
-}
-
-// 提交消息（具体结构体，所有字段公开）
-pub struct CommitMessage {
- pub partition: Vec<u8>,
- pub bucket: i32,
- pub new_files: Vec<DataFileMeta>,
- pub new_changelog_files: Vec<DataFileMeta>,
- pub new_index_files: Vec<IndexFileMeta>,
- pub deleted_index_files: Vec<IndexFileMeta>,
- pub deleted_files: Vec<DataFileMeta>,
- pub check_from_snapshot: Option<i64>,
-}
-```
-
-**关键差异**：
-- Rust 的 `CommitMessage` 是**具体结构体**，所有字段直接暴露，而 Java 通过序列化器隐藏内部细节
-- Rust 使用 Arrow `RecordBatch` 作为数据载体，Java 使用 `InternalRow`
-- Rust 的写入是 async 的，Java 是同步的（但在 Flink 等框架中被异步调用）
-
-### 2.4 写入功能全景
-
-| 功能 | Java SDK | Rust | 说明 |
-|------|----------|------|------|
-| **Append-only 写入** | ✅ | ✅ | 无主键表的顺序写入 |
-| **主键表写入** | ✅ | ✅ | Sorted Merge Tree，按主键排序去重 |
-| **固定桶 (Fixed Bucket)** | ✅ | ✅ | hash(bucket_key) % N |
-| **动态桶 (Dynamic Bucket)** | ✅ | ✅ | bucket=-1，自动分配桶 |
-| **跨分区桶 (Cross Partition)** | ✅ | ✅ | 动态桶 + 跨分区主键 |
-| **Postpone 桶 (bucket=-2)** | ✅ | ✅ | 延迟桶分配，用于高吞吐写入 |
-| **MergeEngine: Normal** | ✅ | ✅ | 最后写入胜出 |
-| **MergeEngine: PartialUpdate** | ✅ | ✅ | 部分列更新（Rust 不支持 +DV） |
-| **MergeEngine: Aggregation** | ✅ | ✅ | 聚合函数（Rust 不支持 +DV） |
-| **MergeEngine: FirstRow** | ✅ | ✅ | 保留第一行 |
-| **MergeEngine: Deduplicate** | ✅ | ✅ | 按主键去重 |
-| **ChangelogProducer: None** | ✅ | ✅ | 不产生 changelog |
-| **ChangelogProducer: Input** | ✅ | ✅ | 透传 changelog |
-| **ChangelogProducer: Lookup** | ✅ | ❌ | 提交时查找旧值生成 changelog |
-| **ChangelogProducer: FullCompaction** | ✅ | ❌ | 全量压缩时生成 changelog |
-| **Compaction** | ✅ | ❌ | 自动/手动合并，层级压缩 |
-| **Deletion Vector** | ✅ | 部分 | Rust: 不支持 PartialUpdate+DV、Aggregation+DV |
-| **Schema Evolution** | ✅ | ✅ | 列添加/删除/重命名/类型变更 |
-| **Row Tracking** | ✅ | ✅ | DataEvolution 的行 ID 追踪 |
-| **DataEvolution (MERGE INTO)** | ✅ | ✅ | 基于行 ID 的部分列更新 |
-| **CoW MERGE INTO** | ✅ | ✅ | 无 PK 表的 Copy-on-Write 合并 |
-| **分区覆盖 (Overwrite)** | ✅ | ✅ | 静态/动态分区覆盖 |
-| **Truncate** | ✅ | ✅ | 清空表/分区 |
-| **Blob 类型** | ✅ | ✅ | 大对象列分离存储 |
-| **全局索引 (Global Index)** | ✅ | 部分 | Rust 仅支持 DropPartitionIndex |
-| **二次索引 (BTree Index)** | ✅ | ✅ | |
-| **文件格式: Parquet** | ✅ | ✅ | 默认 |
-| **文件格式: ORC** | ✅ | ✅ | |
-| **文件格式: Avro** | ✅ | ✅ | |
-| **文件格式: Row** | ✅ | ✅ | Arrow IPC 格式 |
-| **文件格式: Vortex** | ❌ | ✅ (feature) | Rust 特有 |
-| **Stream Writer (持续写入)** | ✅ | ❌ | Flink 流式写入模式 |
-| **Tag/Branch** | ✅ | 部分 | Rust 有 Tag 读取器但无创建 API |
-| **提交重试与幂等** | ✅ | ✅ | commit_user + commitIdentifier |
-
----
-
-## 3. Java SDK vs Rust 差异矩阵
-
-### 3.1 架构差异
-
-| 维度 | Java SDK | Rust |
-|------|----------|------|
-| **数据载体** | `InternalRow` (自定义行格式) | Arrow `RecordBatch` (列式) |
-| **API 粒度** | 逐行 `write(InternalRow)` + 批量 `writeBundle()` | 批量
`write_arrow_batch(RecordBatch)` |
-| **CommitMessage** | 接口 + 序列化器（封装实现） | 具体结构体（所有字段公开） |
-| **I/O 模型** | 同步（阻塞 I/O） | 异步（tokio + opendal） |
-| **内存管理** | Java GC + 直接内存（spillable） | Rust 所有权系统 |
-| **压缩** | 内置 `CompactManager` | 未实现 |
-| **可扩展性** | 通过 SPI 插件 | 通过 feature flags 和 traits |
-
-### 3.2 性能差异
-
-| 维度 | Java SDK | Rust |
-|------|----------|------|
-| **写入吞吐** | 良好 | 优秀（零拷贝 Arrow，异步 I/O） |
-| **内存效率** | 依赖 GC，可能 FGC | 确定性的所有权管理 |
-| **启动时间** | JVM 预热开销 | 原生性能，无预热 |
-| **CPU 效率** | JIT 编译优化 | 编译期优化 |
-| **跨语言调用** | JNI 开销 | FFI 开销（更低） |
-
-### 3.3 功能完备度
-
-**Java SDK 功能完备度**：⭐⭐⭐⭐⭐（参考实现，100% 功能）
-**Rust 功能完备度**：⭐⭐⭐⭐（约 75%，核心缺失：Compaction, Lookup/FullCompaction
ChangelogProducer）
-
----
-
-## 4. Doris 现有写入框架分析
-
-### 4.1 BE 端写入框架
-
-Doris BE 端的 Table Writer 继承体系：
-
-```
-ResultWriter (result_writer.h)
- └── AsyncResultWriter (async_result_writer.h)
- ├── VTabletWriter (vtablet_writer.h) // 内部表写入
- ├── VHiveTableWriter (vhive_table_writer.h) // Hive 外部表
- ├── VJdbcTableWriter (vjdbc_table_writer.h) // JDBC 外部表
- ├── VTVFTableWriter (vtvf_table_writer.h) // TVF 表函数
- ├── VIcebergTableWriter // Iceberg 外部表
- └── VMaxComputeTableWriter // MaxCompute 外部表
-```
-
-**AsyncResultWriter** 的核心接口：
-```cpp
-class AsyncResultWriter : public ResultWriter {
-public:
- virtual Status open(RuntimeState* state, RuntimeProfile* profile) = 0;
- virtual Status write(RuntimeState* state, Block& input_block) = 0; // 实际
I/O
- Status sink(Block* block, bool eos); // 异步入队
- Status start_writer(RuntimeState* state, RuntimeProfile* profile); // 启动
I/O 线程
-};
-```
-
-**关键设计**：
-- `sink()` → 异步队列 → `write()`：解耦计算和 I/O
-- `Block` 是 Doris 的列式数据批次
-- 每个 writer 实例对应一个 BE worker 线程
-
-### 4.2 Iceberg Table Writer 参考
-
-`VIcebergTableWriter` 是最接近 Paimon 写入需求的外部表写入器：
-
-```
-写入流程：
- write(RuntimeState, Block)
- → 分区计算（PartitionTransform）
- → 确定目标 partition path
- → 获取/创建 VIcebergPartitionWriter（每个 partition 一个）
- → 写入 Parquet/ORC 文件
- → 文件滚动（按 target_file_size）
- close()
- → 关闭所有 partition writers
- → 收集产生的 DataFile 列表
- → 通过 Iceberg Catalog API 提交（AddFiles）
-```
-
-**关键差异（vs Paimon）**：
-
-Iceberg 和 Paimon 的写入模型存在根本性差异，这直接影响 Doris 的写入架构设计。
-
-**Iceberg：Commit-on-Close 模型**
-
-```
-BE Worker 1 BE Worker 2 FE Coordinator
- │ │ │
- │ write(block) × N │ write(block) × N │
- │ → 写入 Parquet 文件 │ → 写入 Parquet 文件 │
- │ → 文件滚动 │ → 文件滚动 │
- │ │ │
- │ close() │ close() │
- │ → 收集 DataFile 列表 │ → 收集 DataFile 列表 │
- │ → 生成 TIcebergCommitData │ → 生成 TIcebergCommitData │
- │ {file_path, row_count, │ {file_path, row_count, │
- │ file_size, partition} │ file_size, partition} │
- │ │ │
- │──── TIcebergCommitData ──────│──── TIcebergCommitData ──────→│
- │ │ │
- │ │ 聚合所有 DataFile 列表 │
- │ │ table.newAppend() │
- │ │ .appendFile(files...) │
- │ │ .commit() ← 原子提交 │
-```
-
-Iceberg writer 的输出就是**最终产物**：一堆已经写好的数据文件（Parquet/ORC）+ 文件的元信息（路径、行数、大小、分区值）。FE
Coordinator 收集这些文件列表后，直接调用 Iceberg 的 `appendFile()` + `commit()`
即可完成提交。文件列表是**透明的、可直接使用的**。
-
-**Paimon：Prepare-then-Commit 模型**
-
-```
-BE Worker 1 BE Worker 2 FE Coordinator
- │ │ │
- │ write(row) × N │ write(row) × N │
- │ → Paimon 内部： │ → Paimon 内部： │
- │ 分区/桶路由 │ 分区/桶路由 │
- │ KeyValueFileWriter │ AppendOnlyWriter │
- │ 排序、去重、序列号分配 │ 文件滚动 │
- │ 写入数据文件 + changelog │ │
- │ │ │
- │ prepareCommit() │ prepareCommit() │
- │ → CommitMessage[] │ → CommitMessage[] │
- │ { │ { │
- │ partition: bytes, │ partition: bytes, │
- │ bucket: 3, │ bucket: 0, │
- │ newFiles: [DataFileMeta │ newFiles: [DataFileMeta │
- │ {fileName, level, │ {fileName, level, │
- │ minSeq, maxSeq, │ minSeq, maxSeq, │
- │ minKey, maxKey, │ minKey, maxKey, │
- │ rowCount, ...}], │ rowCount, ...}], │
- │ changelogFiles: [...], │ changelogFiles: [...], │
- │ indexFiles: [...] │ indexFiles: [...] │
- │ } │ } │
- │ │ │
- │── CommitMessage[] (序列化) ──│── CommitMessage[] (序列化) ──→│
- │ │ │
- │ │ 聚合所有 CommitMessage │
- │ │ TableCommit.commit( │
- │ │ allMessages) │
- │ │ ├─ 合并 Manifest │
- │ │ ├─ 冲突检测 │
- │ │ ├─ 写入新 Manifest │
- │ │ └─ 原子创建 Snapshot │
-```
-
-Paimon writer 的输出是**中间产物**：`CommitMessage` 不是简单的"文件路径列表"，而是一个包含内部元数据的结构化消息：
-
-- **`newFiles` vs `deletedFiles`**：同一个 CommitMessage 可能同时包含新增文件和要删除的文件（例如
Compaction 后：新文件是合并结果，旧文件标记删除）
-- **`minSequenceNumber` / `maxSequenceNumber`**：每个文件的序列号范围，Commit 引擎用它来检测冲突和保证
LSM-tree 的一致性
-- **`changelogFiles`**：如果启用了 changelog producer，会产生单独的 changelog 文件，需要在 Commit
时写入独立的 changelog manifest
-- **`indexFiles` / `deletedIndexFiles`**：动态桶模式下的哈希索引文件，需要合并到 IndexManifest
-- **`minKey` / `maxKey`**：用于 Manifest 的分区裁剪优化
-
-这些元数据**必须由 Paimon 的 Commit
引擎来消费和处理**，外部不能简单地"收集文件列表然后提交"。`TableCommit.commit()` 做的事情远比 Iceberg 的
`appendFile()` 复杂——它需要合并 manifest、检测冲突、分配 row ID、维护索引、写 changelog manifest 等。
-
-**对 Doris 架构的影响**：
-
-Doris 侧采用与 Iceberg 相同的单 Writer 架构——每次 `write()` 将完整 Block 交给 Paimon SDK，SDK
内部完成分区路由和文件写入。与 Iceberg 的关键区别在于 Commit：
-
-| 维度 | Iceberg 模式 | Paimon 模式 |
-|------|-------------|-------------|
-| Doris Writer 架构 | 单一 `AsyncResultWriter` + 内部多 partition writer | 单一
`AsyncResultWriter` + 单一 `IPaimonWriter` |
-| Writer 输出 | DataFile 列表（透明） | CommitMessage（不透明内部结构） |
-| Commit 执行者 | FE 直接调 Iceberg API | 需要通过 Paimon `TableCommit` 执行 |
-| Commit 的关键输入 | 文件路径 + 行数 + 分区 | 完整 CommitMessage（含序列号、键范围、索引文件等） |
-| 提交失败处理 | 重试即可（文件已存在） | 需 abort() 清理数据文件，然后重试 |
-| 跨 BE 聚合 | 简单合并文件列表 | 相同 (partition,bucket) 的 messages 需保持原样传递 |
-| FE 需要理解的内容 | 文件元信息 | 无需理解（透传），但不能丢失任何字段 |
-| 分区/桶路由 | Writer 内部计算 Iceberg partition value | Paimon SDK 内部计算，Doris 不感知 |
-
-**为什么 Paimon 不能像 Iceberg 一样 Commit-on-Close？**
-
-根本原因是 Paimon 的 LSM-tree 架构。Paimon 的表不是简单的"文件集合"，而是一个分层的有序结构：
-
-1. **序列号（Sequence Number）**：LSM-tree 依赖全局递增的序列号来决定记录的可见性和冲突解决。每个 writer
写入时分配序列号，但 Commit 时 Commit 引擎需要全局地验证序列号的连续性，合并增量到 base manifest。
-2. **Compaction**：Writer 在 `prepareCommit()` 时可能已经执行了
Compaction，产生了新旧文件的替换关系（`compactBefore` / `compactAfter`），这些必须在 Commit 时原子地反映到
manifest 中。如果多个 writer 各自独立 commit，会导致 manifest 冲突。
-3. **Manifest 的版本一致性**：Paimon 的 Snapshot 是一个完整的 manifest 树（base manifest list
+ delta manifest list + changelog manifest list + index
manifest）。这棵树必须在**一个原子操作**中创建，不能由多个 writer 各自添加叶子节点。
-
-因此，Paimon 的 Commit **必须是一个集中的、原子的操作**——这就是 "Prepare-then-Commit" 的含义：Writer
负责准备（产生 CommitMessage），Coordinator 负责提交（消费 CommitMessage 创建 Snapshot）。
-
-### 4.3 FE 端规划
-
-```
-BaseExternalTableDataSink
- └── IcebergTableSink // 生成 TIcebergTableSink (Thrift)
- // 包含：table location, schema, partition spec, file format, output
expressions
-
-DataSink (Thrift)
- ├── TIcebergTableSink
- ├── THiveTableSink
- ├── TMaxComputeTableSink
- └── TPaimonTableSink ← 需要新增
-```
-
-### 4.4 现有 JNI 框架
-
-Doris 已有 JNI 调用框架用于**读取**外部表：
-
-```
-be/src/format_v2/jni/
- ├── jni_table_reader.{h,cpp} // JNI 读取器框架
- ├── paimon_jni_reader.{h,cpp} // Paimon JNI 读取器
- ├── hudi_jni_reader.{h,cpp} // Hudi JNI 读取器
- └── ...
-
-be/src/format/jni/
- ├── jni_data_bridge.{h,cpp} // C++ ↔ Java 数据桥接
- └── jni_reader.{h,cpp}
-```
-
-**JNI 模式的写路径参考**：目前没有 JNI 写路径（只有读取），需要新建。
-
----
-
-## 5. 架构设计
-
-### 5.1 总体架构
-
-核心设计：**一个 `VPaimonTableWriter` = 一个 `AsyncResultWriter` 实例 = 一个 IO 线程**（对标
Iceberg 的 `VIcebergTableWriter`）。分区路由和桶路由由 **Paimon SDK 内部完成**，Doris
侧不参与路由计算，直接将完整 Block 交给 Paimon SDK 写入。
-
-```
-┌─────────────────────────────────────────────────────────────────────┐
-│ Doris FE │
-│ ┌──────────────┐ ┌──────────────────┐ ┌─────────────────────┐ │
-│ │ Nereids │ │ PaimonTableSink │ │ Thrift: │ │
-│ │ Planner │──▶│ (FE Planner) │──▶│ TPaimonTableSink │ │
-│ └──────────────┘ └──────────────────┘ └─────────┬───────────┘ │
-│ │ │
-│ ★ 无特殊 Shuffle：正常 Exchange 分发数据到各 BE │ │
-│ 分区/桶路由由 Paimon SDK 在写入时内部处理 │ │
-└───────────────────────────────────────────────────────┼─────────────┘
- │ TDataSink
- ▼
-┌───────────────────────────────────────────────────────┴─────────────┐
-│ Doris BE │
-│ │
-│ ┌──────────────────────────────────────────────────────────────┐ │
-│ │ Pipeline Operator: PaimonTableSinkOperatorX │ │
-│ │ │ │
-│ │ sink_impl(block, eos): │ │
-│ │ 直接调用 AsyncWriterSink::sink(block, eos) │ │
-│ │ ★ 无分区/桶路由逻辑，不做 Block 拆分 │ │
-│ └──────────────────────┬───────────────────────────────────────┘ │
-│ │ │
-│ ★ 仅一个 AsyncWriterSink<VPaimonTableWriter> │
-│ │ │
-│ ▼ │
-│ ┌────────────────────────────────────────────────────────────┐ │
-│ │ VPaimonTableWriter (AsyncResultWriter, 单实例) │ │
-│ │ │ │
-│ │ sink(block) → 入队列 (_data_queue, QUEUE_SIZE=3) │ │
-│ │ │ │
-│ │ process_block() ← 单一 IO 线程 │ │
-│ │ write(block): │ │
-│ │ Block → Arrow IPC → IPaimonWriter::write() │ │
-│ │ → Paimon SDK 内部完成: │ │
-│ │ · 分区计算 (BinaryRow partition) │ │
-│ │ · 桶计算 (hash % numBuckets) │ │
-│ │ · 按 (partition, bucket) 路由 │ │
-│ │ · 文件写入 (AppendOnlyWriter / KeyValueFileWriter) │ │
-│ │ · Compaction 自动触发 │ │
-│ │ │ │
-│ │ close(): │ │
-│ │ prepareCommit() → CommitMessage[] │ │
-│ └────────────────────────────────────────────────────────────┘ │
-│ │
-│ ┌──────────────────────────────────────────────────────────────┐ │
-│ │ IPaimonWriteBackend (抽象层) │ │
-│ │ │ │
-│ │ + createWriter() → IPaimonWriter (整个表共享一个 writer) │ │
-│ │ + createCommitter() → IPaimonCommitter │ │
-│ │ │ │
-│ │ ┌──────────────────────┐ ┌──────────────────────┐ │ │
-│ │ │ JniPaimonWriteBackend│ │ FfiPaimonWriteBackend │ │ │
-│ │ │ (Java SDK via JNI) │ │ (Rust SDK via C FFI) │ │ │
-│ │ └──────────────────────┘ └──────────────────────┘ │ │
-│ └──────────────────────────────────────────────────────────────┘ │
-│ │
-│ ┌──────────────────────────────────────────────────────────────┐ │
-│ │ CommitMessage (统一格式，跨后端) │ │
-│ │ 单 writer produce → RuntimeState 汇总 → RPC to FE 统一提交 │ │
-│ └──────────────────────────────────────────────────────────────┘ │
-└─────────────────────────────────────────────────────────────────────┘
-```
-
-### 5.2 核心设计决策
-
-#### 决策 1：统一 CommitMessage 格式
-
-**设计**：在 BE 层定义一个 C++ 版本的 `CommitMessage` 结构体，同时能被 JNI 后端（Java 序列化反序列化）和 FFI
后端（Rust 结构体）填充。
-
-```cpp
-// be/src/exec/sink/writer/paimon/paimon_commit_message.h
-struct PaimonDataFileMeta {
- std::string file_name;
- int64_t file_size;
- int64_t row_count;
- int64_t delete_row_count;
- int64_t min_sequence_number;
- int64_t max_sequence_number;
- int32_t level;
- int32_t schema_id;
- std::string min_key; // serialized BinaryRow
- std::string max_key;
- // ... 更多元数据字段
-};
-
-struct PaimonCommitMessage {
- std::string partition; // serialized BinaryRow bytes
- int32_t bucket;
- std::optional<int32_t> total_buckets;
-
- std::vector<PaimonDataFileMeta> new_files;
- std::vector<PaimonDataFileMeta> new_changelog_files;
- std::vector<PaimonIndexFileMeta> new_index_files;
- std::vector<PaimonIndexFileMeta> deleted_index_files;
- std::vector<PaimonDataFileMeta> deleted_files;
- std::optional<int64_t> check_from_snapshot;
-
- // 序列化为 Thrift 格式，传递给 FE Coordinator
- void to_thrift(TPaimonCommitMessage* t_msg) const;
-};
-```
-
-**理由**：
-- 解耦写入后端和提交逻辑
-- FE Coordinator 需要聚合所有 BE 的 CommitMessage，因此需要一个统一的序列化格式
-- 使用 Thrift（Doris 的标准 RPC 序列化）来传输 CommitMessage 到 FE
-
-#### 决策 2：IPaimonWriteBackend 抽象
-
-**设计**：定义一个 C++ 抽象接口，屏蔽 JNI 和 FFI 的差异。
-
-```cpp
-// be/src/exec/sink/writer/paimon/paimon_write_backend.h
-class IPaimonWriter {
-public:
- virtual ~IPaimonWriter() = default;
- // 写入一个 Block（列式批次）
- virtual Status write(RuntimeState* state, Block& block) = 0;
- // 关闭当前 writer，产生 CommitMessage
- virtual Status prepareCommit(std::vector<PaimonCommitMessage>& messages) =
0;
- // 强制 compaction（可选，仅 Java 后端支持）
- virtual Status compact(bool fullCompaction) = 0;
- // 丢弃当前写入
- virtual Status abort() = 0;
-};
-
-class IPaimonCommitter {
-public:
- virtual ~IPaimonCommitter() = default;
- // 提交所有 messages
- virtual Status commit(const std::vector<PaimonCommitMessage>& messages) =
0;
- // 分区覆盖提交
- virtual Status overwrite(const std::vector<PaimonCommitMessage>& messages,
- const std::optional<std::map<std::string,
std::string>>& staticPartition) = 0;
- // 清空表
- virtual Status truncateTable() = 0;
- // 清空分区
- virtual Status truncatePartitions(const std::vector<std::map<std::string,
std::string>>& partitions) = 0;
- // 中止（删除数据文件）
- virtual Status abort(const std::vector<PaimonCommitMessage>& messages) = 0;
-};
-
-class IPaimonWriteBackend {
-public:
- virtual ~IPaimonWriteBackend() = default;
-
- // 初始化
- virtual Status open(const TPaimonTableSink& sink) = 0;
-
- // 为特定 partition+bucket 创建 writer
- virtual Status createWriter(const std::string& partition,
- int32_t bucket,
- std::unique_ptr<IPaimonWriter>* writer) = 0;
-
- // 获取 committer（整个 sink 共享一个 committer）
- virtual Status createCommitter(std::unique_ptr<IPaimonCommitter>*
committer) = 0;
-
- // 后端类型标识
- virtual PaimonBackendType type() const = 0;
-
- // 功能查询
- virtual bool supportsCompaction() const = 0;
- virtual bool supportsLookupChangelog() const = 0;
- virtual bool supportsFullCompactionChangelog() const = 0;
-};
-```
-
-#### 决策 3：双后端，可互相替代
-
-Java writer 和 Rust writer 是**互相替代**的关系，不是互补关系。Rust 是 Java 的功能子集：
-
-```
-功能全集 (Java SDK)
-┌─────────────────────────────────────────┐
-│ ┌─────────────────────────────────┐ │
-│ │ 功能子集 (Rust) │ │
-│ │ · Append-only 写入 │ │
-│ │ · 固定桶 PK 写入 │ │
-│ │ · 动态桶 PK 写入 │ │
-│ │ · MergeEngine: Normal/ │ │
-│ │ PartialUpdate/Aggregation/ │ │
-│ │ FirstRow/Deduplicate │ │
-│ │ · ChangelogProducer: None/Input│ │
-│ │ · Overwrite/Truncate │ │
-│ └─────────────────────────────────┘ │
-│ Rust 缺失（需回退 Java）： │
-│ · Compaction │
-│ · Lookup/FullCompaction Changelog │
-│ · DeletionVector + PartialUpdate/ │
-│ Aggregation 组合 │
-│ · MERGE INTO / DataEvolution │
-└─────────────────────────────────────────┘
-```
-
-**选择策略：优先 Rust，功能不足时回退 Java**
-
-```
-IF table options 包含 Rust 不支持的功能:
- → Java JNI Backend ← 功能完备保证
-ELSE:
- → Rust FFI Backend ← 优先选择，性能更优
-```
-
-**v1 实现策略**：
-
-```
-v1: 仅实现 Java JNI Backend，但：
- - IPaimonWriteBackend 接口从第一天就存在
- - VPaimonTableWriter 通过 IPaimonWriteBackend 调用
- - Java 后端是实现细节，不是架构主体
- - 新增 Rust 后端 = 新增一个 IPaimonWriteBackend 实现类，不碰上层代码
-
-v2: 实现 Rust FFI Backend
- - 扩展 paimon-rust C binding
- - 新增 FfiPaimonWriteBackend（<1000行新代码）
- - 上层 VPaimonTableWriter 无需修改
-```
-
-| 场景 | v1 (仅 Java) | v2 (Java + Rust) |
-|------|-------------|-------------------|
-| Append-only 表 | JNI Backend | **FFI Backend** (优先) |
-| PK 表，固定桶，No Compaction | JNI Backend | **FFI Backend** (优先) |
-| PK 表，需要 Compaction | JNI Backend | JNI Backend (Rust 不支持) |
-| Lookup Changelog Producer | JNI Backend | JNI Backend (Rust 不支持) |
-| MERGE INTO | JNI Backend | JNI Backend (Rust 不支持) |
-
-#### 决策 4：Single Writer — 对标 Iceberg 的单 Writer 架构
-
-**问题**：v0 (初始实现) 采用了 "一个 `(partition, bucket)` = 一个 `AsyncResultWriter`"
的设计。每个 `AsyncResultWriter` 在 `fragment_mgr` 线程池中创建一个独立的 IO 线程。当 Paimon 表的
bucket 数量较大（例如 128 个 bucket × 10 个分区 = 1280 个 writer 实例），会导致线程数爆炸，严重浪费系统资源。
-
-**根因分析**：`AsyncResultWriter::start_writer()` 调用
`fragment_mgr()->get_thread_pool()->submit_func()` 为每个实例提交一个 long-running 的
`process_block()` 任务。这个任务在 writer 的整个生命周期中持续占用一个线程。当 writer 数量远超 CPU
核数时，大量线程空转等待数据，造成无效的上下文切换和内存开销。
-
-**解决方案**：改为 **一个 `VPaimonTableWriter` = 一个 `AsyncResultWriter` 实例 = 一个 IO
线程**，对标 Iceberg 的 `VIcebergTableWriter`。分区路由和桶路由完全交给 Paimon SDK 内部处理。
-
-Doris 侧不再：
-- 在 BE 端计算 Paimon partition bytes
-- 在 BE 端计算 bucket id
-- 在 Pipeline Operator 层做 Block 拆分和路由
-- 为每个 (partition, bucket) 创建独立的 writer
-
-Doris 侧只负责：
-- 将完整 Block 通过 `IPaimonWriter::write()` 交给 Paimon SDK
-- Paimon SDK 内部完成分区/桶路由、文件写入、Compaction
-
-```
-Doris Side Paimon SDK Side
-═══════════ ════════════════
-
-Block (完整批次) TableWrite.write(row × N)
- │ │
- ├── Arrow IPC 转换 ├── getPartition(row)
- │ ↓ ├── getBucket(row)
- ├── JNI/FFI 传递 ├── writeBundle(p, b, records)
- │ ↓ │ → AppendOnlyWriter
- └── IPaimonWriter::write() │ → KeyValueFileWriter
- │ → CompactManager
- │
- └── prepareCommit()
- → CommitMessage[]
-```
-
-**参考**：Iceberg 的 `VIcebergTableWriter` 采用相同架构（单一 `AsyncResultWriter` + 内部管理多个
`VIcebergPartitionWriter`），已在生产环境验证。
-
-**架构优势**：
-1. **线程数可控**：每个 BE worker 仅 1 个 IO 线程，不受 bucket/partition 数量影响
-2. **简化 Doris 侧代码**：移除 `paimon_bucket.h/cpp`、移除 operator 层的路由/拆分逻辑
-3. **复用 Paimon SDK 成熟逻辑**：分区/桶计算与 Paimon 原生语义完全一致，避免兼容性问题
-4. **减少内存开销**：不再需要为每个 (p,b) 维护独立的 AsyncResultWriter 队列和 block pool
-5. **简化 FE 端**：无需特殊的 shuffle key，正常 Exchange 即可
-
-**线程数对比**：
-
-| 场景 | v0 (per-bucket writer) | v1 (single writer) |
-|------|----------------------|---------------------|
-| 1 分区 × 128 bucket | 128 个 IO 线程 | 1 个 IO 线程 |
-| 10 分区 × 64 bucket | 640 个 IO 线程 | 1 个 IO 线程 |
-| Unaware bucket | 1 个 IO 线程 | 1 个 IO 线程 |
-
-**潜在风险**：
-- 单线程写入可能成为吞吐瓶颈 → Paimon SDK 内部使用异步 I/O (Rust) 或线程池 (Java)，单线程调用非瓶颈
-- 如需更高并发，可在 FE 层增加 `parallelism` 参数，创建多个 BE fragment 实例（每个持有独立 writer），这是标准的
Doris 水平扩展方式
-
-#### 决策 5：Bucket Affinity — 不再由 Doris 保证
-
-**背景变化**：由于分区/桶路由已移交 Paimon SDK 内部处理，Doris 不再需要感知 bucket affinity。
-
-多个 BE worker 可能会写入同一个 (partition, bucket)，每个 worker 的 Paimon TableWrite
实例独立产生文件。这与 Paimon Flink connector 的默认行为一致——多个 Subtask 各自写各自的文件，最终由 Commit 引擎合并
Manifest。
-
-**文件数量控制**：由 Paimon SDK 内部处理：
-- `target-file-size` 控制单个文件大小
-- Compaction 自动合并小文件
-- 对于 Append-only 表，可通过 `write-only` 模式 + `compaction` 任务定期合并
-
-**长期优化（可选）**：如需减少小文件，可后续在 FE Planner 层引入 shuffle by
hash(bucket_key)，但不影响当前架构设计。
-
-### 5.3 提交流程
-
-Paimon 的两阶段提交对 Doris 的分布式架构提出了特殊挑战：
-
-```
-时间线：分布式写入 + 集中式提交
-
-BE1: write() → prepareCommit() → CommitMessage[]
-BE2: write() → prepareCommit() → CommitMessage[]
-BE3: write() → prepareCommit() → CommitMessage[]
- │ │ │
- └───────────────┼────────────────────┘
- │ (通过 Thrift RPC 发送到 FE Coordinator)
- ▼
-FE Coordinator: 聚合所有 CommitMessage[]
- │
- ▼
- TableCommit.commit(allMessages)
- │
- ▼
- 新的 Snapshot 创建
-```
-
-**关键问题**：Commit 应该在哪发生？
-
-**选择 1：FE Coordinator 执行 Commit**
-- 优点：自然的两阶段提交，FE 已有 Iceberg 的集中式提交模式
-- 缺点：FE 需要持有 Java SDK 依赖（或通过 BE 代理）
-
-**选择 2：选出一个 BE 执行 Commit（Leader BE）**
-- 优点：BE 端直接调用 JNI/FFI
-- 缺点：需要选举逻辑，增加复杂度
-
-**选择 3：混合方案** ✅ **推荐**
-
-```
-BE 端：
- - 数据写入（全并行）
- - prepareCommit() → CommitMessage[]（序列化为 Thrift）
-
-FE Coordinator 端：
- - 聚合所有 BE 的 CommitMessage[]
- - 通过专用的 "Commit BE" 执行 TableCommit.commit()
-
-Commit BE（可以是任意 BE 或专用 BE）：
- - 接收 FE 发来的聚合后的 CommitMessage[]
- - 调用 JNI Backend 的 TableCommit.commit()
- - 返回结果给 FE
-```
-
-对于 Append-only 表（简单场景），可以直接在 FE 端通过 Java SDK 执行提交。对于需要 Compaction 的表（复杂场景），通过
Commit BE 执行。
-
-### 5.4 架构图总览（分场景）
-
-#### 场景 1：Append-only 表写入
-
-```
-Doris Sink
- │
- ▼
-BE: PaimonTableSinkOperatorX (Pipeline Operator)
- │ sink_impl(block, eos) → 直接调用 AsyncWriterSink::sink()
- │ ★ 无分区/桶路由，Block 原样传递
- │
- ▼
-VPaimonTableWriter (单一 AsyncResultWriter)
- │ process_block() → write(block):
- │ Block → Arrow IPC → IPaimonWriter::write()
- │
- ▼
-IPaimonWriter (Rust FFI preferred)
- │ Paimon SDK 内部:
- │ · 分区/桶计算
- │ · write_arrow_batch() → AppendOnlyWriter
- │ · 文件写入
- │
- ▼
-prepareCommit() → CommitMessage[]
- │
- ▼ (Thrift RPC)
-FE Coordinator → TableCommit.commit()
- ▼
-New Snapshot ✓
-```
-
-#### 场景 2：主键表写入
-
-```
-Doris Sink
- │
- ▼
-BE: PaimonTableSinkOperatorX (Pipeline Operator)
- │ sink_impl(block, eos) → 直接调用 AsyncWriterSink::sink()
- │
- ▼
-VPaimonTableWriter (单一 AsyncResultWriter)
- │ process_block() → write(block):
- │ Block → Arrow IPC → JNI → Java
- │
- ▼
-JniPaimonWriter (Java SDK)
- │ Paimon BatchTableWrite:
- │ · 分区/桶计算 + writeBundle(p,b,records)
- │ · KeyValueFileWriter (排序、去重、序列号分配)
- │ · CompactManager 自动 compaction
- │ · 写入数据文件 + changelog
- │
- ▼
-prepareCommit() → CommitMessage[]
- │
- ▼ (Thrift RPC)
-FE Coordinator → TableCommit.commit()
- ▼
-New Snapshot ✓
-```
-
----
-
-## 6. 接口定义
-
-### 6.1 Thrift 定义
-
-```thrift
-// gensrc/thrift/DataSinks.thrift
-
-enum TPaimonWriteBackendType {
- JNI = 0, // Java SDK via JNI
- FFI = 1, // Rust via C FFI
-}
-
-enum TPaimonWriteMode {
- APPEND = 0, // INSERT INTO
- OVERWRITE = 1, // INSERT OVERWRITE
- DYNAMIC_OVERWRITE = 2, // 动态分区覆盖
- MERGE_INTO = 3, // DataEvolution MERGE INTO
-}
-
-struct TPaimonTableSink {
- // 表标识
- 1: required string table_location; // Paimon 表路径
- 2: required string paimon_catalog_type; // filesystem / hive / rest
- 3: optional map<string, string> catalog_options;
- 4: optional map<string, string> table_options;
-
- // Schema
- 5: required string schema_json; // Paimon TableSchema JSON
- 6: required list<string> partition_keys;
- 7: required list<string> primary_keys;
- 8: required list<i32> primary_key_indices;
-
- // 写入配置
- 9: required TPaimonWriteMode write_mode;
- 10: optional map<string, string> static_partition; // overwrite 分区 spec
- 11: required TPaimonWriteBackendType backend_type;
-
- // 输出表达式（Doris Block → Paimon Row 映射）
- 12: required list<TExpr> output_exprs;
-
- // 文件配置
- 13: optional string file_format; // parquet / orc / avro
- 14: optional string file_compression; // zstd / snappy / lz4 / none
- 15: optional i64 target_file_size; // bytes
-
- // 提交配置
- 16: optional string commit_user;
- 17: optional i32 commit_max_retries;
- 18: optional i64 commit_timeout_ms;
-
- // Hadoop 配置（对于 Hive Metastore catalog）
- 19: optional map<string, string> hadoop_config;
-}
-
-struct TPaimonCommitMessage {
- 1: required binary partition; // serialized BinaryRow
- 2: required i32 bucket;
- 3: optional i32 total_buckets;
- 4: required list<TPaimonDataFileMeta> new_files;
- 5: optional list<TPaimonDataFileMeta> new_changelog_files;
- 6: optional list<TPaimonIndexFileMeta> new_index_files;
- 7: optional list<TPaimonIndexFileMeta> deleted_index_files;
- 8: optional list<TPaimonDataFileMeta> deleted_files;
- 9: optional i64 check_from_snapshot;
-}
-
-struct TPaimonDataFileMeta {
- 1: required string file_name;
- 2: required i64 file_size;
- 3: required i64 row_count;
- 4: optional i64 delete_row_count;
- 5: required i64 min_sequence_number;
- 6: required i64 max_sequence_number;
- 7: required i32 level;
- 8: required i32 schema_id;
- 9: required binary min_key;
- 10: required binary max_key;
- 11: optional binary min_partition_key;
- 12: optional binary max_partition_key;
- 13: optional list<string> write_cols;
- 14: optional i64 first_row_id;
- 15: optional i32 file_source;
- // ... 其他元数据
-}
-
-struct TPaimonIndexFileMeta {
- 1: required string file_name;
- 2: required i64 file_size;
- 3: required i64 row_count;
- 4: required string index_type; // "HASH" / "BTREE" / "GLOBAL"
- 5: optional TPaimonGlobalIndexMeta global_index_meta;
-}
-
-struct TPaimonCommitResult {
- 1: required bool success;
- 2: optional i64 new_snapshot_id;
- 3: optional string error_message;
-}
-
-// 添加到 TDataSink
-struct TDataSink {
- // ... existing fields ...
- 16: optional TPaimonTableSink paimon_table_sink
-}
-```
-
-### 6.2 C++ 抽象接口（完整版）
-
-```cpp
-// be/src/exec/sink/writer/paimon/paimon_write_backend.h
-
-namespace doris {
-
-// ──── 枚举 ────────────────────────────────────────────────
-
-enum class PaimonBackendType {
- JNI, // Java SDK
- FFI // Rust C FFI
-};
-
-enum class PaimonWriteMode {
- APPEND,
- OVERWRITE,
- DYNAMIC_OVERWRITE,
- MERGE_INTO
-};
-
-// ──── 数据结构 ────────────────────────────────────────────
-
-struct PaimonWriteOptions {
- std::string table_location;
- std::string commit_user;
- PaimonWriteMode write_mode;
- std::optional<std::unordered_map<std::string, std::string>>
static_partition;
- std::string file_format; // "parquet", "orc", "avro"
- std::string file_compression; // "zstd", "snappy", "lz4", "none"
- int64_t target_file_size_bytes = 128 * 1024 * 1024; // 128MB default
- int32_t commit_max_retries = 5;
- int64_t commit_timeout_ms = 300000; // 5min
-};
-
-// ──── IPaimonWriter ───────────────────────────────────────
-
-class IPaimonWriter {
-public:
- virtual ~IPaimonWriter() = default;
-
- /// 写入一个 Doris Block（完整批次，可能包含多分区、多桶的数据）
- /// Paimon SDK 内部完成：分区计算、桶计算、按 (partition,bucket) 路由、
- /// 文件写入（AppendOnlyWriter / KeyValueFileWriter）、Compaction 触发。
- ///
- /// @param state 运行时状态
- /// @param block 输入数据（Doris 列式格式，未按分区/桶拆分）
- virtual Status write(RuntimeState* state, Block& block) = 0;
-
- /// 准备提交：关闭所有内部 writer，生成 CommitMessage
- /// 对所有 (partition, bucket) 内部 writer 调用 prepareCommit()
- /// @param messages 输出：本次写入产生的所有 CommitMessage 列表
- virtual Status prepareCommit(
- std::vector<PaimonCommitMessage>& messages) = 0;
-
- /// 强制 compaction（仅部分后端支持）
- /// @param fullCompaction true=全量压缩, false=增量压缩
- virtual Status compact(bool fullCompaction) {
- return Status::NotSupported("compact not supported by this backend");
- }
-
- /// 丢弃本次写入的所有数据
- virtual Status abort() = 0;
-};
-
-// ──── IPaimonCommitter ────────────────────────────────────
-
-class IPaimonCommitter {
-public:
- virtual ~IPaimonCommitter() = default;
-
- /// 追加提交：将所有 CommitMessage 中的数据文件原子提交
- virtual Status commit(
- const std::vector<PaimonCommitMessage>& messages) = 0;
-
- /// 覆盖提交：删除旧数据，写入新数据
- virtual Status overwrite(
- const std::vector<PaimonCommitMessage>& messages,
- const std::optional<std::unordered_map<std::string, std::string>>&
- static_partition) = 0;
-
- /// 清空整张表
- virtual Status truncateTable() = 0;
-
- /// 清空指定分区
- virtual Status truncatePartitions(
- const std::vector<std::unordered_map<std::string, std::string>>&
- partitions) = 0;
-
- /// 中止：删除已写入但未提交的数据文件
- virtual Status abort(
- const std::vector<PaimonCommitMessage>& messages) = 0;
-};
-
-// ──── IPaimonWriteBackend ─────────────────────────────────
-
-class IPaimonWriteBackend {
-public:
- virtual ~IPaimonWriteBackend() = default;
-
- /// 初始化后端
- virtual Status open(const TPaimonTableSink& sink,
- RuntimeState* state) = 0;
-
- /// 为整个表创建一个 writer（管理所有 partition + bucket）
- /// 分区/桶路由由 Paimon SDK 在 write() 内部处理
- /// @param writer 输出：创建的 writer
- virtual Status createWriter(
- std::unique_ptr<IPaimonWriter>* writer) = 0;
-
- /// 创建一个 committer（每个 sink 只需要一个）
- virtual Status createCommitter(
- std::unique_ptr<IPaimonCommitter>* committer) = 0;
-
- /// 获取后端类型
- virtual PaimonBackendType type() const = 0;
-
- // ──── 功能查询 ────────────────────────────────────────
-
- virtual bool supportsCompaction() const = 0;
- virtual bool supportsLookupChangelogProducer() const = 0;
- virtual bool supportsFullCompactionChangelogProducer() const = 0;
- virtual bool supportsPartialUpdateWithDV() const = 0;
- virtual bool supportsAggregationWithDV() const = 0;
-};
-
-// ──── 后端工厂 ────────────────────────────────────────────
-
-class PaimonWriteBackendFactory {
-public:
- /// 根据配置创建合适的后端
- static Status create(
- const TPaimonTableSink& sink,
- std::unique_ptr<IPaimonWriteBackend>* backend);
-
- /// 根据表属性和用户偏好选择最佳后端类型
- static PaimonBackendType selectBackendType(
- const TPaimonTableSink& sink);
-};
-
-} // namespace doris
-```
-
----
-
-## 7. 数据流与交互时序
-
-### 7.1 完整写入流程（时序图）
-
-```
- FE Coordinator BE Worker 1 BE Worker 2 Commit BE
- │ │ │ │
- │ TDataSink │ │ │
- │ (TPaimonTableSink) │ │ │
- ├──────────────────────┤ │ │
- │ │ │ │
- │ open() open() │
- │ IPaimonWriteBackend IPaimonWriteBackend │
- │ → createWriter() → createWriter() │
- │ → IPaimonWriter → IPaimonWriter │
- │ │ │ │
- │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ │
- │ │ 写入阶段（并行） │ │
- │ │ │ │
- │ │ Block1[rows] │ │
- │ ├─────────────────────────────────────────┤ │
- │ │ write() → IPaimonWriter │ │
- │ │ → Paimon SDK 内部: │ │
- │ │ · 分区计算 + 桶计算 │ │
- │ │ · writeBundle(p,b,records) │ │
- │ │ · Compaction 自动触发 │ │
- │ │ │ │
- │ │ Block2[rows] │ │
- │ ├─────────────────────────────────────────┤ │
- │ │ write() → Paimon SDK │ │
- │ │ │ │
- │ │ ... (更多数据) ... │ │
- │ │ │ │
- │ │ EOS (End of Stream) │ │
- │ ├─────────────────────────────────────────┤ │
- │ │ prepareCommit() │ │
- │ │ → CommitMessage[] │ │
- │ │ (所有 partition+bucket 的消息) │ │
- │ │ │ │
- │ │ prepareCommit() │ │
- │ ├─────────────────────────────────────────┤ │
- │ │ → CommitMessage[] │ │
- │ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ │
- │ │ │ │
- │ CommitMessage[] │ CommitMessage[] │ │
- │ (Thrift RPC) │ (Thrift RPC) │ │
- │◄─────────────────────┼─────────────────────┤ │
- │ │ │ │
- │ 聚合 & 去重 │ │ │
- │ allMessages[] │ │ │
- │ │ │ │
- │ │ commit(allMessages) │
- ├──────────────────────────────────────────────────────────────────┤
- │ │ │ TableCommit │
- │ │ │ .commit() │
- │ │ │ ├─ write manifests│
- │ │ │ ├─ write snapshot │
- │ │ │ └─ atomic rename │
- │ │ │ │
- │ CommitResult │ │ │
- │◄──────────────────────────────────────────────────────────────────┤
- │ │ │ │
- │ (on success) │ │ │
- │ close() writers │ │ │
- ├──────────────────────┤ │ │
- │ │ │ │
- │ (on failure) │ │ │
- │ abort(messages) │ │ │
- ├──────────────────────┼─────────────────────┤ │
-```
-
-### 7.2 错误处理流程
-
-```
-写入阶段可能的错误：
- 1. 网络/存储 I/O 错误
- → 重试（with exponential backoff）
- → 超过重试次数 → 标记 task 失败 → FE 重新调度
-
- 2. 内存不足
- → spill to disk (if spillable enabled)
- → 触发 memory limiter → 减少并发度
-
- 3. Schema mismatch
- → 立即失败，不回滚已写入的文件
- → 依赖 Snapshot 的隔离性，未提交的数据不会被读到
-
-提交阶段可能的错误：
- 1. Commit conflict（并发提交）
- → 自动重试（commit_max_retries times）
- → 检查 idempotency (commit_user + commit_identifier)
-
- 2. 提交超时
- → 检查是否已经提交成功（idempotency check）
- → 如果未成功 → 根据 Snapshot 状态决定重试还是放弃
-
- 3. 写入成功但提交失败 = 孤儿文件
- → 调用 abort(messages) 清理数据文件
- → 下次 Snapshot expiration 时也会清理未引用的文件
-```
-
----
-
-## 8. FE 端设计
-
-### 8.1 PaimonTableSink（FE Planner）
-
-```java
-// fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java
-
-public class PaimonTableSink extends BaseExternalTableDataSink {
-
- private final PaimonExternalTable targetTable;
- private List<Expr> outputExprs;
- private PaimonWriteMode writeMode;
- private Map<String, String> staticPartition;
-
- public PaimonTableSink(PaimonExternalTable targetTable,
InsertCommandContext ctx) {
- this.targetTable = targetTable;
- // 从 ctx 确定写入模式
- if (ctx.isOverwrite()) {
- this.writeMode = PaimonWriteMode.OVERWRITE;
- this.staticPartition = ctx.getStaticPartition();
- } else {
- this.writeMode = PaimonWriteMode.APPEND;
- }
- }
-
- @Override
- public TDataSink toThrift() {
- TDataSink result = new TDataSink(TDataSinkType.PAIMON_TABLE_SINK);
-
- TPaimonTableSink paimonSink = new TPaimonTableSink();
- paimonSink.setTableLocation(targetTable.getTableLocation());
- paimonSink.setPaimonCatalogType(targetTable.getCatalogType());
- paimonSink.setSchemaJson(targetTable.getPaimonSchemaJson());
- paimonSink.setPartitionKeys(targetTable.getPartitionKeys());
- paimonSink.setPrimaryKeys(targetTable.getPrimaryKeys());
- paimonSink.setPrimaryKeyIndices(targetTable.getPrimaryKeyIndices());
- paimonSink.setWriteMode(writeMode.toThrift());
- paimonSink.setOutputExprs(Expr.treesToThrift(outputExprs));
- paimonSink.setFileFormat(targetTable.getFileFormat());
- paimonSink.setFileCompression(targetTable.getFileCompression());
- paimonSink.setTargetFileSize(targetTable.getTargetFileSize());
-
- // 选择后端类型
- paimonSink.setBackendType(selectBackendType().toThrift());
-
- // 生成 commit_user
- paimonSink.setCommitUser(UUID.randomUUID().toString());
-
- result.setPaimonTableSink(paimonSink);
- return result;
- }
-
- /**
- * 智能选择写入后端：
- * - 如果需要 Compaction 或 Lookup/FullCompaction Changelog，必须用 JNI
- * - 如果用户指定了 preferred backend，遵循用户选择
- * - 默认：JNI（功能优先）
- */
- private TPaimonWriteBackendType selectBackendType() {
- // 检查表是否配置了需要 Java SDK 才支持的功能
- if (targetTable.requiresCompaction() ||
- targetTable.getChangelogProducer().requiresJava()) {
- return TPaimonWriteBackendType.JNI;
- }
-
- // 检查用户偏好
- String configuredBackend = targetTable.getTableProperties()
- .get("doris.paimon.write.backend");
- if ("ffi".equalsIgnoreCase(configuredBACKend) ||
- "rust".equalsIgnoreCase(configuredBACKend)) {
- return TPaimonWriteBackendType.FFI;
- }
-
- // 默认使用 JNI（功能最完整）
- return TPaimonWriteBackendType.JNI;
- }
-}
-```
-
-### 8.2 PaimonWriteCommitCoordinator（FE Coordinator）
-
-```java
-//
fe/fe-core/src/main/java/org/apache/doris/transaction/PaimonWriteCommitCoordinator.java
-
-public class PaimonWriteCommitCoordinator {
-
- /**
- * 协调分布式写入的提交：
- * 1. 收集所有 BE 的 CommitMessage[]
- * 2. 聚合去重
- * 3. 选择一个 BE 执行提交（或 FE 本地执行）
- * 4. 处理提交结果
- */
- public TPaimonCommitResult commit(
- Database db,
- PaimonExternalTable table,
- List<List<TPaimonCommitMessage>> allBackendMessages) {
-
- // Step 1: 扁平化所有 BE 的 messages
- List<TPaimonCommitMessage> allMessages = allBackendMessages.stream()
- .flatMap(List::stream)
- .collect(Collectors.toList());
-
- if (allMessages.isEmpty()) {
- return new TPaimonCommitResult(true); // nothing to commit
- }
-
- // Step 2: 聚合相同 (partition, bucket) 的 messages
- Map<Pair<byte[], Integer>, List<TPaimonCommitMessage>> grouped =
- groupByPartitionBucket(allMessages);
-
- // Step 3: 选择 Commit BE
- TNetworkAddress commitBE = selectCommitBE(allBackendMessages);
-
- // Step 4: 通过 RPC 调用 Commit BE 执行提交
- TPaimonCommitRequest request = new TPaimonCommitRequest();
- request.setTableLocation(table.getTableLocation());
- request.setCommitMessages(allMessages);
- request.setCommitUser(generateCommitUser());
-
- try {
- TPaimonCommitResult result = BackendServiceProxy.getInstance()
- .paimonCommit(commitBE, request);
- return result;
- } catch (Exception e) {
- // 提交失败处理
- return handleCommitFailure(allMessages, e);
- }
- }
-}
-```
-
----
-
-## 9. BE 端设计
-
-### 9.1 VPaimonTableWriter
-
-对标 Iceberg 的 `VIcebergTableWriter`：单一 `AsyncResultWriter` 实例，在 `write()` 中将完整
Block 交给 Paimon SDK 处理。分区/桶路由、文件写入、Compaction 全部由 Paimon SDK 内部完成。
-
-```cpp
-// be/src/exec/sink/writer/paimon/vpaimon_table_writer.h
-
-class VPaimonTableWriter final : public AsyncResultWriter {
-public:
- VPaimonTableWriter(const TDataSink& t_sink,
- const VExprContextSPtrs& output_exprs,
- std::shared_ptr<Dependency> dep,
- std::shared_ptr<Dependency> fin_dep);
-
- Status open(RuntimeState* state, RuntimeProfile* profile) override;
- Status write(RuntimeState* state, Block& block) override;
- Status close(Status status) override;
-
-private:
- TDataSink _t_sink;
- RuntimeState* _state = nullptr;
-
- // 抽象后端 — 整个 writer 持有一个 IPaimonWriteBackend 实例
- std::unique_ptr<IPaimonWriteBackend> _backend;
-
- // 单一 IPaimonWriter（管理所有 partition + bucket，对标 Iceberg 的
- // _partitions_to_writers map，但路由逻辑在 Paimon SDK 内部）
- std::unique_ptr<IPaimonWriter> _writer;
-
- // 统计
- int64_t _written_rows = 0;
- int64_t _written_bytes = 0;
-
- // Profile counters
- RuntimeProfile::Counter* _written_rows_counter = nullptr;
- RuntimeProfile::Counter* _written_bytes_counter = nullptr;
- RuntimeProfile::Counter* _send_data_timer = nullptr;
- RuntimeProfile::Counter* _project_timer = nullptr;
- RuntimeProfile::Counter* _arrow_convert_timer = nullptr;
- RuntimeProfile::Counter* _file_store_write_timer = nullptr;
- RuntimeProfile::Counter* _open_timer = nullptr;
- RuntimeProfile::Counter* _close_timer = nullptr;
- RuntimeProfile::Counter* _prepare_commit_timer = nullptr;
- RuntimeProfile::Counter* _serialize_commit_messages_timer = nullptr;
- RuntimeProfile::Counter* _commit_payload_count = nullptr;
- RuntimeProfile::Counter* _commit_payload_bytes_counter = nullptr;
-};
-```
-
-**`write()` 实现逻辑**（伪代码）：
-
-```cpp
-Status VPaimonTableWriter::write(RuntimeState* state, Block& block) {
- if (block.rows() == 0) return Status::OK();
-
- SCOPED_TIMER(_send_data_timer);
-
- // 1. Project: 应用 output expressions
- Block output_block;
- {
- SCOPED_TIMER(_project_timer);
- RETURN_IF_ERROR(_projection_block(block, &output_block));
- }
-
- // 2. 写入 Paimon SDK（分区/桶路由由 SDK 内部处理）
- {
- SCOPED_TIMER(_file_store_write_timer);
- RETURN_IF_ERROR(_writer->write(state, output_block));
- }
-
- COUNTER_UPDATE(_written_rows_counter, block.rows());
- _written_rows += block.rows();
- return Status::OK();
-}
-```
-
-**`open()` 实现逻辑**：
-
-```cpp
-Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
- _state = state;
- _operator_profile = profile;
-
- // 注册 counters
- // ...
-
- // 创建后端（JNI 或 FFI）
-
RETURN_IF_ERROR(PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink,
&_backend));
- RETURN_IF_ERROR(_backend->open(_t_sink.paimon_table_sink, state));
-
- // 创建一个 writer（管理全部分区和桶）
- RETURN_IF_ERROR(_backend->createWriter(&_writer));
-
- return Status::OK();
-}
-```
-
-**`close()` 实现逻辑**：
-
-```cpp
-Status VPaimonTableWriter::close(Status status) {
- SCOPED_TIMER(_close_timer);
-
- std::vector<TPaimonCommitMessage> messages;
- if (status.ok() && _writer) {
- {
- SCOPED_TIMER(_prepare_commit_timer);
- // prepareCommit 对所有内部 (partition, bucket) writer 调用
- Status prep_st = _writer->prepare_commit(messages);
- if (!prep_st.ok()) status = prep_st;
- }
- if (status.ok() && !messages.empty()) {
- _state->add_paimon_commit_messages(messages);
- }
- }
-
- if (!status.ok() && _writer) {
- _writer->abort(); // 清理所有未提交的数据文件
- }
-
- _writer.reset();
- _backend.reset();
- return status;
-}
-```
-
-**与 Iceberg VIcebergTableWriter 的对应关系**：
-
-| 维度 | VIcebergTableWriter | VPaimonTableWriter (新) |
-|------|-------------------|------------------------|
-| AsyncResultWriter 个数 | 1 | 1 |
-| IO 线程数 | 1 | 1 |
-| 分区路由位置 | `write()` 内部 `_write_prepared_block()` | `IPaimonWriter::write()`
内部（SDK） |
-| 多分区管理 | `_partitions_to_writers` map (C++ 侧) | Paimon SDK 内部 map |
-| 文件轮转 | 按 `target_file_size_bytes` | Paimon SDK 内部处理 |
-| close 行为 | 遍历关闭所有 partition writer | `prepareCommit()` 一次性收拢所有消息 |
-
-Pipeline Operator 层简化为最简单的透传，不再做路由：
-
-```cpp
-// PaimonTableSinkOperatorX 不再需要 _get_or_create_writer()
-// 不再需要 _compute_routing()
-// 不再需要 _split_by_key()
-// 不再需要 _writers map
-// 不再需要 _bucket_key_indices / _partition_col_indices / _total_buckets
-
-// **sink_impl() 简化为：**
-// 直接调用 AsyncWriterSink::sink() → VPaimonTableWriter::sink() → 入队列
-```
-
-### 9.1.1 移除的 BE 端代码
-
-以下文件和函数在本次架构调整中**不再需要**，将由后续 PR 清理：
-
-| 文件/函数 | 说明 |
-|----------|------|
-| `be/src/exec/sink/writer/paimon/paimon_bucket.h` | BE 端 MurmurHash 桶计算 |
-| `be/src/exec/sink/writer/paimon/paimon_bucket.cpp` | 同上 |
-| `PaimonTableSinkOperatorX::_compute_routing()` | Pipeline 层路由计算 |
-| `PaimonTableSinkOperatorX::_split_by_key()` | Block 按 (p,b) 拆分 |
-| `PaimonTableSinkOperatorX::_get_or_create_writer()` | 懒创建 per-(p,b) writer |
-| `PaimonTableSinkOperatorX::_writers` map | per-(p,b) writer 注册表 |
-| `PaimonTableSinkOperatorX::_bucket_key_indices` | bucket key 列索引 |
-| `PaimonTableSinkOperatorX::_partition_col_indices` | partition 列索引 |
-| `PaimonTableSinkOperatorX::_total_buckets` | 总桶数配置 |
-| `PaimonTableSinkOperatorX::_routing_enabled` | 路由开关 |
-| `PaimonTableSinkOperatorX::_eos_sent` | 多 writer EOS 跟踪 |
-
-### 9.2 JNI Backend 实现
-
-```cpp
-// be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h
-
-class JniPaimonWriteBackend : public IPaimonWriteBackend {
-public:
- Status open(const TPaimonTableSink& sink, RuntimeState* state) override;
-
- // 创建 writer——管理所有 partition+bucket，SDK 内部路由
- Status createWriter(std::unique_ptr<IPaimonWriter>* writer) override;
-
- Status createCommitter(
- std::unique_ptr<IPaimonCommitter>* committer) override;
-
- PaimonBackendType type() const override { return PaimonBackendType::JNI; }
-
- bool supportsCompaction() const override { return true; }
- bool supportsLookupChangelogProducer() const override { return true; }
- bool supportsFullCompactionChangelogProducer() const override { return
true; }
- bool supportsPartialUpdateWithDV() const override { return true; }
- bool supportsAggregationWithDV() const override { return true; }
-
-private:
- // JNI 全局引用
- jobject _write_builder; // org.apache.paimon.table.sink.BatchWriteBuilder
- jobject _jni_env_context; // JNI 执行环境
-
- // 桥接方法
- Status _callJavaWrite(jobject writer, Block& block);
- Status _callJavaPrepareCommit(jobject writer,
- std::vector<PaimonCommitMessage>& messages);
- Status _serializeCommitMessages(jobject javaMessages,
- std::vector<PaimonCommitMessage>&
cppMessages);
-};
-
-// JniPaimonWriter 内部持有单个 Java BatchTableWrite 实例，
-// 每次 write() 将 Block 逐行或批量传递给 Java SDK，
-// SDK 内部完成分区/桶路由和文件写入。
-class JniPaimonWriter : public IPaimonWriter {
-public:
- Status write(RuntimeState* state, Block& block) override;
- Status prepareCommit(std::vector<PaimonCommitMessage>& messages) override;
- Status compact(bool fullCompaction) override;
- Status abort() override;
-
-private:
- jobject _table_write; // org.apache.paimon.table.sink.BatchTableWrite
- std::unique_ptr<JniPaimonWriteBackend> _backend; // 非拥有引用，仅用于 JNI 调用
-};
-```
-```
-
-### 9.3 FFI (Rust) Backend 实现
-
-```cpp
-// be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.h
-
-class FfiPaimonWriteBackend : public IPaimonWriteBackend {
-public:
- Status open(const TPaimonTableSink& sink, RuntimeState* state) override;
-
- Status createWriter(std::unique_ptr<IPaimonWriter>* writer) override;
-
- Status createCommitter(
- std::unique_ptr<IPaimonCommitter>* committer) override;
-
- PaimonBackendType type() const override { return PaimonBackendType::FFI; }
-
- // 功能查询 — Rust 目前的能力边界
- bool supportsCompaction() const override { return false; }
- bool supportsLookupChangelogProducer() const override { return false; }
- bool supportsFullCompactionChangelogProducer() const override { return
false; }
- bool supportsPartialUpdateWithDV() const override { return false; }
- bool supportsAggregationWithDV() const override { return false; }
-
-private:
- // 动态加载的 Rust 共享库
- void* _lib_handle;
-
- // C FFI 函数指针
- // 通过 paimon-rust bindings/c 提供
- using PaimonWriteBuilderNew = void* (*)(/* args */);
- using PaimonTableWriteNew = void* (*)(void* builder);
- using PaimonTableWriteArrowBatch = int32_t (*)(void* writer,
- FFI_ArrowArray* array, FFI_ArrowSchema* schema);
- using PaimonTableWritePrepareCommit = void* (*)(void* writer); // returns
CommitMessage[]
- using PaimonTableCommitNew = void* (*)(void* builder);
- using PaimonTableCommitCommit = int32_t (*)(void* commit, void* messages);
- // ... 等等
-
- PaimonWriteBuilderNew _ffi_write_builder_new;
- PaimonTableWriteNew _ffi_table_write_new;
- PaimonTableWriteArrowBatch _ffi_write_arrow_batch;
- PaimonTableWritePrepareCommit _ffi_prepare_commit;
- PaimonTableCommitNew _ffi_table_commit_new;
- PaimonTableCommitCommit _ffi_commit;
-};
-```
-
-**注意**：`paimon-rust/bindings/c/` 目前只暴露了**读取** API。写入 API 需要新增。这是实施路线图中的第一步。
-
----
-
-## 10. 功能路由矩阵
-
-### 10.1 写入模式路由
-
-原则：**优先 Rust，功能不足回退 Java**。Rust 是 Java 的功能子集。
-
-| 表类型 | v1 (仅Java) | v2 首选 | v2 回退 | 回退原因 |
-|--------|------------|---------|---------|----------|
-| Append-only（任意分区） | JNI | **FFI** | JNI | Rust 覆盖最充分 |
-| Append-only + Blob 列 | JNI | **FFI** | JNI | Rust 支持 |
-| PK 表，固定桶 | JNI | **FFI** | JNI | KeyValueFileWriter |
-| PK 表，动态桶 | JNI | **FFI** | JNI | DynamicBucketAssigner |
-| PK 表，Postpone 桶 | JNI | **FFI** | JNI | PostponeFileWriter |
-| ChangelogProducer=Input | JNI | **FFI** | JNI | Rust 支持 |
-| INSERT OVERWRITE | JNI | **FFI** | JNI | Rust 支持 |
-| Truncate table/partition | JNI | **FFI** | JNI | Rust 支持 |
-| ChangelogProducer=Lookup | JNI | JNI | - | Rust **不支持** |
-| ChangelogProducer=FullCompaction | JNI | JNI | - | Rust **不支持** |
-| PartialUpdate + DeletionVector | JNI | JNI | - | Rust **不支持** |
-| Aggregation + DeletionVector | JNI | JNI | - | Rust **不支持** |
-| MERGE INTO (有 PK) | JNI | JNI | - | Rust **不支持** |
-| MERGE INTO (无 PK) | JNI | JNI | - | Rust **不支持** |
-| 需要 Compaction | JNI | JNI | - | Rust **不支持** |
-
-**v2 回退规则**：一旦表的 options 中包含任何 Rust 不支持的功能，整个写入退回到 Java Backend。
-
-### 10.2 功能对齐路线
-
-```
-Phase 1 (v1) — Java JNI Backend + 抽象层：
- ✅ 建立 IPaimonWriteBackend / IPaimonWriter / IPaimonCommitter 接口
- ✅ JniPaimonWriteBackend（完整实现）
- ✅ FfiPaimonWriteBackend（stub，所有方法返回 NotSupported）
- ✅ 结构化 Thrift CommitMessage
- ✅ 单 Writer 架构（对标 Iceberg），Paimon SDK 内部路由
- ✅ Append-only + 固定桶 PK 写入
- ✅ INSERT OVERWRITE / Truncate
- ✅ ChangelogProducer: None / Input
- ✅ 所有 MergeEngine
- ⬜ Compaction（Java 后端）
- ⬜ ChangelogProducer: Lookup / FullCompaction（Java 后端）
-
-Phase 2 (v2) — Rust FFI Backend：
- ⬜ 扩展 paimon-rust C binding（写入 API）
- ⬜ FfiPaimonWriteBackend 从 stub 变为完整实现
- ⬜ 零拷贝 Arrow C Data Interface
- ⬜ Rust tokio runtime 集成
- ⬜ PaimonWriteBackendFactory 自动后端选择
- ⬜ 上层代码零改动（VPaimonTableWriter 不变）
-
-Phase 3（高级功能补齐）：
- ⬜ Compaction 触发与执行（Java 后端）
- ⬜ Lookup / FullCompaction Changelog Producer（Java 后端）
- ⬜ MERGE INTO / DataEvolution（Java 后端）
- ⬜ 动态桶 + 跨分区桶
-
-Phase 4（Rust 功能追赶）：
- ⬜ Rust Compaction 实现
- ⬜ Rust Lookup Changelog Producer
- ⬜ Rust DV + PartialUpdate/Aggregation
-```
-
----
-
-## 11. 实施路线图
-
-### Phase 1：v1 — Java JNI Backend + 预设计 Rust 接口（约 6-8 周）
-
-**目标**：基于 Java SDK 实现功能完备的 Paimon 写入，同时建立 `IPaimonWriteBackend` 抽象为 Rust
后端预留空间。
-
-**核心原则**：虽然只实现 Java 后端，但**架构代码从第一天起就通过抽象接口调用**。新增 Rust 后端 = 新增一个
`IPaimonWriteBackend` 的实现类（<1000行），不动上层代码。
-
-1. **Thrift 定义**
- - 新增 `TPaimonTableSink`（含 `backend_type` 字段，v1 固定为 JNI）
- - 新增结构化 `TPaimonCommitMessage`（含完整 DataFileMeta、IndexFileMeta）
- - 新增 `TPaimonCommitResult`
- - 在 `TDataSink` 中添加 `paimon_table_sink` 字段
-
-2. **BE 端 — 抽象层（关键）**
- - **`IPaimonWriteBackend`** 接口：`open()` / `createWriter()` /
`createCommitter()` / 能力查询
- - **`IPaimonWriter`** 接口：`write()` / `prepareCommit()` / `compact()` /
`abort()`
- - **`IPaimonCommitter`** 接口：`commit()` / `overwrite()` / `truncateTable()`
/ `truncatePartitions()` / `abort()`
- - **`PaimonWriteBackendFactory`**：`selectBackendType()` + `create()`
- - 这些接口从第一天起就存在，Java 后端只是第一个实现
-
-3. **BE 端 — JNI 后端实现**
- - **`JniPaimonWriteBackend`**：实现 `IPaimonWriteBackend`
- - **`VPaimonTableWriter`**：继承 `AsyncResultWriter`，通过 `IPaimonWriteBackend`
接口调用（不直接依赖 JNI）
- - Block → Arrow → Java 的数据桥接（Paimon SDK 内部处理分区/桶路由）
-
-4. **BE 端 — FFI 后端预留**
- - **`FfiPaimonWriteBackend`**：stub 实现，所有方法返回 `Status::NotSupported("FFI
backend not yet implemented")`
- - 后端选择逻辑：`PaimonWriteBackendFactory` 根据 `TPaimonTableSink.backend_type`
创建对应后端
- - v1 中 `backend_type` 固定为 JNI，但代码路径已支持切换
-
-5. **Java 端 (be-java-extensions)**
- - `PaimonJniWriteBackend` — JNI 入口，接收调用并转发给 Paimon SDK
- - `BatchTableWrite` / `StreamTableWrite` 的 JNI wrapper
- - `CommitMessage` 的序列化（Paimon 原生格式），同时转换为结构化 Thrift 格式
-
-6. **FE 端**
- - `PaimonTableSink` — 生成 Thrift sink 描述，v1 固定 `backend_type = JNI`
- - `PaimonWriteCommitCoordinator` / `PaimonTransaction` — 聚合
CommitMessages，协调提交
- - Nereids planner 集成（INSERT INTO paimon_table SELECT ...）
-
-7. **测试**
- - 单元测试：`IPaimonWriteBackend` 接口的 mock 测试
- - 集成测试：Doris → Paimon → Doris 读写闭环
- - 性能测试：与直接使用 Paimon Flink 写入的对比
-
-### Phase 2：v2 — Rust FFI Backend（约 4-6 周）
-
-**目标**：引入 Rust 后端作为优先选择，加速覆盖大部分写入场景。
-
-**关键**：由于 Phase 1 已经建立了 `IPaimonWriteBackend`
抽象，此阶段**上层代码（VPaimonTableWriter、FE Planner）无需修改**。仅需：
-
-1. **Rust 端（核心工作）**
- - 扩展 `paimon-rust/bindings/c/`：新增写入相关的 C FFI 函数
- - `paimon_write_builder_new()` / `paimon_table_write_new()` /
`paimon_write_arrow_batch()`
- - `paimon_prepare_commit()` / `paimon_table_commit_new()` /
`paimon_commit()` / `paimon_commit_abort()`
- - Arrow C Data Interface 支持（ArrowArray/ArrowSchema FFI，零拷贝）
-
-2. **BE 端（替换 stub）**
- - 将 `FfiPaimonWriteBackend` 从 stub 替换为真正的实现（<1000行新代码）
- - 动态加载 Rust 共享库（`dlopen`）
- - Doris Block → Arrow FFI 的零拷贝转换
- - Tokio runtime 集成（异步 I/O）
-
-3. **功能路由（自动切换）**
- - `PaimonWriteBackendFactory::selectBackendType()` 实现自动判断：
- - 检查表 options 是否包含 Rust 不支持的功能
- - 是 → JNI Backend；否 → FFI Backend
- - FE 端 `TPaimonTableSink.backend_type` 从固定 JNI 变为**自动选择**
-
-### Phase 3：高级写入功能（约 6-8 周）
-
-**目标**：对齐 Paimon 的全部写入能力
-
-1. **Compaction 支持**
- - 触发策略配置
- - 显式 `compact()` 调用
- - Compaction 结果的 CommitMessage 处理
-
-2. **Changelog Producer 全支持**
- - Lookup Changelog Producer
- - FullCompaction Changelog Producer
-
-3. **MERGE INTO 支持**
- - DataEvolution 表的 RowTracking 集成
- - Copy-on-Write 合并
-
-4. **Schema Evolution**
- - 写入时的 Schema 兼容性检查
- - 自动 Schema Update
-
-### Phase 4：优化与 Rust 长期演进（持续）
-
-1. **性能优化**
- - BE 端数据批处理优化
- - 零拷贝数据路径
- - 自适应后端选择
-
-2. **Rust 功能跟进**
- - 与 Paimon Rust 社区协作，补齐 Compaction 等功能
- - 逐步将更多功能从 JNI 迁移到 FFI
-
-3. **稳定性**
- - 混沌测试
- - 大规模数据写入测试
- - 多版本兼容性测试
-
-
-## 附录 A：关键文件路径
-
-### Paimon Java SDK
-| 文件 | 说明 |
-|------|------|
-| `paimon-core/.../table/sink/TableWrite.java` | 写入接口 |
-| `paimon-core/.../table/sink/BatchTableWrite.java` | 批写入接口 |
-| `paimon-core/.../table/sink/BatchWriteBuilder.java` | 写入构建器 |
-| `paimon-core/.../table/sink/BatchTableCommit.java` | 批提交接口 |
-| `paimon-core/.../table/sink/StreamTableWrite.java` | 流写入接口 |
-| `paimon-core/.../table/sink/StreamWriteBuilder.java` | 流写入构建器 |
-| `paimon-core/.../table/sink/CommitMessage.java` | 提交消息接口 |
-| `paimon-core/.../table/sink/CommitMessageSerializer.java` | 提交消息序列化器 |
-
-### Paimon Rust
-| 文件 | 说明 |
-|------|------|
-| `crates/paimon/src/table/write_builder.rs` | WriteBuilder 实现 |
-| `crates/paimon/src/table/table_write.rs` | TableWrite 实现（~3000行） |
-| `crates/paimon/src/table/table_commit.rs` | TableCommit 实现（~4800行） |
-| `crates/paimon/src/table/commit_message.rs` | CommitMessage 结构体 |
-| `crates/paimon/src/table/data_file_writer.rs` | Append-only 文件写入器 |
-| `crates/paimon/src/table/kv_file_writer.rs` | PK 表 KeyValue 写入器 |
-| `crates/paimon/src/table/bucket_assigner*.rs` | 桶分配策略 |
-| `bindings/c/src/table.rs` | C FFI 绑定（当前只有读取） |
-| `bindings/c/Cargo.toml` | C 绑定 crate 配置 |
-
-### Doris
-| 文件 | 说明 |
-|------|------|
-| `be/src/exec/sink/writer/async_result_writer.h` | 异步写入器基类 |
-| `be/src/exec/sink/writer/iceberg/viceberg_table_writer.{h,cpp}` | Iceberg
写入参考 |
-| `be/src/format_v2/jni/paimon_jni_reader.{h,cpp}` | Paimon JNI 读取器（参考） |
-| `be/src/format/jni/jni_data_bridge.{h,cpp}` | JNI 数据桥接 |
-| `gensrc/thrift/DataSinks.thrift` | Thrift sink 定义 |
-| `fe/fe-core/.../planner/IcebergTableSink.java` | Iceberg Sink FE 参考 |
-| `fe/fe-core/.../datasource/paimon/` | Paimon 数据源相关 |
-
----
-
-## 附录 B：设计模式与交互规则
-
-### 关键抽象对应关系
-
-```
-Doris 概念 Paimon 概念 说明
-─────────────────────────────────────────────────────
-VPaimonTableWriter BatchWriteBuilder 整个表的写入器（单实例）
-IPaimonWriter BatchTableWrite 管理所有分区/桶的内部路由
-Block (batch) InternalRow / RecordBatch 数据批次
-prepareCommit() prepareCommit() 生成 CommitMessage
-close() + RPC TableCommit.commit() 提交
-```
-
-### CommitMessage 生命周期
-
-```
-创建：IPaimonWriter::prepareCommit()
- ↓
-序列化（Thrift）：BE → FE RPC
- ↓
-聚合：FE Coordinator
- ↓
-去重（idempotency）：对比 commit_user + commit_identifier
- ↓
-提交：Commit BE → IPaimonCommitter::commit()
- ↓
-清理（success）：释放 writer
-清理（failure）：IPaimonCommitter::abort() 删除数据文件
-```
-
----
-
-## 12. 功能矩阵审查：实现 vs 需求
-
-> 基于 `PAIMON_WRITE_FEATURE_MATRIX.md` 对当前 v1 实现的功能覆盖审查。
-
-### 12.1 已覆盖（✅）
-
-| 功能 | 矩阵优先级 | 实现方式 |
-|------|----------|----------|
-| Append-only 表写入 | P0 | JNI backend → Paimon BatchTableWrite (SDK 内部路由) |
-| Primary-key 表写入（full row） | P1 | JNI backend → KeyValueFileWriter，完整 row |
-| Fixed bucket 写入 | P0/P1 | Paimon SDK 内部 MurmurHash 桶计算，Doris 不感知 |
-| Unaware bucket 写入 | P0 | 单 writer，Paimon SDK 内部处理 |
-| Partition 表写入 | P0 | FE 下发 partition columns，Paimon SDK 内部路由 |
-| 非 partition 表写入 | P0 | 单 writer，正常并发写 |
-| `INSERT INTO` | P0 | 主路径，Nereids planner 完整支持 |
-| `INSERT OVERWRITE` | P1 | Thrift 下发 write_mode + static_partition，SDK 内
`withOverwrite()` |
-| Commit message 序列化 | P0 | 结构化 Thrift CommitMessage 传递 |
-| Stream commit 幂等 | P0 | commitUser + txnId |
-| Abort 未提交文件 | P0 | `_writer->abort()` → SDK 内清理 |
-| Changelog producer none | P0 | 默认模式，透明写入 |
-| 多 Catalog 类型 | P1 | FE 下发 Hadoop config + catalog options |
-| 对象存储 | P1 | Hadoop config 透传，支持 S3/OSS/HDFS |
-| 基础类型（Boolean/Integer/Float/String/Decimal/Date） | P0 | Arrow → Paimon 类型转换 |
-| 并发 writer（多 BE） | P0 | BE 天然多实例，FE 汇总 CommitMessage |
-| Nullability 校验 | P0 | Paimon SDK 写入层校验 |
-| 幂等提交 | P0 | commitIdentifier = Doris txnId |
-| 线程数可控 | **P0** | 单 Writer 架构，线程数 = BE worker 数，不随 bucket 数增长 |
-
-### 12.2 部分覆盖（⚠️）
-
-| 功能 | 矩阵优先级 | 现状 | 差距 |
-|------|----------|------|------|
-| Deduplicate merge engine | P1 | 完整 row 写入，理论支持 | 缺少专项测试验证 |
-| 复杂类型（ARRAY/MAP/STRUCT） | P1 | JNI writer 已实现转换逻辑 | 缺少专项测试覆盖 |
-| 小文件控制 | P1 | Paimon SDK 内部按 target-file-size 控制 | SDK 行为，Doris 侧无需干预 |
-| Partition 规范化 | P0 | 依赖 Paimon SDK 内部处理 | 与 Paimon 原生语义完全一致 |
-
-### 12.3 未覆盖 — v2 计划（❌）
-
-| 功能 | 矩阵优先级 | 说明 |
-|------|----------|------|
-| Dynamic hash bucket | P2 | 需要多 assigner 分片并发设计 |
-| Dynamic partition overwrite | P2 | 需要在 static overwrite 稳定后再做 |
-| Row-level delete/update | P3 | 需要 RowKind 映射和 merge engine 联动 |
-| Partial-update merge engine | P3 | Paimon PK writer 路径不支持 `withWriteType` |
-| Aggregation merge engine | P2/P3 | 需要确认聚合字段和 sequence field |
-| First-row merge engine | P3 | 需要明确重复 key 语义 |
-| Sequence field | P2 | 需要从表 option 读取并保证写入列完整性 |
-| Schema evolution | P2 | 需处理计划生成后 schema 变化 |
-| Compaction | P3 | 写入路径不应承担 compaction 调度 |
-| Changelog producer input | P2 | 需要 RowKind 输入支持 |
-| Changelog producer lookup/full-compaction | P3 | 依赖 lookup 或 compaction |
-
-### 12.4 未覆盖 — 暂缓（❌）
-
-| 功能 | 矩阵优先级 | 原因 |
-|------|----------|------|
-| Key dynamic bucket | P3 | 需要全局 key index，复杂度高 |
-| Postpone bucket | P3 | 对 Doris 分发拓扑依赖较强 |
-| Batch commit (`Long.MAX_VALUE`) | 不支持 | 与 Doris 外部事务模型不匹配 |
-
-### 12.5 关键差距总结
-
-当前实现覆盖了矩阵中 **P0 项目的 100%**，P1 项目的 **~80%**。
-
-与 v0（per-bucket writer）相比，v1 架构调整的核心变化：
-- **移除** BE 端 partition/bucket 计算与路由（`paimon_bucket.h/.cpp`、operator 层路由逻辑）
-- **移除** per-(partition, bucket) VPaimonTableWriter 创建与生命周期管理
-- **新增** 单一 VPaimonTableWriter，对标 Iceberg 的 VIcebergTableWriter 架构
-- **复用** Paimon SDK 的 `TableWrite::write()` 内部路由机制
-
-剩余差距集中在 P2/P3 级别：Dynamic bucket、Partial-update、Row-level delete/update、
-Compaction、Changelog producer (input/lookup/full-compaction) 等，按设计文档
-Phase 2-4 路线图推进。
-
----
-
-## 13. 测试计划
-
-### 13.1 功能测试矩阵
-
-#### 13.1.1 基础写入场景
-
-| 用例 ID | 场景 | 表配置 | 数据特征 | 验证点 |
-|---------|------|--------|---------|--------|
-| FT-001 | Append-only 无分区无 bucket | `bucket=1`, 无 PK, 无 partition | 1000 行，3
列 INT/VARCHAR/DOUBLE | commit 成功，Paimon 可读回全部行 |
-| FT-002 | Append-only 有分区 | 按 region 分区, 无 PK | 5 个分区，每分区 200 行 |
分区路径正确，manifest 正确 |
-| FT-003 | Append-only + fixed bucket | `bucket=4`, `bucket-key=id` | 10000 行
id 从 1-10000 | 4 个 bucket 均有数据文件 |
-| FT-004 | PK 表 + fixed bucket + full row | `bucket=2`, PK=id,
merge-engine=deduplicate | 1000 行含重复 id | 重复 key 保留最新值 |
-| FT-005 | 多 BE 并发写入 | bucket unaware, 4 个 BE | 每个 BE 写 5000 行 | commit
message 去重，总行数正确 |
-
-#### 13.1.2 事务与容错
-
-| 用例 ID | 场景 | 操作 | 验证点 |
-|---------|------|------|--------|
-| FT-010 | FE commit 成功 | INSERT INTO → commit | Snapshot 创建，可读 |
-| FT-011 | FE commit 失败 → rollback | 模拟 commit 异常 | abort 清理数据文件，无孤儿文件 |
-| FT-012 | BE 写入失败 → abort | BE 端 JNI 调用异常 | writer.abort() 被调用 |
-| FT-013 | Fragment retry 去重 | 同 txnId 多次上报 commit message | base64 去重生效 |
-| FT-014 | 空表写入 | 0 行数据 | prepareCommit 返回空，不创建 Snapshot |
-| FT-015 | 大事务提交 | 10 万行，4 BE | commit message 分块正确，payload ≤ 8MB |
-
-#### 13.1.3 类型覆盖
-
-| 用例 ID | Paimon 类型 | Doris 类型 | 特殊值 |
-|---------|-----------|-----------|--------|
-| FT-020 | BOOLEAN | BOOLEAN | true, false, NULL |
-| FT-021 | INT / BIGINT | INT / BIGINT | 0, -1, INT_MAX, INT_MIN, NULL |
-| FT-022 | FLOAT / DOUBLE | FLOAT / DOUBLE | 0.0, NaN, INF, -0.0, NULL |
-| FT-023 | DECIMAL(10,2) | DECIMAL(10,2) | 0, 99999999.99, -1.50, NULL |
-| FT-024 | VARCHAR(100) | VARCHAR(100) | "", "hello", 100-char string, NULL |
-| FT-025 | CHAR(10) | CHAR(10) | "abc", 10-char pad, NULL |
-| FT-026 | DATE | DATE | 1970-01-01, 2099-12-31, NULL |
-| FT-027 | TIMESTAMP(6) | DATETIME(6) | 微秒精度, NULL |
-| FT-028 | ARRAY<INT> | ARRAY<INT> | [], [1,2,3], NULL |
-| FT-029 | MAP<STRING,INT> | MAP<STRING,INT> | {}, {"a":1}, NULL |
-| FT-030 | ROW<name STRING, age INT> | STRUCT | 嵌套 NULL 字段 |
-
-#### 13.1.4 边界场景
-
-| 用例 ID | 场景 | 验证点 |
-|---------|------|--------|
-| FT-040 | NULL 列全部为 NULL | 写入正确，查询返回 NULL |
-| FT-041 | 主键全 NULL（如允许） | 行为符合 Paimon 规范 |
-| FT-042 | 超长 VARCHAR > 65535 | 写入失败有清晰错误信息 |
-| FT-043 | DECIMAL 精度溢出 | 写入失败有清晰错误信息 |
-| FT-044 | Bucket key 为复合键 | 路由正确 |
-
-#### 13.1.5 存储后端
-
-| 用例 ID | 存储 | 验证点 |
-|---------|------|--------|
-| FT-050 | Local FS | 基本功能 |
-| FT-051 | HDFS | rename 语义, commit |
-| FT-052 | S3 (MinIO 模拟) | 对象存储 rename 行为 |
-
-### 13.2 性能测试计划
-
-#### 13.2.1 吞吐量基准测试
-
-**目标**：确定 Doris → Paimon 写入的吞吐基线。
-
-| 测试场景 | 表类型 | 数据量 | BE 数 | 对比基准 |
-|---------|--------|--------|-------|----------|
-| PT-001 | Append-only, bucket unaware | 1 亿行, 100GB | 1/4/8 | Paimon Flink
Writer 同等条件 |
-| PT-002 | PK 表, fixed bucket=16 | 1 亿行, 100GB | 1/4/8 | Paimon Flink Writer
同等条件 |
-| PT-003 | Append-only, partition=10 | 1 亿行, 100GB | 4 | Paimon Flink Writer
同等条件 |
-
-**测量指标**：
-- 写入吞吐（rows/s, MB/s）
-- BE 内存占用（peak, avg）
-- Arrow IPC 序列化/反序列化耗时占比
-- JNI 调用耗时占比
-- Paimon FileStoreWrite 耗时占比
-- Commit 耗时（prepareCommit + FE commit）
-
-#### 13.2.2 扩展性测试
-
-| 测试场景 | 变量 | 期望 |
-|---------|------|------|
-| PT-010 | BE 数 1→2→4→8 | 吞吐接近线性增长 |
-| PT-011 | 分区数 1→10→100 | 吞吐稳定，文件数可控 |
-| PT-012 | Bucket 数 1→4→16→64 | 并发 writer 数增长，吞吐改善 |
-
-#### 13.2.3 大数据量压力测试
-
-| 测试场景 | 数据量 | 关注点 |
-|---------|--------|--------|
-| PT-020 | 10 亿行, 1TB Append-only | 长时间运行稳定性，内存无泄漏 |
-| PT-021 | 10 亿行, 1TB PK 表 | merge engine 排序内存，spill 行为 |
-| PT-022 | 1000 分区, 每分区 100 万行 | manifest 膨胀，commit 性能 |
-
-#### 13.2.4 资源限制测试
-
-| 测试场景 | 限制 | 验证点 |
-|---------|------|--------|
-| PT-030 | BE 内存限制 4GB | spill 正常触发，不 OOM |
-| PT-031 | 限流场景（网络带宽 100Mbps） | 吞吐自适应，无超时重试雪崩 |
-| PT-032 | 并发 10 个 INSERT 任务 | 资源隔离，无互相影响 |
-
-### 13.3 测试基础设施
-
-#### 13.3.1 回归测试框架
-
-```
-regression-test/suites/paimon_write_p0/
-├── test_insert_append_only.groovy # FT-001
-├── test_insert_partitioned.groovy # FT-002
-├── test_insert_fixed_bucket.groovy # FT-003
-├── test_insert_pk_dedup.groovy # FT-004
-├── test_insert_multi_be.groovy # FT-005
-├── test_transaction_commit.groovy # FT-010
-├── test_transaction_rollback.groovy # FT-011
-├── test_fragment_retry_dedup.groovy # FT-013
-├── test_empty_write.groovy # FT-014
-├── test_types_basic.groovy # FT-020..027
-├── test_types_complex.groovy # FT-028..030
-├── test_nullability.groovy # FT-040
-├── test_boundary.groovy # FT-041..044
-├── test_storage_hdfs.groovy # FT-051
-└── test_storage_s3.groovy # FT-052
-```
-
-#### 13.3.2 性能测试脚本
-
-```
-tools/paimon_perf/
-├── benchmark.sh # 一键启动性能测试
-├── schema/
-│ ├── append_only_ddl.sql # Append-only 表 DDL
-│ ├── pk_table_ddl.sql # PK 表 DDL
-│ └── partitioned_table_ddl.sql # 分区表 DDL
-├── workload/
-│ ├── gen_data.sh # 数据生成（TPC-H / TPC-DS）
-│ └── run_benchmark.sh # 执行基准测试
-└── report/
- └── analyze.sh # 结果汇总和分析
-```
-
-### 13.4 推荐里程碑
-
-与 `PAIMON_WRITE_FEATURE_MATRIX.md` Section 13 对齐：
-
-| 里程碑 | 覆盖范围 | 验收标准 | 测试用例 |
-|--------|---------|---------|----------|
-| **M1** | append-only + bucket unaware + INSERT INTO | 多 BE 并发写成功，FE commit
成功，失败可 abort | FT-001, FT-010..014, PT-001 |
-| **M2** | partition + fixed bucket | 按 partition/bucket 并发写，Paimon 正确读取 |
FT-002..003, FT-005, PT-003 |
-| **M3** | primary-key fixed bucket full row | PK upsert 语义，重复 key 符合 merge
engine | FT-004, PT-002 |
-| **M4** | 类型全覆盖 + 复杂类型 | 全部基础类型 + ARRAY/MAP/STRUCT | FT-020..030 |
-| **M5** | 存储后端全覆盖 | HDFS + S3 通过回归测试 | FT-050..052 |
-| **M6** | INSERT OVERWRITE + dynamic bucket | v2 功能 | 待 v2 定义 |
-
-### 13.5 M1 准入标准（当前 v1 应达到）
-
-1. ✅ FE 编译通过，checkstyle 0 违规
-2. ⬜ BE 编译通过（CI 验证）
-3. ⬜ FT-001 ~ FT-005 全部通过的回归测试
-4. ⬜ FT-010 ~ FT-015 事务容错测试通过
-5. ⬜ PT-001 吞吐不低于 Flink Writer 的 70%
-6. ⬜ 10 小时连续写入无内存泄漏
-
----
-
-> **文档版本**：v8.0
-> **作者**：Doris Paimon Write Team
-> **日期**：2026-07-13
-> **状态**：架构重构：Single Writer（对标 Iceberg）+ Paimon SDK 内部路由。移除 BE 端
partition/bucket 路由计算。
-> **变更摘要**：
-> - 从 "One Writer per (Partition,Bucket)" 改为 "Single Writer" 架构
-> - 分区路由由 Paimon SDK 内部处理，Doris 不再参与路由计算
-> - 移除 `paimon_bucket.h/.cpp`、operator 层路由/拆分/多 writer 管理逻辑
-> - 线程数从 O(partition_count × bucket_count) 降为 O(BE_worker_count)
-> - `IPaimonWriter::write()` 接收完整 Block，SDK 内部路由
-> - `IPaimonWriteBackend::createWriter()` 不再需要 partition/bucket 参数
diff --git
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
index dbb5ead0dc5..2eac6e95f56 100644
---
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
+++
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
@@ -47,14 +47,17 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.io.BundleRecords;
import org.apache.paimon.io.DataOutputSerializer;
import org.apache.paimon.memory.HeapMemorySegmentPool;
import org.apache.paimon.options.Options;
@@ -86,6 +89,8 @@ import java.nio.file.Paths;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -98,8 +103,9 @@ import java.util.stream.Collectors;
* C++ Block → Arrow IPC Stream → direct memory address
* → PaimonJniWriter.write(address, length)
* → ArrowStreamReader → VectorSchemaRoot
- * → convert row-by-row → Paimon GenericRow
- * → BatchTableWrite.write(row) [SDK handles partition+bucket routing]
+ * → extractColumnValues (column-major typed extraction)
+ * → group by writer.getPartition(row) + writer.getBucket(row)
+ * → writer.writeBundle(partition, bucket, bundle) [batch write]
*
* Commit path:
*
@@ -395,94 +401,170 @@ public class PaimonJniWriter {
List<FieldVector> vectors = root.getFieldVectors();
int colCount = fields.size();
DataType[] currentTypes = resolveTargetTypes(fields);
- GenericRow reusedRow = new GenericRow(colCount);
- for (int i = 0; i < rowCount; i++) {
- for (int col = 0; col < colCount; col++) {
- try {
- reusedRow.setField(col,
- readArrowValue(vectors.get(col), i,
fields.get(col), currentTypes[col]));
- } catch (Throwable t) {
- throw new RuntimeException("PaimonJniWriter row conversion
failed: row=" + i
- + ", col=" + col + ", field=" +
fields.get(col).getName()
- + ", targetType=" + (currentTypes[col] != null
- ? currentTypes[col].asSQLString() :
"null"), t);
- }
- }
- try {
- // Generic write — Paimon SDK handles partition computation
- // (getPartition) and bucket computation (getBucket)
internally.
- writer.write(reusedRow);
- } catch (Throwable t) {
- throw new RuntimeException("PaimonJniWriter write failed:
row=" + i
- + ", totalRows=" + rowCount, t);
+ // Column-major extraction: one instanceof per column, N direct getter
calls.
+ Object[][] columnValues = new Object[colCount][];
+ for (int col = 0; col < colCount; col++) {
+ columnValues[col] = extractColumnValues(vectors.get(col),
fields.get(col),
+ currentTypes[col], rowCount);
+ }
+
+ // Group rows by (partition, bucket) using the Paimon SDK's routing
methods,
+ // then write each group via writeBundle for batch-level I/O
optimization.
+ Map<PartitionBucketKey, List<InternalRow>> groups = new
LinkedHashMap<>();
+ for (int r = 0; r < rowCount; r++) {
+ GenericRow row = new GenericRow(colCount);
+ for (int c = 0; c < colCount; c++) {
+ row.setField(c, columnValues[c][r]);
}
+ BinaryRow partition = writer.getPartition(row);
+ int bucket = writer.getBucket(row);
+ groups.computeIfAbsent(new PartitionBucketKey(partition, bucket),
+ k -> new ArrayList<>()).add(row);
+ }
+
+ for (Map.Entry<PartitionBucketKey, List<InternalRow>> entry :
groups.entrySet()) {
+ PartitionBucketKey key = entry.getKey();
+ writer.writeBundle(key.partition, key.bucket, new
ListBundleRecords(entry.getValue()));
}
}
- // ────────────────────────────────────────────────────────────
- // Arrow → Paimon type conversion
- // ────────────────────────────────────────────────────────────
+ /**
+ * Extract all values for one column using a type-specific fast path.
+ * Single instanceof per column — no per-cell type dispatch.
+ */
+ private Object[] extractColumnValues(FieldVector vector, Field arrowField,
+ DataType targetType, int rowCount) {
+ Object[] values = new Object[rowCount];
- private Object readArrowValue(FieldVector vector, int row, Field
arrowField,
- DataType targetType) {
- if (vector == null || vector.isNull(row)) {
- return null;
- }
- if (targetType instanceof BinaryType || targetType instanceof
VarBinaryType) {
- if (vector instanceof VarBinaryVector) {
- return ((VarBinaryVector) vector).get(row);
- }
- if (vector instanceof VarCharVector) {
- return ((VarCharVector) vector).get(row);
- }
- } else {
- if (vector instanceof VarCharVector) {
- return BinaryString.fromBytes(((VarCharVector)
vector).get(row));
+ // ── Primitive int types ──────────────────────────────
+ if (vector instanceof IntVector) {
+ IntVector v = (IntVector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
}
+ return values;
}
- if (vector instanceof BitVector) {
- return ((BitVector) vector).get(row) == 1;
- }
- if (vector instanceof TinyIntVector) {
- return ((TinyIntVector) vector).get(row);
+ if (vector instanceof BigIntVector) {
+ BigIntVector v = (BigIntVector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
+ }
+ return values;
}
if (vector instanceof SmallIntVector) {
- return ((SmallIntVector) vector).get(row);
- }
- if (vector instanceof IntVector) {
- return ((IntVector) vector).get(row);
+ SmallIntVector v = (SmallIntVector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
+ }
+ return values;
}
- if (vector instanceof BigIntVector) {
- return ((BigIntVector) vector).get(row);
+ if (vector instanceof TinyIntVector) {
+ TinyIntVector v = (TinyIntVector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
+ }
+ return values;
}
if (vector instanceof Float4Vector) {
- return ((Float4Vector) vector).get(row);
+ Float4Vector v = (Float4Vector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
+ }
+ return values;
}
if (vector instanceof Float8Vector) {
- return ((Float8Vector) vector).get(row);
+ Float8Vector v = (Float8Vector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
+ }
+ return values;
+ }
+ if (vector instanceof BitVector) {
+ BitVector v = (BitVector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i) == 1;
+ }
+ return values;
}
if (vector instanceof DateDayVector) {
- return ((DateDayVector) vector).get(row);
+ DateDayVector v = (DateDayVector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
+ }
+ return values;
+ }
+
+ // ── String / Binary ──────────────────────────────────
+ if (vector instanceof VarCharVector) {
+ VarCharVector v = (VarCharVector) vector;
+ if (targetType instanceof BinaryType || targetType instanceof
VarBinaryType) {
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
+ }
+ } else {
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null :
BinaryString.fromBytes(v.get(i));
+ }
+ }
+ return values;
}
+ if (vector instanceof VarBinaryVector) {
+ VarBinaryVector v = (VarBinaryVector) vector;
+ for (int i = 0; i < rowCount; i++) {
+ values[i] = v.isNull(i) ? null : v.get(i);
+ }
+ return values;
+ }
+
+ // ── Timestamp ────────────────────────────────────────
if (vector instanceof TimeStampMicroVector) {
+ TimeStampMicroVector v = (TimeStampMicroVector) vector;
NullableTimeStampMicroHolder holder = new
NullableTimeStampMicroHolder();
- ((TimeStampMicroVector) vector).get(row, holder);
- return Timestamp.fromMicros(holder.value);
+ for (int i = 0; i < rowCount; i++) {
+ if (v.isNull(i)) {
+ values[i] = null;
+ } else {
+ v.get(i, holder);
+ values[i] = Timestamp.fromMicros(holder.value);
+ }
+ }
+ return values;
}
+
+ // ── Decimal ──────────────────────────────────────────
if (vector instanceof DecimalVector) {
- return readDecimal((DecimalVector) vector, row);
+ DecimalVector v = (DecimalVector) vector;
+ int precision = v.getPrecision();
+ int scale = v.getScale();
+ org.apache.arrow.memory.ArrowBuf dataBuf = v.getDataBuffer();
+ for (int i = 0; i < rowCount; i++) {
+ if (v.isNull(i)) {
+ values[i] = null;
+ } else {
+ BigDecimal bd = getBigDecimalFromArrowBuf(dataBuf, i,
scale,
+ DecimalVector.TYPE_WIDTH);
+ values[i] = Decimal.fromBigDecimal(bd, precision, scale);
+ }
+ }
+ return values;
}
- Object val = vector.getObject(row);
- return convertToPaimonType(val, arrowField, targetType);
- }
- private Decimal readDecimal(DecimalVector vector, int row) {
- BigDecimal bd = getBigDecimalFromArrowBuf(vector.getDataBuffer(), row,
- vector.getScale(), DecimalVector.TYPE_WIDTH);
- return Decimal.fromBigDecimal(bd, vector.getPrecision(),
vector.getScale());
+ // ── Complex types (Array / Map / Struct) — fallback ──
+ for (int i = 0; i < rowCount; i++) {
+ if (vector.isNull(i)) {
+ values[i] = null;
+ } else {
+ values[i] = convertToPaimonType(vector.getObject(i),
arrowField, targetType);
+ }
+ }
+ return values;
}
+ // ────────────────────────────────────────────────────────────
+ // Arrow → Paimon type conversion (used by extractColumnValues fallback)
+ // ────────────────────────────────────────────────────────────
+
@SuppressWarnings({"unchecked", "rawtypes"})
private Object convertToPaimonType(Object val, Field arrowField, DataType
targetType) {
if (val == null) {
@@ -820,6 +902,57 @@ public class PaimonJniWriter {
+ ", writerNull=" + (writer == null);
}
+ // ────────────────────────────────────────────────────────────
+ // Bundle helpers
+ // ────────────────────────────────────────────────────────────
+
+ /** Compound key for grouping rows by (partition, bucket). */
+ private static final class PartitionBucketKey {
+ final BinaryRow partition;
+ final int bucket;
+
+ PartitionBucketKey(BinaryRow partition, int bucket) {
+ this.partition = partition;
+ this.bucket = bucket;
+ }
+
+ @Override
+ public int hashCode() {
+ return partition.hashCode() * 31 + bucket;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof PartitionBucketKey)) {
+ return false;
+ }
+ PartitionBucketKey other = (PartitionBucketKey) obj;
+ return bucket == other.bucket && partition.equals(other.partition);
+ }
+ }
+
+ /** BundleRecords backed by a List of pre-built InternalRows. */
+ private static final class ListBundleRecords implements BundleRecords {
+ private final List<InternalRow> rows;
+
+ ListBundleRecords(List<InternalRow> rows) {
+ this.rows = rows;
+ }
+
+ @Override
+ public long rowCount() {
+ return rows.size();
+ }
+
+ @Override
+ public Iterator<InternalRow> iterator() {
+ return rows.iterator();
+ }
+ }
+
/** InputStream over a direct ByteBuffer (no copy). */
private static class DirectBufInputStream extends InputStream {
private final ByteBuffer buf;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java
index 2b98c2a4618..f8d53ba3e23 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java
@@ -21,10 +21,8 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.datasource.paimon.PaimonExternalDatabase;
import org.apache.doris.datasource.paimon.PaimonExternalTable;
import org.apache.doris.nereids.memo.GroupExpression;
-import org.apache.doris.nereids.properties.DistributionSpecHash;
import org.apache.doris.nereids.properties.LogicalProperties;
import org.apache.doris.nereids.properties.PhysicalProperties;
-import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.plans.AbstractPlan;
import org.apache.doris.nereids.trees.plans.Plan;
@@ -32,10 +30,8 @@ import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.statistics.Statistics;
-import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
-import java.util.stream.Collectors;
/**
* Physical Paimon table sink.
@@ -103,31 +99,8 @@ public class PhysicalPaimonTableSink<CHILD_TYPE extends
Plan>
@Override
public PhysicalProperties getRequirePhysicalProperties() {
- if (physicalProperties != null) {
- return physicalProperties;
- }
- // For HASH_FIXED tables, require hash distribution on (partition cols
+ bucket key cols)
- // so that the same (partition, bucket) data is routed to the same BE.
- // This allows one AsyncResultWriter per (partition, bucket) with no
cross-BE conflicts.
- PaimonExternalTable table = (PaimonExternalTable) targetTable;
- List<String> partitionKeys = table.getPartitionColumnNames();
- List<String> bucketKeys = table.getBucketKeys();
-
- List<Integer> shuffleColIdx = new ArrayList<>();
- List<Column> fullSchema = table.getFullSchema();
- for (int i = 0; i < fullSchema.size(); i++) {
- String colName = fullSchema.get(i).getName();
- if (partitionKeys.contains(colName) ||
bucketKeys.contains(colName)) {
- shuffleColIdx.add(i);
- }
- }
-
- if (!shuffleColIdx.isEmpty()) {
- List<ExprId> exprIds = shuffleColIdx.stream()
- .map(idx -> child().getOutput().get(idx).getExprId())
- .collect(Collectors.toList());
- return new PhysicalProperties(new DistributionSpecHash(exprIds,
DistributionSpecHash.ShuffleType.REQUIRE));
- }
+ // Partition and bucket routing is handled by the Paimon SDK
internally.
+ // Doris does not need to shuffle data by partition/bucket keys.
return PhysicalProperties.SINK_RANDOM_PARTITIONED;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java
index 0c83f41e366..0fb6d392a09 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java
@@ -51,9 +51,9 @@ import java.util.Set;
* Paimon table sink.
*
* Generates TPaimonTableSink payload consumed by BE, including table location,
- * Paimon options, Hadoop config, bucket metadata, and sink column names.
+ * Paimon options, Hadoop config, write mode, and sink column names.
*
- * v1: always sets backend_type = JNI.
+ * v1: single-writer architecture; partition/bucket routing delegated to SDK.
*/
public class PaimonTableSink extends BaseExternalTableDataSink {
private static final Logger LOG =
LogManager.getLogger(PaimonTableSink.class);
@@ -162,30 +162,7 @@ public class PaimonTableSink extends
BaseExternalTableDataSink {
tSink.setSerializedTable(encodeObjectToString(paimonTable));
}
- // Bucket info
- if (paimonTable instanceof org.apache.paimon.table.FileStoreTable) {
- org.apache.paimon.table.FileStoreTable fst =
- (org.apache.paimon.table.FileStoreTable) paimonTable;
- int bucketNum = fst.schema().numBuckets();
- if (bucketNum > 0) {
- tSink.setBucketNum(bucketNum);
- }
- // Extract bucket key column indices for BE-side routing
- List<String> bucketKeys = fst.schema().bucketKeys();
- if (!bucketKeys.isEmpty() && cols != null) {
- List<Integer> bucketKeyIndices = new ArrayList<>();
- for (int i = 0; i < cols.size(); i++) {
- if (bucketKeys.contains(cols.get(i).getName())) {
- bucketKeyIndices.add(i);
- }
- }
- if (!bucketKeyIndices.isEmpty()) {
- tSink.setBucketKeyIndices(bucketKeyIndices);
- }
- }
- }
-
- // Write mode: APPEND or OVERWRITE
+ // Serialize table for BE-side fast loading
tSink.setBackendType(TPaimonWriteBackendType.JNI);
if (insertCtx.isPresent() && insertCtx.get() instanceof
PaimonInsertCommandContext) {
PaimonInsertCommandContext ctx =
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index a4078c96e42..3f6eeb438c9 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -630,9 +630,7 @@ enum TPaimonWriteMode {
}
struct TPaimonCommitMessage {
- 1: optional binary partition // serialized BinaryRow bytes
- 2: optional i32 bucket
- 3: optional binary payload // Paimon native
CommitMessageSerializer bytes
+ 1: optional binary payload // Paimon native
CommitMessageSerializer bytes (DPCM-framed)
}
struct TPaimonTableSink {
@@ -641,13 +639,11 @@ struct TPaimonTableSink {
3: optional string table_location
4: optional map<string, string> paimon_options
5: optional map<string, string> hadoop_config
- 6: optional i32 bucket_num
- 7: optional list<string> column_names
- 8: optional string serialized_table // serialized Paimon Table
object (base64)
- 9: optional TPaimonWriteBackendType backend_type // v1 fixed to JNI
- 10: optional TPaimonWriteMode write_mode // APPEND or OVERWRITE
- 11: optional list<i32> bucket_key_indices // indices of
bucket-key columns in column_names
- 12: optional map<string, string> static_partition // static partition
spec for INSERT OVERWRITE
+ 6: optional list<string> column_names
+ 7: optional string serialized_table // serialized Paimon Table
object (base64)
+ 8: optional TPaimonWriteBackendType backend_type
+ 9: optional TPaimonWriteMode write_mode
+ 10: optional map<string, string> static_partition // static partition
spec for INSERT OVERWRITE
}
struct TDataSink {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
