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
commit 85b24ad0214e5b434cd3fef46817025d157aff89 Author: Socrates <[email protected]> AuthorDate: Mon Jul 13 12:49:30 2026 +0800 [refactor](paimon) Single Writer architecture — SDK-internal routing ### What problem does this PR solve? Issue Number: N/A Problem Summary: The v0 Paimon write implementation used a "one AsyncResultWriter per (partition, bucket)" design. Each AsyncResultWriter creates an IO thread in the fragment_mgr thread pool, so tables with many buckets caused thread explosion (e.g. 128 buckets × 10 partitions = 1280 threads). This refactor switches to the same single-writer architecture used by Icebergs VIcebergTableWriter. A single VPaimonTableWriter (one AsyncResultWriter, one IO thread) manages the entire table. Partition and bucket routing is delegated to the Paimon SDK (Java or Rust) via the generic write(row) interface, instead of being computed in Doris BE. Root cause: AsyncResultWriter::start_writer() submits a long-running process_block() task to fragment_mgr thread pool. With many (p,b) pairs, each creating its own AsyncResultWriter, thread count scaled with the product of partition count × bucket count. Key changes: - BE: IPaimonWriteBackend::create_writer() removed partition_bytes/bucket params - BE: IPaimonWriter::write() takes full Block, SDK does internal routing - BE: PaimonTableSinkOperatorX simplified from routing+split+multi-writer (~160 lines) to simple pass-through to AsyncWriterSink (~40 lines) - Java: PaimonJniWriter always uses writer.write(row), removed RowKeyExtractor, bucketMode, useExplicitBucketWrite, initBucketMode() - Thread count: O(partition_count × bucket_count) → O(BE_worker_count) - Design doc updated from v7.0 (multi-writer) to v8.0 (single-writer) ### Release note None (this branch is not yet released) ### Check List (For Author) - Test: No regression tests run yet (WIP feature on branch) - Behavior changed: Yes (architecture change, not user-visible yet) - Does this need documentation: No (design doc updated in this commit) --- .../exec/operator/paimon_table_sink_operator.cpp | 132 +----- be/src/exec/operator/paimon_table_sink_operator.h | 55 +-- .../writer/paimon/ffi_paimon_write_backend.cpp | 3 +- .../sink/writer/paimon/ffi_paimon_write_backend.h | 3 +- .../writer/paimon/jni_paimon_write_backend.cpp | 7 +- .../sink/writer/paimon/jni_paimon_write_backend.h | 12 +- .../exec/sink/writer/paimon/paimon_write_backend.h | 21 +- .../sink/writer/paimon/vpaimon_table_writer.cpp | 13 +- .../exec/sink/writer/paimon/vpaimon_table_writer.h | 30 +- .../paimon-scanner/PAIMON_WRITE_DESIGN.md | 495 ++++++++++++++------- .../org/apache/doris/paimon/PaimonJniWriter.java | 40 +- 11 files changed, 396 insertions(+), 415 deletions(-) diff --git a/be/src/exec/operator/paimon_table_sink_operator.cpp b/be/src/exec/operator/paimon_table_sink_operator.cpp index b775e6200a1..e177682bb2c 100644 --- a/be/src/exec/operator/paimon_table_sink_operator.cpp +++ b/be/src/exec/operator/paimon_table_sink_operator.cpp @@ -18,15 +18,11 @@ #include "exec/operator/paimon_table_sink_operator.h" #include "common/logging.h" -#include "pipeline/exec/operator.h" namespace doris { Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) { - RETURN_IF_ERROR(Base::init(state, info)); - // _writer is set lazily per (partition, bucket) in the operator, - // so we don't create a single _writer here. - return Status::OK(); + return Base::init(state, info); } Status PaimonTableSinkOperatorX::sink_impl(RuntimeState* state, Block* in_block, bool eos) { @@ -34,128 +30,10 @@ Status PaimonTableSinkOperatorX::sink_impl(RuntimeState* state, Block* in_block, SCOPED_TIMER(local_state.exec_time_counter()); COUNTER_UPDATE(local_state.rows_input_counter(), static_cast<int64_t>(in_block->rows())); - if (in_block->rows() == 0) { - if (eos && !_eos_sent) { - _eos_sent = true; - for (auto& [key, writer] : _writers) { - RETURN_IF_ERROR(writer->sink(in_block, true)); - } - } - return Status::OK(); - } - - // Project: apply output expressions - Block output_block; - { - SCOPED_TIMER(local_state.exec_time_counter()); - RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs( - _output_vexpr_ctxs, *in_block, &output_block, false)); - } - - // Compute (partition_bytes, bucket) for each row - std::vector<std::string> partition_bytes; - std::vector<int32_t> buckets; - RETURN_IF_ERROR(_compute_routing(output_block, partition_bytes, buckets)); - - // Split block by (partition_bytes, bucket) - auto grouped = _split_by_key(output_block, partition_bytes, buckets); - - // Route each group to the corresponding writer - for (auto& [key, sub_block] : grouped) { - VPaimonTableWriter* writer = nullptr; - RETURN_IF_ERROR(_get_or_create_writer(state, key, &writer)); - RETURN_IF_ERROR(writer->sink(sub_block.get(), false)); - } - - if (eos && !_eos_sent) { - _eos_sent = true; - Block empty_block; - for (auto& [key, writer] : _writers) { - RETURN_IF_ERROR(writer->sink(&empty_block, true)); - } - } - - return Status::OK(); -} - -Status PaimonTableSinkOperatorX::_get_or_create_writer( - RuntimeState* state, const PaimonPartitionBucketKey& key, - VPaimonTableWriter** writer) { - auto it = _writers.find(key); - if (it != _writers.end()) { - *writer = it->second.get(); - return Status::OK(); - } - - const auto& paimon_sink = _t_output_expr; // need TDataSink reference - // Access TDataSink from the local state info - // We get it from the parent's init path - auto new_writer = std::make_shared<VPaimonTableWriter>( - _t_sink, _output_vexpr_ctxs, _dependency, _finish_dependency); - RETURN_IF_ERROR(new_writer->init_properties(_pool)); - RETURN_IF_ERROR(new_writer->open(state, _operator_profile)); - - _writers[key] = new_writer; - *writer = new_writer.get(); - - LOG(INFO) << "Created writer for (partition=" << key.first.size() << "bytes" - << ", bucket=" << key.second << "), total writers: " << _writers.size(); - return Status::OK(); -} - -Status PaimonTableSinkOperatorX::_compute_routing( - const Block& block, - std::vector<std::string>& partition_bytes, - std::vector<int32_t>& buckets) { - size_t rows = block.rows(); - partition_bytes.resize(rows); - buckets.resize(rows); - - // Compute bucket ids using BE-side MurmurHash - if (_routing_enabled && !_bucket_key_indices.empty()) { - buckets = paimon_compute_buckets(block, _bucket_key_indices, _total_buckets); - } else { - std::fill(buckets.begin(), buckets.end(), 0); - } - - // Compute partition bytes: serialize partition col values to BinaryRow format. - // For now, use empty partition (Java SDK handles it internally). - // Full partition computation in BE will be added in a follow-up. - for (size_t i = 0; i < rows; ++i) { - partition_bytes[i].clear(); - } - - return Status::OK(); -} - -std::unordered_map<PaimonPartitionBucketKey, std::unique_ptr<Block>> -PaimonTableSinkOperatorX::_split_by_key( - const Block& block, - const std::vector<std::string>& partition_bytes, - const std::vector<int32_t>& buckets) { - std::unordered_map<PaimonPartitionBucketKey, std::unique_ptr<Block>> result; - - // Group row indices by (partition_bytes, bucket) - std::unordered_map<PaimonPartitionBucketKey, std::vector<size_t>> key_rows; - for (size_t i = 0; i < block.rows(); ++i) { - PaimonPartitionBucketKey key = {partition_bytes[i], buckets[i]}; - key_rows[key].push_back(i); - } - - // Build sub-blocks for each key - for (auto& [key, row_indices] : key_rows) { - auto sub_block = Block::create_unique(block.clone_empty()); - auto columns = std::move(*sub_block).mutate_columns(); - for (int col = 0; col < block.columns(); ++col) { - const auto& src_col = block.get_by_position(col).column; - for (size_t idx : row_indices) { - columns[col]->insert_from(*src_col, idx); - } - } - sub_block->set_columns(std::move(columns)); - result[key] = std::move(sub_block); - } - return result; + // Delegate directly to the single AsyncWriterSink → VPaimonTableWriter. + // Partition and bucket routing is handled internally by the Paimon SDK + // inside IPaimonWriter::write(). + return local_state.sink(state, in_block, eos); } } // namespace doris diff --git a/be/src/exec/operator/paimon_table_sink_operator.h b/be/src/exec/operator/paimon_table_sink_operator.h index 579d42b2132..3ee0181129a 100644 --- a/be/src/exec/operator/paimon_table_sink_operator.h +++ b/be/src/exec/operator/paimon_table_sink_operator.h @@ -21,33 +21,26 @@ #include <memory> #include <string> -#include <unordered_map> #include "common/status.h" #include "core/block/block.h" #include "exec/operator/operator.h" -#include "exec/sink/writer/async_result_writer.h" -#include "exec/sink/writer/paimon/paimon_bucket.h" #include "exec/sink/writer/paimon/vpaimon_table_writer.h" #include "runtime/runtime_state.h" namespace doris { -/// Key for per-(partition, bucket) writer grouping. -/// partition_bytes is the serialized Paimon BinaryRow of partition values. -/// bucket is the bucket id. -using PaimonPartitionBucketKey = std::pair<std::string, int32_t>; - -/// Paimon table sink operator with per-(partition, bucket) routing. +/// Paimon table sink operator — simple pass-through to AsyncWriterSink. /// -/// Architecture: -/// sink_impl(block): -/// compute (partition_bytes, bucket) for each row -/// split block by (p,b) -/// for each (p,b): get_or_create_writer(p,b)->sink(sub_block) +/// A single VPaimonTableWriter (one AsyncResultWriter, one IO thread) +/// handles the entire table. Partition and bucket routing is performed +/// internally by the Paimon SDK (Java or Rust) inside +/// IPaimonWriter::write(). Doris does not compute partition values +/// or bucket ids; it passes complete Blocks to the SDK. /// -/// Each (p,b) gets its own VPaimonTableWriter (and hence its own -/// AsyncResultWriter + thread pool task + independent JNI/FFI object). +/// This mirrors Iceberg's approach: IcebergTableSinkOperatorX simply +/// delegates to AsyncWriterSink<VIcebergTableWriter>, and all +/// partition routing happens inside VIcebergTableWriter::write(). class PaimonTableSinkOperatorX; class PaimonTableSinkLocalState final @@ -99,40 +92,10 @@ private: requires(std::is_base_of_v<AsyncResultWriter, Writer>) friend class AsyncWriterSink; - /// Get or lazily create a writer for the given (partition, bucket). - Status _get_or_create_writer(RuntimeState* state, - const PaimonPartitionBucketKey& key, - VPaimonTableWriter** writer); - - /// Compute partition bytes and bucket ids for all rows in the block. - /// partition computation serializes partition cols to Paimon BinaryRow format. - Status _compute_routing(const Block& block, - std::vector<std::string>& partition_bytes, - std::vector<int32_t>& buckets); - - /// Split block by (partition_bytes, bucket) key. - std::unordered_map<PaimonPartitionBucketKey, std::unique_ptr<Block>> _split_by_key( - const Block& block, - const std::vector<std::string>& partition_bytes, - const std::vector<int32_t>& buckets); - const RowDescriptor& _row_desc; VExprContextSPtrs _output_vexpr_ctxs; const std::vector<TExpr>& _t_output_expr; ObjectPool* _pool = nullptr; - - // Per-(partition, bucket) writers. Lazily created. - std::unordered_map<PaimonPartitionBucketKey, - std::shared_ptr<VPaimonTableWriter>> _writers; - - // Bucket routing config from sink - std::vector<int> _bucket_key_indices; - std::vector<int> _partition_col_indices; - int32_t _total_buckets = 1; - bool _routing_enabled = false; - - // EOS has been sent to all writers - bool _eos_sent = false; }; } // namespace doris diff --git a/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp b/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp index 9760cbdadda..a11f3385a8a 100644 --- a/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp +++ b/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp @@ -26,8 +26,7 @@ Status FfiPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* s "The FFI backend will be introduced in v2 for high-throughput append-only writes."); } -Status FfiPaimonWriteBackend::create_writer(const std::string& partition_bytes, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) { +Status FfiPaimonWriteBackend::create_writer(std::unique_ptr<IPaimonWriter>* writer) { return Status::NotSupported("FFI backend: create_writer not yet implemented"); } diff --git a/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.h b/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.h index ee2b268b138..fe9cfd49e18 100644 --- a/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.h +++ b/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.h @@ -41,8 +41,7 @@ public: Status open(const TPaimonTableSink& sink, RuntimeState* state) override; - Status create_writer(const std::string& partition_bytes, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) override; + Status create_writer(std::unique_ptr<IPaimonWriter>* writer) override; Status create_committer(std::unique_ptr<IPaimonCommitter>* committer) override; diff --git a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp index a16fe5f2ee5..04116fffd09 100644 --- a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp @@ -196,15 +196,12 @@ Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* s return st; } -Status JniPaimonWriteBackend::create_writer(const std::string& partition_bytes, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) { +Status JniPaimonWriteBackend::create_writer(std::unique_ptr<IPaimonWriter>* writer) { DCHECK(_opened) << "Backend must be opened before creating writers"; + JNIEnv* env = nullptr; RETURN_IF_ERROR(_get_jni_env(&env)); - // For the JNI backend, each (partition, bucket) shares the same - // underlying Java BatchTableWrite instance. We create a lightweight - // wrapper that delegates to the shared Java object. auto jni_writer = std::make_unique<JniPaimonWriter>( env, _jni_writer_obj, _write_id, _prepare_commit_id, _abort_id, std::make_unique<ArrowMemoryPool<>>(), _sink); diff --git a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h index f3f47ce3db6..97dd06af703 100644 --- a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h @@ -34,8 +34,9 @@ class RuntimeState; /// /// Bridges Doris BE (C++) to the Paimon Java SDK via JNI. /// On open() it loads the Java class `PaimonJniWriter` and caches -/// method IDs. On create_writer() it instantiates a Java -/// BatchTableWrite for a given (partition, bucket) pair. +/// method IDs. On create_writer() it instantiates a wrapper that +/// delegates Block writes to the shared Java BatchTableWrite +/// instance; partition/bucket routing is handled by the Java SDK. /// /// Data path: /// Block → Arrow RecordBatch → IPC Stream → JNI direct buffer → @@ -51,8 +52,7 @@ public: Status open(const TPaimonTableSink& sink, RuntimeState* state) override; - Status create_writer(const std::string& partition_bytes, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) override; + Status create_writer(std::unique_ptr<IPaimonWriter>* writer) override; Status create_committer(std::unique_ptr<IPaimonCommitter>* committer) override; @@ -89,7 +89,9 @@ private: bool _opened = false; }; -/// Per-(partition, bucket) writer backed by the shared JNI context. +/// Writer backed by the shared JNI BatchTableWrite context. +/// Partition and bucket routing is handled by the Java Paimon SDK +/// internally during write(). class JniPaimonWriter final : public IPaimonWriter { public: JniPaimonWriter(JNIEnv* env, jobject jni_writer_obj, jmethodID write_id, diff --git a/be/src/exec/sink/writer/paimon/paimon_write_backend.h b/be/src/exec/sink/writer/paimon/paimon_write_backend.h index b46d4717309..595a355cba4 100644 --- a/be/src/exec/sink/writer/paimon/paimon_write_backend.h +++ b/be/src/exec/sink/writer/paimon/paimon_write_backend.h @@ -40,14 +40,25 @@ enum class PaimonBackendType { }; // ──────────────────────────────────────────────────────────── -// IPaimonWriter — per-(partition, bucket) writer +// IPaimonWriter — manages all (partition, bucket) pairs // ──────────────────────────────────────────────────────────── +// +// A single IPaimonWriter handles the entire table. Partition and +// bucket routing is performed internally by the Paimon SDK (Java +// or Rust), not by Doris. This is analogous to Iceberg's +// VIcebergPartitionWriter — Doris passes a full Block to write(), +// and the SDK routes rows to the correct file writers. class IPaimonWriter { public: virtual ~IPaimonWriter() = default; /// Write a Doris columnar block to this writer. + /// The block may contain rows belonging to different partitions and + /// buckets. Partition computation (BinaryRow partition), bucket + /// computation (hash % numBuckets), and routing to the correct + /// file writer (AppendOnlyWriter / KeyValueFileWriter) are all + /// handled internally by the Paimon SDK. virtual Status write(RuntimeState* state, Block& block) = 0; /// Prepare commit: close files and produce commit messages. @@ -110,11 +121,9 @@ public: /// Initialize the backend from the Thrift sink description. virtual Status open(const TPaimonTableSink& sink, RuntimeState* state) = 0; - /// Create a writer for a specific (partition, bucket) pair. - /// May be called multiple times for the same (partition, bucket); - /// each call produces an independent file writer. - virtual Status create_writer(const std::string& partition_bytes, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) = 0; + /// Create a writer for the table. One writer manages all partitions + /// and buckets; routing is handled internally by the Paimon SDK. + virtual Status create_writer(std::unique_ptr<IPaimonWriter>* writer) = 0; /// Create a committer. One per sink; shared across all writers. virtual Status create_committer(std::unique_ptr<IPaimonCommitter>* committer) = 0; diff --git a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp index 971c1d0c5c9..a217829eae6 100644 --- a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp +++ b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp @@ -40,15 +40,13 @@ Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) { _written_bytes_counter = ADD_COUNTER(_operator_profile, "WrittenBytes", TUnit::BYTES); _send_data_timer = ADD_TIMER(_operator_profile, "SendDataTime"); _project_timer = ADD_CHILD_TIMER(_operator_profile, "ProjectTime", "SendDataTime"); - _arrow_convert_timer = - ADD_CHILD_TIMER(_operator_profile, "ArrowConvertTime", "SendDataTime"); + _arrow_convert_timer = ADD_CHILD_TIMER(_operator_profile, "ArrowConvertTime", "SendDataTime"); _file_store_write_timer = ADD_CHILD_TIMER(_operator_profile, "FileStoreWriteTime", "SendDataTime"); _open_timer = ADD_TIMER(_operator_profile, "OpenTime"); _close_timer = ADD_TIMER(_operator_profile, "CloseTime"); _prepare_commit_timer = ADD_TIMER(_operator_profile, "PrepareCommitTime"); - _serialize_commit_messages_timer = - ADD_TIMER(_operator_profile, "SerializeCommitMessagesTime"); + _serialize_commit_messages_timer = ADD_TIMER(_operator_profile, "SerializeCommitMessagesTime"); _commit_payload_count = ADD_COUNTER(_operator_profile, "CommitPayloadCount", TUnit::UNIT); _commit_payload_bytes_counter = ADD_COUNTER(_operator_profile, "CommitPayloadBytes", TUnit::BYTES); @@ -56,12 +54,11 @@ Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) { SCOPED_TIMER(_open_timer); // Create backend via factory — one independent JNI/FFI object per writer - RETURN_IF_ERROR( - PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink, &_backend)); + RETURN_IF_ERROR(PaimonWriteBackendFactory::create(_t_sink.paimon_table_sink, &_backend)); RETURN_IF_ERROR(_backend->open(_t_sink.paimon_table_sink, state)); - // Create a single writer (partition_bytes + bucket set by the caller via sink config) - RETURN_IF_ERROR(_backend->create_writer("" /* partition_bytes */, 0 /* bucket */, &_writer)); + // Create a single writer — routing handled internally by Paimon SDK + RETURN_IF_ERROR(_backend->create_writer(&_writer)); LOG(INFO) << "VPaimonTableWriter opened: table=" << _t_sink.paimon_table_sink.tb_name << ", backend=" << static_cast<int>(_backend->type()); 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 4d710976d93..d4ce5b04bde 100644 --- a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h +++ b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h @@ -34,23 +34,29 @@ namespace doris { class ObjectPool; class RuntimeState; -/// VPaimonTableWriter is a per-(partition, bucket) writer. +/// VPaimonTableWriter is the single AsyncResultWriter for a Paimon table +/// sink. One writer = one IO thread, analogous to Iceberg's +/// VIcebergTableWriter. /// -/// Each instance serves exactly one (partition_bytes, bucket) pair. -/// Routing (partition + bucket computation and grouping) happens in -/// PaimonTableSinkOperatorX::sink_impl(), upstream of this writer. +/// Partition and bucket routing is delegated to the Paimon SDK (Java or +/// Rust) via IPaimonWriter::write(). Doris passes complete Blocks — the +/// SDK internally computes partition values, bucket ids, and routes rows +/// to the correct file writers. /// /// Architecture: /// PaimonTableSinkOperatorX -/// │ (route by partition,bucket) -/// ├── VPaimonTableWriter for (dt=01, bkt-0) -/// │ └── AsyncResultWriter → write() → IPaimonWriter → JNI/FFI -/// ├── VPaimonTableWriter for (dt=01, bkt-1) -/// │ └── AsyncResultWriter → write() → IPaimonWriter → JNI/FFI -/// └── ... +/// │ sink_impl() → AsyncWriterSink::sink() (no routing) +/// ▼ +/// VPaimonTableWriter (single AsyncResultWriter) +/// │ write() → IPaimonWriter::write() +/// │ → Paimon SDK internal routing +/// │ → AppendOnlyWriter / KeyValueFileWriter +/// │ → CompactManager (auto compaction) +/// ▼ +/// close() → prepareCommit() → CommitMessage[] /// /// Commit flow: -/// close() → writer->prepareCommit() +/// close() → writer->prepare_commit() /// → collect TPaimonCommitMessage[] /// → RuntimeState::add_paimon_commit_messages() /// → RPC to FE Coordinator @@ -76,7 +82,7 @@ private: // The backend abstraction — JNI or FFI, one per writer std::unique_ptr<IPaimonWriteBackend> _backend; - // Single writer (one per (partition, bucket)) + // Single writer managing all partitions and buckets (SDK-internal routing) std::unique_ptr<IPaimonWriter> _writer; // Statistics diff --git a/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md b/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md index 568372f2363..324a38704be 100644 --- a/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md +++ b/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md @@ -398,14 +398,18 @@ Paimon writer 的输出是**中间产物**：`CommitMessage` 不是简单的"文 **对 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？** @@ -455,7 +459,7 @@ be/src/format/jni/ ### 5.1 总体架构 -核心设计：**一个 `(partition, bucket)` = 一个独立的 AsyncResultWriter 实例**。路由在 Pipeline Operator 层完成，每个 writer 各自在 `fragment_mgr` 线程池中并行运行。 +核心设计：**一个 `VPaimonTableWriter` = 一个 `AsyncResultWriter` 实例 = 一个 IO 线程**（对标 Iceberg 的 `VIcebergTableWriter`）。分区路由和桶路由由 **Paimon SDK 内部完成**，Doris 侧不参与路由计算，直接将完整 Block 交给 Paimon SDK 写入。 ``` ┌─────────────────────────────────────────────────────────────────────┐ @@ -465,8 +469,8 @@ be/src/format/jni/ │ │ Planner │──▶│ (FE Planner) │──▶│ TPaimonTableSink │ │ │ └──────────────┘ └──────────────────┘ └─────────┬───────────┘ │ │ │ │ -│ ★ FE Shuffle: Exchange by hash(partition + bucket) │ │ -│ 保证同一 (partition,bucket) 的数据分发到同一个 BE │ │ +│ ★ 无特殊 Shuffle：正常 Exchange 分发数据到各 BE │ │ +│ 分区/桶路由由 Paimon SDK 在写入时内部处理 │ │ └───────────────────────────────────────────────────────┼─────────────┘ │ TDataSink ▼ @@ -476,35 +480,38 @@ be/src/format/jni/ │ ┌──────────────────────────────────────────────────────────────┐ │ │ │ Pipeline Operator: PaimonTableSinkOperatorX │ │ │ │ │ │ -│ │ sink_impl(block): │ │ -│ │ paimon_compute_partition(block) → partition bytes │ │ -│ │ paimon_compute_buckets(block) → bucket id │ │ -│ │ split by (partition_bytes, bucket) │ │ -│ │ for each ((p,b), sub_block): │ │ -│ │ sinks[(p,b)]->sink(sub_block) ← 路由到对应 writer │ │ +│ │ sink_impl(block, eos): │ │ +│ │ 直接调用 AsyncWriterSink::sink(block, eos) │ │ +│ │ ★ 无分区/桶路由逻辑，不做 Block 拆分 │ │ │ └──────────────────────┬───────────────────────────────────────┘ │ │ │ │ -│ ★ 每个 (partition,bucket) 对应一个独立的 AsyncWriterSink │ +│ ★ 仅一个 AsyncWriterSink<VPaimonTableWriter> │ │ │ │ -│ ┌───────────────┼───────────────┬───────────────┐ │ -│ ▼ ▼ ▼ ▼ │ -│ ┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ -│ │ AsyncRes │ │ AsyncRes │ │ AsyncRes │ │ AsyncRes │ │ -│ │ Writer │ │ Writer │ │ Writer │ │ Writer │ │ -│ │ (dt=01,b0) │ │ (dt=01,b1) │ │ (dt=02,b0) │ │ (dt=02,b1) │ │ -│ │ │ │ │ │ │ │ │ │ -│ │ process_ │ │ process_ │ │ process_ │ │ process_ │ │ -│ │ block() │ │ block() │ │ block() │ │ block() │ │ -│ │ write() │ │ write() │ │ write() │ │ write() │ │ -│ │ → JNI │ │ → JNI │ │ → JNI │ │ → JNI │ │ -│ └────────────┘ └────────────┘ └────────────┘ └────────────┘ │ -│ 各自在 fragment_mgr 线程池中并行 │ +│ ▼ │ +│ ┌────────────────────────────────────────────────────────────┐ │ +│ │ 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(partition_bytes, bucket) → IPaimonWriter │ │ -│ │ + createCommitter() → IPaimonCommitter │ │ +│ │ + createWriter() → IPaimonWriter (整个表共享一个 writer) │ │ +│ │ + createCommitter() → IPaimonCommitter │ │ │ │ │ │ │ │ ┌──────────────────────┐ ┌──────────────────────┐ │ │ │ │ │ JniPaimonWriteBackend│ │ FfiPaimonWriteBackend │ │ │ @@ -514,7 +521,7 @@ be/src/format/jni/ │ │ │ ┌──────────────────────────────────────────────────────────────┐ │ │ │ CommitMessage (统一格式，跨后端) │ │ -│ │ BE 各 writer 独立 produce → RuntimeState 汇总 → FE 统一提交 │ │ +│ │ 单 writer produce → RuntimeState 汇总 → RPC to FE 统一提交 │ │ │ └──────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────┘ ``` @@ -682,67 +689,74 @@ v2: 实现 Rust FFI Backend | Lookup Changelog Producer | JNI Backend | JNI Backend (Rust 不支持) | | MERGE INTO | JNI Backend | JNI Backend (Rust 不支持) | -#### 决策 4：Bucket Affinity — 分布式写入的文件膨胀控制 +#### 决策 4：Single Writer — 对标 Iceberg 的单 Writer 架构 -**问题**：多个 BE 同时写入同一个 `(partition, bucket)`，每个 BE 独立创建 writer，产生多个小文件。 +**问题**：v0 (初始实现) 采用了 "一个 `(partition, bucket)` = 一个 `AsyncResultWriter`" 的设计。每个 `AsyncResultWriter` 在 `fragment_mgr` 线程池中创建一个独立的 IO 线程。当 Paimon 表的 bucket 数量较大（例如 128 个 bucket × 10 个分区 = 1280 个 writer 实例），会导致线程数爆炸，严重浪费系统资源。 -Paimon Flink connector 通过上游 shuffle 解决：`keyBy(partition + bucket)` 保证相同 (partition, bucket) 的数据落在同一个 Subtask。但 Paimon SDK 本身**不做路由保证**——它假设上游已经做好了。 +**根因分析**：`AsyncResultWriter::start_writer()` 调用 `fragment_mgr()->get_thread_pool()->submit_func()` 为每个实例提交一个 long-running 的 `process_block()` 任务。这个任务在 writer 的整个生命周期中持续占用一个线程。当 writer 数量远超 CPU 核数时，大量线程空转等待数据，造成无效的上下文切换和内存开销。 -Doris 当前方案：接受同 bucket 多 writer 的情况，依赖 Paimon Compaction 兜底合并。 +**解决方案**：改为 **一个 `VPaimonTableWriter` = 一个 `AsyncResultWriter` 实例 = 一个 IO 线程**，对标 Iceberg 的 `VIcebergTableWriter`。分区路由和桶路由完全交给 Paimon SDK 内部处理。 -``` -当前 (无 shuffle): 长期 (FE shuffle): -═══════════════════════════ ═══════════════════════ +Doris 侧不再： +- 在 BE 端计算 Paimon partition bytes +- 在 BE 端计算 bucket id +- 在 Pipeline Operator 层做 Block 拆分和路由 +- 为每个 (partition, bucket) 创建独立的 writer -BE-1: bucket-0 → writer → file₁ Exchange shuffle by (partition, bucket) -BE-2: bucket-0 → writer → file₂ ┌──────┐ ┌──────┐ ┌──────┐ -BE-3: bucket-0 → writer → file₃ │ bkt-0│ │ bkt-2│ │bkt-1│ - │ bkt-3│ - 3 个小文件 └──────┘ └──────┘ └──────┘ - Compaction 合并 BE-1 BE-2 BE-3 +Doris 侧只负责： +- 将完整 Block 通过 `IPaimonWriter::write()` 交给 Paimon SDK +- Paimon SDK 内部完成分区/桶路由、文件写入、Compaction - 实际影响有限: 每个 bucket 全局只有一个 writer - - Bucket 数 >> BE 数时, → 一个文件 (到 target-file-size) - 每 bucket 数据量足以打满 - target-file-size - - Compaction 自动兜底 ``` +Doris Side Paimon SDK Side +═══════════ ════════════════ -**当前选择**：延迟到后续版本引入 FE shuffle。理由： -1. Bucket 数通常远大于 BE 数，每个 bucket 的数据量足够 -2. Paimon Compaction 机制成熟，文件最终会合并 -3. FE shuffle 需要改动 Planner 和 Exchange 算子，影响范围较大 -4. 当前 `IPaimonWriteBackend` 接口已支持多 writer，架构上不需要调整 +Block (完整批次) TableWrite.write(row × N) + │ │ + ├── Arrow IPC 转换 ├── getPartition(row) + │ ↓ ├── getBucket(row) + ├── JNI/FFI 传递 ├── writeBundle(p, b, records) + │ ↓ │ → AppendOnlyWriter + └── IPaimonWriter::write() │ → KeyValueFileWriter + │ → CompactManager + │ + └── prepareCommit() + → CommitMessage[] +``` -**升级路径**：v1 稳定后，在 `PhysicalPaimonTableSink` 之前插入 shuffle 算子，通过 `paimon_bucket.h` 的 hash 值做分发键。VPaimonTableWriter 和 IPaimonWriteBackend 接口零改动。 +**参考**：Iceberg 的 `VIcebergTableWriter` 采用相同架构（单一 `AsyncResultWriter` + 内部管理多个 `VIcebergPartitionWriter`），已在生产环境验证。 -#### 决策 5：One Writer per (Partition,Bucket) — 基于 AsyncResultWriter 的并发模型 +**架构优势**： +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 即可 -Doris 所有外部表 Sink 的写入都基于 `AsyncResultWriter`——每个实例 = 一个线程池任务 + 一个队列 + 一个消费者循环。实例之间完全独立，各自在 `fragment_mgr` 线程池中并行。 +**线程数对比**： -Paimon 中 bucket 在不同分区之间是独立命名空间，`(partition, bucket)` 是并发写入的基本单位。因此架构设计为：**一个 `(partition, bucket)` = 一个独立的 AsyncResultWriter 实例**。 +| 场景 | 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 线程 | -``` -FE: Exchange shuffle by (partition, bucket) - → 保证同一 (p,b) 的数据分发到同一个 BE +**潜在风险**： +- 单线程写入可能成为吞吐瓶颈 → Paimon SDK 内部使用异步 I/O (Rust) 或线程池 (Java)，单线程调用非瓶颈 +- 如需更高并发，可在 FE 层增加 `parallelism` 参数，创建多个 BE fragment 实例（每个持有独立 writer），这是标准的 Doris 水平扩展方式 -BE: Pipeline Operator 层路由 - → paimon_compute_partition() + paimon_compute_buckets() - → 按 (partition_bytes, bucket) 分组 - → sinks[(p,b)]->sink(sub_block) +#### 决策 5：Bucket Affinity — 不再由 Doris 保证 - 每个 (p,b) 对应: - └── AsyncWriterSink<VPaimonTableWriter> - └── AsyncResultWriter (独立线程池任务) - └── IPaimonWriter (独立 JNI/FFI 对象) -``` +**背景变化**：由于分区/桶路由已移交 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` 任务定期合并 -1. **复用 AsyncResultWriter 抽象**：不需要内部线程池，拆多个实例即可获得线程池并发 -2. **天然线程安全**：每个 writer 持有独立的 JNI/FFI 后端对象，无共享状态 -3. **隔离性好**：一个 `(p,b)` 的 I/O 阻塞不影响其他 `(p,b)` -4. **Bucket 和 Partition 计算都在 BE 端完成**：MurmurHash + Paimon BinaryRow 序列化，与 Paimon SDK 输出格式完全一致 +**长期优化（可选）**：如需减少小文件，可后续在 FE Planner 层引入 shuffle by hash(bucket_key)，但不影响当前架构设计。 ### 5.3 提交流程 @@ -804,30 +818,29 @@ Commit BE（可以是任意 BE 或专用 BE）： Doris Sink │ ▼ -FE: Exchange shuffle by (partition, bucket) - │ 相同 (p,b) → 同一 BE +BE: PaimonTableSinkOperatorX (Pipeline Operator) + │ sink_impl(block, eos) → 直接调用 AsyncWriterSink::sink() + │ ★ 无分区/桶路由，Block 原样传递 + │ + ▼ +VPaimonTableWriter (单一 AsyncResultWriter) + │ process_block() → write(block): + │ Block → Arrow IPC → IPaimonWriter::write() + │ ▼ -BE: PaimonTableSinkOperatorX (Pipeline Operator 层路由) +IPaimonWriter (Rust FFI preferred) + │ Paimon SDK 内部: + │ · 分区/桶计算 + │ · write_arrow_batch() → AppendOnlyWriter + │ · 文件写入 │ - ├── paimon_compute_partition(block) → partition bytes - ├── paimon_compute_buckets(block) → bucket id - ├── split by (partition_bytes, bucket) + ▼ +prepareCommit() → CommitMessage[] │ - └── for each ((p,b), sub_block): - │ - ▼ - VPaimonTableWriter (per (p,b) 的 AsyncResultWriter) - │ - ▼ - IPaimonWriter (Rust FFI preferred) - │ RecordBatch → write_arrow_batch() 零拷贝 - ▼ - prepareCommit() → CommitMessage[] - │ - ▼ (Thrift RPC) - FE Coordinator → PaimonTransaction.commit() - ▼ - New Snapshot ✓ + ▼ (Thrift RPC) +FE Coordinator → TableCommit.commit() + ▼ +New Snapshot ✓ ``` #### 场景 2：主键表写入 @@ -836,28 +849,29 @@ BE: PaimonTableSinkOperatorX (Pipeline Operator 层路由) Doris Sink │ ▼ -FE: Exchange shuffle by (partition, bucket) - │ 相同 (p,b) → 同一 BE +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 + │ ▼ -BE: PaimonTableSinkOperatorX (Pipeline Operator 层路由) - │ 同上 Parition/Bucket 计算 + 分组 +prepareCommit() → CommitMessage[] │ - └── for each ((p,b), sub_block): - │ - ▼ - VPaimonTableWriter (per (p,b) 的 AsyncResultWriter) - │ 使用 JNI Backend（需要 KeyValueFileWriter） - ▼ - JniPaimonWriter.write(Block) - │ Block → Arrow IPC → JNI → Java → BatchTableWrite.write() - │ Paimon 内部 CompactManager 自动处理 compaction - ▼ - prepareCommit() → CommitMessage[] - │ - ▼ (Thrift RPC) - FE Coordinator → PaimonTransaction.commit() - ▼ - New Snapshot ✓ + ▼ (Thrift RPC) +FE Coordinator → TableCommit.commit() + ▼ +New Snapshot ✓ ``` --- @@ -1009,13 +1023,17 @@ class IPaimonWriter { public: virtual ~IPaimonWriter() = default; - /// 写入一个 Doris Block + /// 写入一个 Doris Block（完整批次，可能包含多分区、多桶的数据） + /// Paimon SDK 内部完成：分区计算、桶计算、按 (partition,bucket) 路由、 + /// 文件写入（AppendOnlyWriter / KeyValueFileWriter）、Compaction 触发。 + /// /// @param state 运行时状态 - /// @param block 输入数据（Doris 列式格式） + /// @param block 输入数据（Doris 列式格式，未按分区/桶拆分） virtual Status write(RuntimeState* state, Block& block) = 0; - /// 准备提交：关闭文件，生成 CommitMessage - /// @param messages 输出：本次写入产生的 CommitMessage 列表 + /// 准备提交：关闭所有内部 writer，生成 CommitMessage + /// 对所有 (partition, bucket) 内部 writer 调用 prepareCommit() + /// @param messages 输出：本次写入产生的所有 CommitMessage 列表 virtual Status prepareCommit( std::vector<PaimonCommitMessage>& messages) = 0; @@ -1068,11 +1086,10 @@ public: virtual Status open(const TPaimonTableSink& sink, RuntimeState* state) = 0; - /// 为特定的 (partition, bucket) 创建一个新的 writer - /// 同一个 (partition, bucket) 可以被多次调用（每次创建一个新的文件 writer） + /// 为整个表创建一个 writer（管理所有 partition + bucket） + /// 分区/桶路由由 Paimon SDK 在 write() 内部处理 + /// @param writer 输出：创建的 writer virtual Status createWriter( - const std::string& partition_bytes, - int32_t bucket, std::unique_ptr<IPaimonWriter>* writer) = 0; /// 创建一个 committer（每个 sink 只需要一个） @@ -1123,18 +1140,23 @@ public: │ │ │ │ │ open() open() │ │ IPaimonWriteBackend IPaimonWriteBackend │ + │ → createWriter() → createWriter() │ + │ → IPaimonWriter → IPaimonWriter │ │ │ │ │ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ │ │ │ 写入阶段（并行） │ │ │ │ │ │ │ │ Block1[rows] │ │ │ ├─────────────────────────────────────────┤ │ - │ │ partition/bucket 路由 │ │ │ │ write() → IPaimonWriter │ │ + │ │ → Paimon SDK 内部: │ │ + │ │ · 分区计算 + 桶计算 │ │ + │ │ · writeBundle(p,b,records) │ │ + │ │ · Compaction 自动触发 │ │ │ │ │ │ │ │ Block2[rows] │ │ │ ├─────────────────────────────────────────┤ │ - │ │ 路由 → write() │ │ + │ │ write() → Paimon SDK │ │ │ │ │ │ │ │ ... (更多数据) ... │ │ │ │ │ │ @@ -1142,6 +1164,7 @@ public: │ ├─────────────────────────────────────────┤ │ │ │ prepareCommit() │ │ │ │ → CommitMessage[] │ │ + │ │ (所有 partition+bucket 的消息) │ │ │ │ │ │ │ │ prepareCommit() │ │ │ ├─────────────────────────────────────────┤ │ @@ -1345,7 +1368,7 @@ public class PaimonWriteCommitCoordinator { ### 9.1 VPaimonTableWriter -每个 `(partition, bucket)` 对应一个 `VPaimonTableWriter` 实例。路由（partition + bucket 计算与分组）在 Pipeline Operator 层完成，writer 只负责对已分组的数据进行写入。 +对标 Iceberg 的 `VIcebergTableWriter`：单一 `AsyncResultWriter` 实例，在 `write()` 中将完整 Block 交给 Paimon SDK 处理。分区/桶路由、文件写入、Compaction 全部由 Paimon SDK 内部完成。 ```cpp // be/src/exec/sink/writer/paimon/vpaimon_table_writer.h @@ -1365,38 +1388,152 @@ private: TDataSink _t_sink; RuntimeState* _state = nullptr; - // 抽象后端 — 每个 writer 持有一个独立的 IPaimonWriteBackend 实例 + // 抽象后端 — 整个 writer 持有一个 IPaimonWriteBackend 实例 std::unique_ptr<IPaimonWriteBackend> _backend; - // 当前 writer 对应的 (partition_bytes, bucket) - std::string _partition_bytes; - int32_t _bucket = 0; - - // 单个 IPaimonWriter（不再维护 writer map） + // 单一 IPaimonWriter（管理所有 partition + bucket，对标 Iceberg 的 + // _partitions_to_writers map，但路由逻辑在 Paimon SDK 内部） std::unique_ptr<IPaimonWriter> _writer; // 统计 - int64_t _rows_written = 0; - int64_t _bytes_written = 0; + 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; }; ``` -Pipeline Operator 层的路由逻辑： +**`write()` 实现逻辑**（伪代码）： ```cpp -// PaimonTableSinkOperatorX::sink_impl() -// 在数据进入各 writer 之前完成 (partition, bucket) 路由 +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)); + } -sink_impl(state, block, eos): - partition_bytes = paimon_compute_partition(block, partition_cols) - buckets = paimon_compute_buckets(block, bucket_key_cols, num_buckets) + // 2. 写入 Paimon SDK（分区/桶路由由 SDK 内部处理） + { + SCOPED_TIMER(_file_store_write_timer); + RETURN_IF_ERROR(_writer->write(state, output_block)); + } - // 按 (partition_bytes, bucket) 分组 - for each ((p,b), sub_block): - get_or_create_sink(p, b)->sink(sub_block, eos) - // 每个 (p,b) 对应一个独立的 AsyncWriterSink<VPaimonTableWriter> + 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 @@ -1406,8 +1543,8 @@ class JniPaimonWriteBackend : public IPaimonWriteBackend { public: Status open(const TPaimonTableSink& sink, RuntimeState* state) override; - Status createWriter(const std::string& partition, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) override; + // 创建 writer——管理所有 partition+bucket，SDK 内部路由 + Status createWriter(std::unique_ptr<IPaimonWriter>* writer) override; Status createCommitter( std::unique_ptr<IPaimonCommitter>* committer) override; @@ -1432,6 +1569,22 @@ private: 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 实现 @@ -1443,8 +1596,7 @@ class FfiPaimonWriteBackend : public IPaimonWriteBackend { public: Status open(const TPaimonTableSink& sink, RuntimeState* state) override; - Status createWriter(const std::string& partition, int32_t bucket, - std::unique_ptr<IPaimonWriter>* writer) override; + Status createWriter(std::unique_ptr<IPaimonWriter>* writer) override; Status createCommitter( std::unique_ptr<IPaimonCommitter>* committer) override; @@ -1520,7 +1672,7 @@ Phase 1 (v1) — Java JNI Backend + 抽象层： ✅ JniPaimonWriteBackend（完整实现） ✅ FfiPaimonWriteBackend（stub，所有方法返回 NotSupported） ✅ 结构化 Thrift CommitMessage - ✅ BE 端 partition/bucket 路由 + ✅ 单 Writer 架构（对标 Iceberg），Paimon SDK 内部路由 ✅ Append-only + 固定桶 PK 写入 ✅ INSERT OVERWRITE / Truncate ✅ ChangelogProducer: None / Input @@ -1574,8 +1726,7 @@ Phase 4（Rust 功能追赶）： 3. **BE 端 — JNI 后端实现** - **`JniPaimonWriteBackend`**：实现 `IPaimonWriteBackend` - **`VPaimonTableWriter`**：继承 `AsyncResultWriter`，通过 `IPaimonWriteBackend` 接口调用（不直接依赖 JNI） - - Block → Arrow → Java 的数据桥接 - - partition/bucket 路由逻辑 + - Block → Arrow → Java 的数据桥接（Paimon SDK 内部处理分区/桶路由） 4. **BE 端 — FFI 后端预留** - **`FfiPaimonWriteBackend`**：stub 实现，所有方法返回 `Status::NotSupported("FFI backend not yet implemented")` @@ -1706,8 +1857,8 @@ Phase 4（Rust 功能追赶）： ``` Doris 概念 Paimon 概念 说明 ───────────────────────────────────────────────────── -VPaimonTableWriter BatchWriteBuilder 整个表的写入器 -VPaimonPartitionWriter BatchTableWrite 单个 (partition, bucket) 的写入器 +VPaimonTableWriter BatchWriteBuilder 整个表的写入器（单实例） +IPaimonWriter BatchTableWrite 管理所有分区/桶的内部路由 Block (batch) InternalRow / RecordBatch 数据批次 prepareCommit() prepareCommit() 生成 CommitMessage close() + RPC TableCommit.commit() 提交 @@ -1740,18 +1891,17 @@ close() + RPC TableCommit.commit() 提交 | 功能 | 矩阵优先级 | 实现方式 | |------|----------|----------| -| Append-only 表写入 | P0 | JNI backend → Paimon BatchTableWrite | +| Append-only 表写入 | P0 | JNI backend → Paimon BatchTableWrite (SDK 内部路由) | | Primary-key 表写入（full row） | P1 | JNI backend → KeyValueFileWriter，完整 row | -| Fixed bucket 写入 | P0/P1 | BE 端 murmur hash 计算 bucket，per-bucket writer 并发写 | -| Fixed bucket BE 端路由 | **P0** | VPaimonTableWriter 按 bucket 拆分 Block，各自独立 writer | -| `INSERT OVERWRITE` | P1 | Thrift 下发 write_mode + static_partition，Java 端 `withOverwrite()` / `withIgnorePreviousFiles()` | -| Unaware bucket 写入 | P0 | 单 writer 路由，Paimon 内部处理 | -| Partition 表写入 | P0 | FE 下发 partition columns，Paimon 内部路由 | -| 非 partition 表写入 | P0 | 随机并发写 | +| 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 完整支持 | -| Commit message 序列化 | P0 | DPCM 自定义协议（CommitMessageSerializer） | -| Stream commit 幂等 | P0 | `StreamTableCommit.filterAndCommit()` + commitUser + txnId | -| Abort 未提交文件 | P0 | `PaimonTransaction.rollback()` → `committer.abort()` | +| `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 | @@ -1759,15 +1909,16 @@ close() + RPC TableCommit.commit() 提交 | 并发 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 | PaimonJniWriter 已实现转换逻辑 | 缺少专项测试覆盖 | -| 小文件控制 | P1 | batch buffer 合并小批次 | buffer 阈值需根据实际场景调优 | -| Partition 规范化 | P0 | 依赖 Paimon SDK 内部处理 | 需验证 Doris 列值与 Paimon partition extractor 一致性 | +| 复杂类型（ARRAY/MAP/STRUCT） | P1 | JNI writer 已实现转换逻辑 | 缺少专项测试覆盖 | +| 小文件控制 | P1 | Paimon SDK 内部按 target-file-size 控制 | SDK 行为，Doris 侧无需干预 | +| Partition 规范化 | P0 | 依赖 Paimon SDK 内部处理 | 与 Paimon 原生语义完全一致 | ### 12.3 未覆盖 — v2 计划（❌） @@ -1797,8 +1948,11 @@ close() + RPC TableCommit.commit() 提交 当前实现覆盖了矩阵中 **P0 项目的 100%**，P1 项目的 **~80%**。 -P0 项目中"去 gather 分发"、"Fixed bucket BE 端计算"、"INSERT OVERWRITE" 三项 -已补齐，BE 端通过 Paimon 兼容的 murmur hash 实现 per-bucket 路由分发。 +与 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) 等，按设计文档 @@ -1972,7 +2126,14 @@ tools/paimon_perf/ --- -> **文档版本**：v7.0 +> **文档版本**：v8.0 > **作者**：Doris Paimon Write Team -> **日期**：2026-07-09 -> **状态**：架构升级：One Writer per (Partition,Bucket) + AsyncResultWriter 线程池并发 + BE 端 partition+bucket 双计算。含测试计划。 +> **日期**：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 ae7c955dcaf..dbb5ead0dc5 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 @@ -58,13 +58,11 @@ import org.apache.paimon.disk.IOManagerImpl; import org.apache.paimon.io.DataOutputSerializer; import org.apache.paimon.memory.HeapMemorySegmentPool; import org.apache.paimon.options.Options; -import org.apache.paimon.table.BucketMode; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; import org.apache.paimon.table.sink.BatchTableWrite; import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.CommitMessageSerializer; -import org.apache.paimon.table.sink.RowKeyExtractor; import org.apache.paimon.table.sink.TableWriteImpl; import org.apache.paimon.types.ArrayType; import org.apache.paimon.types.BinaryType; @@ -101,7 +99,7 @@ import java.util.stream.Collectors; * → PaimonJniWriter.write(address, length) * → ArrowStreamReader → VectorSchemaRoot * → convert row-by-row → Paimon GenericRow - * → BatchTableWrite.write(row) / write(row, bucket) + * → BatchTableWrite.write(row) [SDK handles partition+bucket routing] * * Commit path: * @@ -125,7 +123,6 @@ public class PaimonJniWriter { private BatchTableWrite writer; private TableWriteImpl<?> tableWrite; - private RowKeyExtractor rowKeyExtractor; private final BufferAllocator allocator; private final CommitMessageSerializer serializer = new CommitMessageSerializer(); @@ -136,10 +133,7 @@ public class PaimonJniWriter { private Map<String, DataField> paimonFieldMap; private DataType[] targetTypes; - private BucketMode bucketMode; - private boolean useExplicitBucketWrite; private long commitIdentifier; - private int totalBuckets; private IOManager ioManager; private HeapMemorySegmentPool memorySegmentPool; @@ -340,7 +334,6 @@ public class PaimonJniWriter { FileStoreTable fileStoreTable = (FileStoreTable) table; CoreOptions coreOptions = CoreOptions.fromMap(fileStoreTable.options()); - this.totalBuckets = coreOptions.bucket(); String commitUser = options.get(KEY_COMMIT_USER); if (commitUser == null || commitUser.isEmpty()) { @@ -352,7 +345,6 @@ public class PaimonJniWriter { this.tableWrite.withIgnorePreviousFiles(true); } this.writer = this.tableWrite; - initBucketMode(fileStoreTable); initSpillIfNeeded(coreOptions, options); } @@ -372,21 +364,6 @@ public class PaimonJniWriter { return result; } - private void initBucketMode(FileStoreTable fileStoreTable) { - this.bucketMode = fileStoreTable.bucketMode(); - if (bucketMode == BucketMode.HASH_DYNAMIC - || bucketMode == BucketMode.KEY_DYNAMIC - || bucketMode == BucketMode.POSTPONE_MODE) { - throw new UnsupportedOperationException( - "Unsupported Paimon bucket mode for write: " + bucketMode - + ". Only HASH_FIXED and BUCKET_UNAWARE are supported."); - } - this.useExplicitBucketWrite = bucketMode == BucketMode.HASH_FIXED; - if (useExplicitBucketWrite) { - this.rowKeyExtractor = fileStoreTable.createRowKeyExtractor(); - } - } - private void initSpillIfNeeded(CoreOptions coreOptions, Map<String, String> options) throws Exception { if (!coreOptions.writeBufferSpillable()) { @@ -433,12 +410,9 @@ public class PaimonJniWriter { } } try { - if (useExplicitBucketWrite) { - rowKeyExtractor.setRecord(reusedRow); - writer.write(reusedRow, rowKeyExtractor.bucket()); - } else { - writer.write(reusedRow); - } + // 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); @@ -752,8 +726,6 @@ public class PaimonJniWriter { writer = null; } tableWrite = null; - rowKeyExtractor = null; - useExplicitBucketWrite = false; if (ioManager != null) { ioManager.close(); ioManager = null; @@ -845,9 +817,7 @@ public class PaimonJniWriter { @SuppressWarnings("unused") private String currentState() { return "location=" + tableLocation - + ", writerNull=" + (writer == null) - + ", bucketMode=" + bucketMode - + ", explicitBucket=" + useExplicitBucketWrite; + + ", writerNull=" + (writer == null); } /** InputStream over a direct ByteBuffer (no copy). */ --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
