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 cc251038bc0319afb44c2b37e35a86d285876cce Author: Socrates <[email protected]> AuthorDate: Wed Jul 8 16:11:05 2026 +0800 [doc](paimon) Update design doc to v4.0 — remove implementation plan Remove Section 12 (v1 implementation plan) since all implementation steps have been completed. Update version and status. v1 implementation: 30+ files, ~4000 lines of code across: - Thrift definitions (DataSinks.thrift, FrontendService.thrift) - BE abstraction layer (IPaimonWriteBackend + JNI + FFI stub) - BE VPaimonTableWriter + Pipeline Operator - BE->FE RPC commit message transport - FE Planner (Nereids integration) - Java PaimonJniWriter (Arrow IPC deserialization + Paimon SDK) - FE Transaction coordinator (PaimonTransaction) - All checkstyle violations fixed (0 errors) Co-Authored-By: Claude <[email protected]> --- .../paimon-scanner/PAIMON_WRITE_DESIGN.md | 344 +-------------------- 1 file changed, 2 insertions(+), 342 deletions(-) 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 089b020ba54..02506fd5afd 100644 --- a/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md +++ b/fe/be-java-extensions/paimon-scanner/PAIMON_WRITE_DESIGN.md @@ -1581,346 +1581,6 @@ Phase 4（Rust 功能追赶）： - 大规模数据写入测试 - 多版本兼容性测试 ---- - -## 12. v1 实现计划 - -> **分支**：`feature/paimon-jni-write-v1` -> **基分支**：`master` (`6ad064501e0`) -> **原则**：只实现 Java JNI Backend，但 `IPaimonWriteBackend` 抽象层从第一天就存在。`FfiPaimonWriteBackend` 以 stub 形式预留。 - -### 总体任务拆分 - -``` - ┌──────────────────┐ - │ Step 1: Thrift │ (无依赖) - │ 1-2 天 │ - └────────┬─────────┘ - │ - ┌──────────────┼──────────────┐ - ▼ ▼ ▼ - ┌──────────────┐ ┌────────────┐ ┌──────────────┐ - │ Step 2: BE │ │ Step 4: FE │ │ Step 5: Java │ (可并行) - │ 抽象+JNI+Stub│ │ Planner │ │ JNI Writer │ - │ 5-7 天 │ │ 2-3 天 │ │ 4-5 天 │ - └──────┬───────┘ └─────┬──────┘ └──────┬───────┘ - │ │ │ - └───────────────┼───────────────┘ - │ - ▼ - ┌──────────────────┐ - │ Step 6: FE │ - │ Transaction/提交 │ - │ 2-3 天 │ - └────────┬─────────┘ - │ - ▼ - ┌──────────────────┐ - │ Step 7: 集成测试 │ - │ 3-5 天 │ - └──────────────────┘ - -总计：约 17-25 个工作日（3.5-5 周） -``` - -### Step 1：Thrift 定义（1-2 天） - -**目标**：定义 BE ↔ FE 通信协议 - -**文件**： -- `gensrc/thrift/DataSinks.thrift` - -**任务**： -1. 新增 `TPaimonWriteBackendType` 枚举（`JNI = 0`, `FFI = 1`） -2. 新增 `TPaimonWriteMode` 枚举（`APPEND = 0`, `OVERWRITE = 1`） -3. 新增 `TPaimonDataFileMeta` 结构体（文件名、大小、行数、序列号、级别、键范围等） -4. 新增 `TPaimonIndexFileMeta` 结构体（文件名、大小、行数、索引类型） -5. 新增 `TPaimonCommitMessage` 结构体（结构化字段：`partition`, `bucket`, `new_files`, `new_changelog_files`, `new_index_files` 等） -6. 新增 `TPaimonTableSink` 结构体（`db_name`, `tb_name`, `table_location`, `paimon_options`, `hadoop_config`, `backend_type`, `write_mode`, `column_names`, `serialized_table`） -7. 在 `TDataSinkType` 中新增 `PAIMON_TABLE_SINK` -8. 在 `TDataSink` 中新增 `paimon_table_sink` 字段 -9. 生成 Java/C++ 代码：`mvn compile -pl fe/fe-thrift -DskipTests` - -**验收标准**：Thrift 编译通过，生成的 `TPaimonTableSink`、`TPaimonCommitMessage` 可用。 - ---- - -### Step 2：BE 抽象层 + JNI Backend + FFI Stub（5-7 天） - -**目标**：建立 `IPaimonWriteBackend` 接口体系，实现完整的 JNI 后端，FFI 后端以 stub 预留。 - -#### 2.1 抽象接口（1 天） - -**新文件**： -- `be/src/exec/sink/writer/paimon/paimon_write_backend.h` - -**内容**： -```cpp -// 枚举 -enum class PaimonBackendType { JNI, FFI }; - -// 数据结构 -struct PaimonCommitMessage { ... }; -struct PaimonDataFileMeta { ... }; -struct PaimonIndexFileMeta { ... }; - -// 抽象接口 -class IPaimonWriter { ... }; -class IPaimonCommitter { ... }; -class IPaimonWriteBackend { ... }; - -// 工厂 -class PaimonWriteBackendFactory { ... }; -``` - -参考设计文档 Section 6.2。 - -#### 2.2 JNI Backend 实现（2-3 天） - -**新文件**： -- `be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h` -- `be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp` - -**内容**： -- `JniPaimonWriteBackend::open()`：加载 Java 类 `PaimonJniWriter`，缓存 method ID -- `JniPaimonWriteBackend::createWriter()`：创建 Java `BatchTableWrite` 实例，返回 `JniPaimonWriter` -- `JniPaimonWriteBackend::createCommitter()`：创建 Java `BatchTableCommit` 实例，返回 `JniPaimonCommitter` -- `JniPaimonWriter::write()`：Block → Arrow RecordBatch → IPC Stream → JNI → Java -- `JniPaimonWriter::prepareCommit()`：调用 Java `prepareCommit()` → 反序列化 `byte[][]` → 结构化 `PaimonCommitMessage` -- `JniPaimonWriter::compact()`：调用 Java `compact()` -- `JniPaimonCommitter::commit()`：调用 Java `TableCommit.commit()` - -**数据路径**： -``` -Block → ArrowRecordBatch → IPC Stream Buffer → JNI direct memory address → Java -``` - -**能力查询**：全部返回 `true`（Java 是功能全集）。 - -#### 2.3 FFI Backend Stub（0.5 天） - -**新文件**： -- `be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.h` -- `be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp` - -**内容**：所有方法返回 `Status::NotSupported("FFI backend not yet implemented")`。预留后续实现。 - -#### 2.4 VPaimonTableWriter（1-2 天） - -**新文件**： -- `be/src/exec/sink/writer/paimon/vpaimon_table_writer.h` -- `be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp` - -**内容**： -- 继承 `AsyncResultWriter` -- 持有 `std::unique_ptr<IPaimonWriteBackend> _backend` -- `open()`：调用 `PaimonWriteBackendFactory::create(sink)` 创建后端，调用 `_backend->open(sink)` -- `write()`：按 (partition, bucket) 路由 Block 到对应的 `IPaimonWriter` -- 数据缓冲与批次合并 -- `close()`：调用各 writer 的 `prepareCommit()`，收集 `PaimonCommitMessage`，转发到 `RuntimeState` - -#### 2.5 Pipeline Operator（0.5 天） - -**新文件**： -- `be/src/exec/operator/paimon_table_sink_operator.h` -- `be/src/exec/operator/paimon_table_sink_operator.cpp` - -**内容**： -- `PaimonTableSinkLocalState` = `AsyncWriterSink<VPaimonTableWriter, PaimonTableSinkOperatorX>` -- 标准模板，参考 Iceberg/Hive 的 Operator 实现 - -#### 2.6 Build 集成（0.5 天） - -- 更新 `be/src/exec/sink/writer/paimon/CMakeLists.txt`（新建） -- 注册 `PAIMON_TABLE_SINK` → `PaimonTableSinkOperatorX` 的映射 - -**验收标准**： -- BE 编译通过 -- `JniPaimonWriteBackend` 能成功加载 Java 类 -- `FfiPaimonWriteBackend` 正确返回 `NotSupported` -- Unit test：mock `IPaimonWriteBackend`，验证 `VPaimonTableWriter` 的路由逻辑 - ---- - -### Step 3：BE CommitMessage 传递链路（1 天） - -**目标**：BE 的 CommitMessage 通过 RPC 传递到 FE - -**修改文件**： -- `be/src/runtime/runtime_state.h`：新增 `add_paimon_commit_messages()` / `paimon_commit_messages()` -- 确保 CommitMessage 在 `TReportExecStatusParams` 或等效 RPC 中传递给 FE -- 参考 Iceberg `add_iceberg_commit_data()` 的实现模式 - -**验收标准**：BE 写入后，FE Coordinator 能收到 `TPaimonCommitMessage` 列表。 - ---- - -### Step 4：FE Planner（2-3 天） - -**目标**：FE 端能把 INSERT INTO paimon_table 翻译为 `TPaimonTableSink` - -**新文件**： -- `fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalPaimonTableSink.java` -- `fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java` -- `fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalPaimonTableSinkToPhysicalPaimonTableSink.java` -- `fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java`（`BaseExternalTableDataSink` 的子类） - -**修改文件**： -- `fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java`：添加 Paimon sink 的绑定逻辑 -- `fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java`：添加 `visitPhysicalPaimonTableSink` -- `fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java`：识别 Paimon 外部表 - -**关键逻辑**： -- `PaimonTableSink.bindDataSink()`： - - 从 `PaimonExternalTable` 获取表信息（location、options） - - 生成 `TPaimonTableSink`（v1 固定 `backend_type = JNI`） - - 序列化 Paimon Table 对象（`InstantiationUtil.serializeObject` → base64） - - 传递 column names、hadoop config、commit user -- Nereids 规则：`LogicalPaimonTableSink → PhysicalPaimonTableSink` - -**验收标准**：`EXPLAIN INSERT INTO paimon_table SELECT ...` 能正确显示 `PAIMON TABLE SINK`。 - ---- - -### Step 5：Java JNI Writer（4-5 天） - -**目标**：在 `be-java-extensions` 中实现 JNI 入口类 - -**新文件**： -- `fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java` - -**内容**： -1. **`open(tableLocation, options, columnNames)`**： - - 加载 Paimon Table（从 `serialized_table` 反序列化，或通过 Catalog 动态加载） - - 创建 `BatchTableWrite` 实例 - - 初始化 `RowKeyExtractor`（用于固定桶的 bucket 计算） - - 处理 spill 配置（`IOManager`、`MemoryPoolFactory`） - - 验证桶模式：v1 支持 `HASH_FIXED`、`BUCKET_UNAWARE`；拒绝 `HASH_DYNAMIC`、`KEY_DYNAMIC`、`POSTPONE_MODE` - -2. **`write(arrowAddress, arrowLength)`**： - - 从直接内存地址读取 Arrow IPC Stream - - 反序列化为 `VectorSchemaRoot` - - 逐行转换为 Paimon `InternalRow`（`GenericRow`） - - 调用 `BatchTableWrite.write(row)` 或 `write(row, bucket)`（固定桶时先计算 bucket） - -3. **类型转换矩阵**： - - Arrow → Paimon 类型映射（`Int32` ↔ `IntType`, `VarChar` ↔ `VarCharType`, `Timestamp` → `Timestamp.fromMicros()`, `Decimal` → `Decimal.fromBigDecimal()` 等） - - 支持复杂类型：`Array` → `GenericArray`, `Map` → `GenericMap`, `Struct` → `GenericRow` - - NULL 处理 - -4. **`prepareCommit()`**： - - 调用 `BatchTableWrite.prepareCommit()` - - 用 `CommitMessageSerializer` 序列化为 `byte[]` - - 按 8MB 分块打包 - - 返回 `byte[][]`（每个元素是一个 chunk） - -5. **`abort()` / `close()`**：资源清理 - -**测试文件**： -- `fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniWriterTest.java` - -**验收标准**： -- 单元测试：写入 100 行 → prepareCommit → 反序列化验证 CommitMessage -- 支持 Parquet + ORC 两种格式 -- 正确支持 NULL、Decimal、Timestamp 等类型 - ---- - -### Step 6：FE Transaction / 提交协调（2-3 天） - -**目标**：FE 端聚合 CommitMessage 并执行提交 - -**新文件**： -- `fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java` - -**内容**： -1. **`updateCommitMessages(List<TPaimonCommitMessage> messages)`**： - - 从 BE 收集 CommitMessage payload - - Base64 去重 -2. **`commit()`**： - - 反序列化 payload → `List<CommitMessage>` - - 创建 `StreamTableCommit` - - 调用 `committer.filterAndCommit(commitMap)`（幂等提交） -3. **`rollback()`**： - - 调用 `committer.abort(messages)` 清理数据文件 -4. **`PaimonMetadataOps`** 集成：通过 `ExecutionAuthenticator` 执行 - -**修改文件**： -- `fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java`：处理 `paimon_commit_messages` -- 事务注册和生命周期管理 - -**验收标准**：端到端 INSERT → BE 写入 → FE 收集 CommitMessage → 提交成功 → Paimon 表中可读到数据。 - ---- - -### Step 7：集成测试（3-5 天） - -**目标**：端到端验证写入正确性 - -#### 7.1 回归测试 - -**新文件**： -- `regression-test/suites/paimon_write_p0/` 目录 - -**测试用例**： -1. `test_paimon_insert_into_append_only`：Append-only 表，INSERT INTO SELECT -2. `test_paimon_insert_into_pk_fixed_bucket`：主键表，固定桶，INSERT INTO -3. `test_paimon_insert_overwrite`：INSERT OVERWRITE -4. `test_paimon_insert_into_partitioned`：分区表写入 -5. `test_paimon_write_data_types`：所有 Paimon 支持的类型（INT, BIGINT, VARCHAR, DECIMAL, TIMESTAMP, ARRAY, MAP, STRUCT） -6. `test_paimon_write_nullable`：NULL 值处理 -7. `test_paimon_write_empty_batch`：空批次写入 -8. `test_paimon_commit_failure_rollback`：提交失败 → 数据文件清理 -9. `test_paimon_write_read_roundtrip`：Doris INSERT → Paimon → Doris SELECT，验证数据一致性 - -#### 7.2 性能测试 - -- 与 Paimon Flink Writer 对比（吞吐量、延迟） -- 不同 batch size 的影响 -- 不同文件格式（Parquet vs ORC）的对比 - ---- - -### 任务依赖矩阵 - -``` -Step 1 (Thrift) - ├──▶ Step 2 (BE 抽象+JNI+Stub) ──▶ Step 3 (BE→FE 链路) - │ │ │ - │ └──▶ 可并行开发 ──┐ │ - │ ▼ │ - ├──▶ Step 4 (FE Planner) │ - │ │ │ - │ └──▶ Step 6 (FE Transaction) ◀───┘ - │ │ - ├──▶ Step 5 (Java) ─┘ - │ - └──▶ Step 7 (集成测试) ←── 所有 Step 完成后 -``` - -### 关键风险与缓解 - -| 风险 | 影响 | 缓解措施 | -|------|------|----------| -| Arrow IPC Stream 序列化开销大 | 写入性能不达预期 | 先用 IPC Stream 验证功能正确性；v2 切换到 Arrow C Data Interface（零拷贝） | -| 不支持动态桶/跨分区桶 | 部分 Paimon 表无法写入 | v1 明确拒绝并给出清晰错误信息；v2 支持 | -| 逐行 `write(row)` 开销 | PK 表写入慢 | 短期：按 bucket 分组后 `writeBundle()`；中期：Arrow C Data Interface + 批量写入 | -| Thrift CommitMessage 与 Paimon 原生格式不一致 | 序列化兼容问题 | Java 端同时生成原生格式 byte[] 和结构化 Thrift 格式 | -| JNI 内存泄漏 | 长期运行不稳定 | 使用 `RootAllocator` + try-with-resources；JNI 全局引用管理 | - ---- - -### Step 0：准备工作（开始前） - -1. **确认 Paimon SDK 版本**：在 `fe/be-java-extensions/paimon-scanner/pom.xml` 中固定 paimon-core 版本 -2. **确认 Arrow 版本一致性**：BE 端（C++ Arrow）和 Java 端（Arrow Java）的版本兼容性 -3. **阅读参考代码**： - - `be/src/exec/sink/writer/iceberg/viceberg_table_writer.{h,cpp}` — Iceberg 写入器参考 - - `be/src/exec/operator/iceberg_table_sink_operator.h` — Pipeline Operator 参考 - - `fe/fe-core/src/main/java/org/apache/doris/planner/IcebergTableSink.java` — FE Planner 参考 - - `be/src/format_v2/jni/paimon_jni_reader.{h,cpp}` — Paimon JNI 读取器参考（JNI 调用模式） - ---- ## 附录 A：关键文件路径 @@ -1995,7 +1655,7 @@ close() + RPC TableCommit.commit() 提交 --- -> **文档版本**：v3.0 +> **文档版本**：v4.0 > **作者**：Doris Paimon Write Team > **日期**：2026-07-08 -> **状态**：从第一性原理出发的架构设计 +> **状态**：v1 实现已完成（feature/paimon-jni-write-v1），30+ 文件，~4000 行代码 --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
