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 b57034f9ff928c48e28edc61334548a7b8efad0a Author: Socrates <[email protected]> AuthorDate: Wed Jul 8 15:13:06 2026 +0800 [feature](paimon) Step 2: BE abstraction layer + JNI backend + FFI stub - Add IPaimonWriteBackend/IPaimonWriter/IPaimonCommitter interfaces - Implement JniPaimonWriteBackend (full JNI bridge with Arrow IPC) - Add FfiPaimonWriteBackend stub (returns NotSupported for all ops) - Add VPaimonTableWriter (AsyncResultWriter, buffer management) - Register PaimonTableSinkOperatorX in pipeline_fragment_context - Add RuntimeState::add_paimon_commit_messages() for commit coordination The abstraction layer ensures Rust FFI backend can be added in v2 without changing upper-layer code (VPaimonTableWriter, Pipeline Operator). Co-Authored-By: Claude <[email protected]> --- .../exec/operator/paimon_table_sink_operator.cpp | 31 ++ be/src/exec/operator/paimon_table_sink_operator.h | 86 ++++ be/src/exec/pipeline/pipeline_fragment_context.cpp | 9 + .../writer/paimon/ffi_paimon_write_backend.cpp | 39 ++ .../sink/writer/paimon/ffi_paimon_write_backend.h | 59 +++ .../writer/paimon/jni_paimon_write_backend.cpp | 446 +++++++++++++++++++++ .../sink/writer/paimon/jni_paimon_write_backend.h | 140 +++++++ .../exec/sink/writer/paimon/paimon_write_backend.h | 151 +++++++ .../writer/paimon/paimon_write_backend_factory.cpp | 65 +++ .../sink/writer/paimon/vpaimon_table_writer.cpp | 253 ++++++++++++ .../exec/sink/writer/paimon/vpaimon_table_writer.h | 125 ++++++ be/src/runtime/runtime_state.h | 15 + 12 files changed, 1419 insertions(+) diff --git a/be/src/exec/operator/paimon_table_sink_operator.cpp b/be/src/exec/operator/paimon_table_sink_operator.cpp new file mode 100644 index 00000000000..9c3ae8a19aa --- /dev/null +++ b/be/src/exec/operator/paimon_table_sink_operator.cpp @@ -0,0 +1,31 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "exec/operator/paimon_table_sink_operator.h" + +#include "pipeline/exec/operator.h" + +namespace doris { + +Status PaimonTableSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) { + RETURN_IF_ERROR(Base::init(state, info)); + _writer = std::make_shared<VPaimonTableWriter>( + info.tsink, _output_vexpr_ctxs, _dependency, _finish_dependency); + return Status::OK(); +} + +} // namespace doris diff --git a/be/src/exec/operator/paimon_table_sink_operator.h b/be/src/exec/operator/paimon_table_sink_operator.h new file mode 100644 index 00000000000..4de0fa74d3b --- /dev/null +++ b/be/src/exec/operator/paimon_table_sink_operator.h @@ -0,0 +1,86 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include "exec/operator/operator.h" +#include "exec/sink/writer/paimon/vpaimon_table_writer.h" + +namespace doris { + +class PaimonTableSinkOperatorX; + +class PaimonTableSinkLocalState final + : public AsyncWriterSink<VPaimonTableWriter, PaimonTableSinkOperatorX> { +public: + using Base = AsyncWriterSink<VPaimonTableWriter, PaimonTableSinkOperatorX>; + using Parent = PaimonTableSinkOperatorX; + ENABLE_FACTORY_CREATOR(PaimonTableSinkLocalState); + PaimonTableSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state) + : Base(parent, state) {} + Status init(RuntimeState* state, LocalSinkStateInfo& info) override; + Status open(RuntimeState* state) override { + SCOPED_TIMER(exec_time_counter()); + SCOPED_TIMER(_open_timer); + return Base::open(state); + } + + friend class PaimonTableSinkOperatorX; +}; + +class PaimonTableSinkOperatorX final : public DataSinkOperatorX<PaimonTableSinkLocalState> { +public: + using Base = DataSinkOperatorX<PaimonTableSinkLocalState>; + PaimonTableSinkOperatorX(ObjectPool* pool, int operator_id, const RowDescriptor& row_desc, + const std::vector<TExpr>& t_output_expr) + : Base(operator_id, 0, 0), + _row_desc(row_desc), + _t_output_expr(t_output_expr), + _pool(pool) {} + + Status init(const TDataSink& thrift_sink) override { + RETURN_IF_ERROR(Base::init(thrift_sink)); + DCHECK(thrift_sink.__isset.paimon_table_sink); + RETURN_IF_ERROR(VExpr::create_expr_trees(_t_output_expr, _output_vexpr_ctxs)); + return Status::OK(); + } + + Status prepare(RuntimeState* state) override { + RETURN_IF_ERROR(Base::prepare(state)); + RETURN_IF_ERROR(VExpr::prepare(_output_vexpr_ctxs, state, _row_desc)); + return VExpr::open(_output_vexpr_ctxs, state); + } + + Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override { + auto& local_state = get_local_state(state); + SCOPED_TIMER(local_state.exec_time_counter()); + COUNTER_UPDATE(local_state.rows_input_counter(), static_cast<int64_t>(in_block->rows())); + return local_state.sink(state, in_block, eos); + } + +private: + friend class PaimonTableSinkLocalState; + template <typename Writer, typename Parent> + requires(std::is_base_of_v<AsyncResultWriter, Writer>) + friend class AsyncWriterSink; + const RowDescriptor& _row_desc; + VExprContextSPtrs _output_vexpr_ctxs; + const std::vector<TExpr>& _t_output_expr; + ObjectPool* _pool = nullptr; +}; + +} // namespace doris diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp index f8bd3120846..ad1e3947ebb 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.cpp +++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp @@ -87,6 +87,7 @@ #include "exec/operator/olap_scan_operator.h" #include "exec/operator/olap_table_sink_operator.h" #include "exec/operator/olap_table_sink_v2_operator.h" +#include "exec/operator/paimon_table_sink_operator.h" #include "exec/operator/partition_sort_sink_operator.h" #include "exec/operator/partition_sort_source_operator.h" #include "exec/operator/partitioned_aggregation_sink_operator.h" @@ -1371,6 +1372,14 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS output_exprs); break; } + case TDataSinkType::PAIMON_TABLE_SINK: { + if (!thrift_sink.__isset.paimon_table_sink) { + return Status::InternalError("Missing paimon table sink."); + } + _sink = std::make_shared<PaimonTableSinkOperatorX>(pool, next_sink_operator_id(), + row_desc, output_exprs); + break; + } case TDataSinkType::JDBC_TABLE_SINK: { if (!thrift_sink.__isset.jdbc_table_sink) { return Status::InternalError("Missing data jdbc sink."); 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 new file mode 100644 index 00000000000..f69b0311ad9 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.cpp @@ -0,0 +1,39 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "exec/sink/writer/paimon/ffi_paimon_write_backend.h" + +namespace doris { + +Status FfiPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* state) { + return Status::NotSupported( + "FFI (Rust) Paimon write backend is not yet implemented. " + "Currently only the JNI (Java) backend is available. " + "The FFI backend will be introduced in v2 for high-throughput append-only writes."); +} + +Status FfiPaimonWriteBackend::create_writer(const std::string& partition_bytes, int32_t bucket, + std::unique_ptr<IPaimonWriter>* writer) { + return Status::NotSupported("FFI backend: create_writer not yet implemented"); +} + +Status FfiPaimonWriteBackend::create_committer( + std::unique_ptr<IPaimonCommitter>* committer) { + return Status::NotSupported("FFI backend: create_committer not yet implemented"); +} + +} // namespace doris 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 new file mode 100644 index 00000000000..ee2b268b138 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/ffi_paimon_write_backend.h @@ -0,0 +1,59 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include "exec/sink/writer/paimon/paimon_write_backend.h" + +namespace doris { + +/// Stub implementation of Paimon write backend via Rust C FFI. +/// +/// This is a placeholder for future use. In v1, all methods return +/// Status::NotSupported. When paimon-rust is integrated (v2), this +/// will be replaced with a real implementation that: +/// +/// 1. Dynamically loads the paimon-rust shared library (libpaimon_c.so) +/// 2. Converts Doris Block → Arrow C Data Interface (zero-copy) +/// 3. Calls Rust TableWrite::write_arrow_batch() via FFI +/// 4. Calls Rust TableCommit::commit() via FFI +/// +/// v1 scope: stub only. Upper layers (VPaimonTableWriter) are coded +/// against IPaimonWriteBackend and never need to change. +class FfiPaimonWriteBackend final : public IPaimonWriteBackend { +public: + FfiPaimonWriteBackend() = default; + ~FfiPaimonWriteBackend() override = default; + + 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_committer(std::unique_ptr<IPaimonCommitter>* committer) override; + + PaimonBackendType type() const override { return PaimonBackendType::FFI; } + + // Rust currently supports a subset of Java's features + bool supports_compaction() const override { return false; } + bool supports_lookup_changelog_producer() const override { return false; } + bool supports_full_compaction_changelog_producer() const override { return false; } + bool supports_partial_update_with_dv() const override { return false; } + bool supports_aggregation_with_dv() const override { return false; } +}; + +} // namespace doris 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 new file mode 100644 index 00000000000..66ed1fc4f2f --- /dev/null +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp @@ -0,0 +1,446 @@ +// 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/jni_paimon_write_backend.h" + +#include <arrow/buffer.h> +#include <arrow/io/memory.h> +#include <arrow/ipc/writer.h> +#include <arrow/record_batch.h> +#include <arrow/type.h> + +#include "common/logging.h" +#include "format/arrow/arrow_block_convertor.h" +#include "format/arrow/arrow_row_batch.h" +#include "runtime/runtime_state.h" +#include "util/jni-util.h" + +namespace doris { + +// ──────────────────────────────────────────────────────────── +// JniPaimonWriteBackend +// ──────────────────────────────────────────────────────────── + +static constexpr const char* PAIMON_JNI_WRITER_CLASS = + "org/apache/doris/paimon/PaimonJniWriter"; +static constexpr const char* PAIMON_OUTPUT_COLUMN_NAMES_KEY = + "doris.output_column_names"; +static constexpr char PAIMON_COLUMN_NAME_SEPARATOR = '\x01'; + +JniPaimonWriteBackend::~JniPaimonWriteBackend() { + JNIEnv* env = nullptr; + if (_get_jni_env(&env).ok()) { + if (_jni_writer_obj != nullptr) { + env->DeleteGlobalRef(_jni_writer_obj); + _jni_writer_obj = nullptr; + } + if (_jni_writer_cls != nullptr) { + env->DeleteGlobalRef(_jni_writer_cls); + _jni_writer_cls = nullptr; + } + } +} + +Status JniPaimonWriteBackend::_get_jni_env(JNIEnv** env) { + JavaVM* jvm = nullptr; + jsize n_vms = 0; + jint result = JNI_GetCreatedJavaVMs(&jvm, 1, &n_vms); + if (result != JNI_OK || n_vms == 0) { + return Status::InternalError("Failed to get created JavaVM"); + } + result = jvm->GetEnv(reinterpret_cast<void**>(env), JNI_VERSION_1_8); + if (result == JNI_EDETACHED) { + result = jvm->AttachCurrentThread(reinterpret_cast<void**>(env), nullptr); + if (result != JNI_OK) { + return Status::InternalError("Failed to attach current thread to JVM"); + } + } else if (result != JNI_OK) { + return Status::InternalError("Failed to get JNIEnv"); + } + return Status::OK(); +} + +Status JniPaimonWriteBackend::_check_jni_exception(JNIEnv* env, + const std::string& method_name) { + if (env->ExceptionCheck()) { + Status st = Jni::Env::GetJniExceptionMsg( + env, true, "JNI exception in " + method_name + ": "); + LOG(WARNING) << st.to_string(); + return st; + } + return Status::OK(); +} + +static std::vector<std::string> _split_column_names(const std::string& column_names_str) { + std::vector<std::string> names; + size_t begin = 0; + while (begin <= column_names_str.size()) { + size_t end = column_names_str.find(PAIMON_COLUMN_NAME_SEPARATOR, begin); + if (end == std::string::npos) { + names.emplace_back(column_names_str.substr(begin)); + break; + } + names.emplace_back(column_names_str.substr(begin, end - begin)); + begin = end + 1; + } + return names; +} + +static jobject _to_java_options(JNIEnv* env, + const std::map<std::string, std::string>& options) { + jclass map_cls = env->FindClass("java/util/HashMap"); + jmethodID map_ctor = env->GetMethodID(map_cls, "<init>", "()V"); + jmethodID put_method = env->GetMethodID( + map_cls, "put", + "(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;"); + + jobject map_obj = env->NewObject(map_cls, map_ctor); + for (const auto& kv : options) { + jstring key = env->NewStringUTF(kv.first.c_str()); + jstring val = env->NewStringUTF(kv.second.c_str()); + env->CallObjectMethod(map_obj, put_method, key, val); + env->DeleteLocalRef(key); + env->DeleteLocalRef(val); + } + env->DeleteLocalRef(map_cls); + return map_obj; +} + +Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* state) { + _sink = sink; + _arrow_pool = std::make_unique<ArrowMemoryPool<>>(); + + JNIEnv* env = nullptr; + RETURN_IF_ERROR(_get_jni_env(&env)); + + // Find the Java writer class + jclass local_cls = env->FindClass(PAIMON_JNI_WRITER_CLASS); + RETURN_IF_ERROR(_check_jni_exception(env, "FindClass")); + if (local_cls == nullptr) { + return Status::InternalError("Failed to find Java class: {}", + PAIMON_JNI_WRITER_CLASS); + } + _jni_writer_cls = static_cast<jclass>(env->NewGlobalRef(local_cls)); + env->DeleteLocalRef(local_cls); + + // Cache method IDs + _open_id = env->GetMethodID(_jni_writer_cls, "open", + "(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;)V"); + _write_id = env->GetMethodID(_jni_writer_cls, "write", "(JI)V"); + _prepare_commit_id = env->GetMethodID(_jni_writer_cls, "prepareCommit", "()[[B"); + _abort_id = env->GetMethodID(_jni_writer_cls, "abort", "()V"); + _close_id = env->GetMethodID(_jni_writer_cls, "close", "()V"); + RETURN_IF_ERROR(_check_jni_exception(env, "GetMethodID")); + + // Create the JNI writer object + jmethodID ctor_id = env->GetMethodID(_jni_writer_cls, "<init>", "()V"); + jobject local_obj = env->NewObject(_jni_writer_cls, ctor_id); + RETURN_IF_ERROR(_check_jni_exception(env, "NewObject")); + _jni_writer_obj = env->NewGlobalRef(local_obj); + env->DeleteLocalRef(local_obj); + + // Build options map + std::map<std::string, std::string> jni_options; + if (sink.__isset.paimon_options) { + jni_options.insert(sink.paimon_options.begin(), sink.paimon_options.end()); + } + if (sink.__isset.hadoop_config) { + for (const auto& [key, value] : sink.hadoop_config) { + jni_options[key] = value; + } + } + jni_options["db_name"] = sink.db_name; + jni_options["table_name"] = sink.tb_name; + if (sink.__isset.serialized_table && !sink.serialized_table.empty()) { + jni_options["serialized_table"] = sink.serialized_table; + } + + // Build column names array + jstring j_location = env->NewStringUTF(sink.table_location.c_str()); + jobject j_options = _to_java_options(env, jni_options); + + jclass string_cls = env->FindClass("java/lang/String"); + jobjectArray j_cols = nullptr; + if (sink.__isset.column_names && !sink.column_names.empty()) { + j_cols = env->NewObjectArray(static_cast<jsize>(sink.column_names.size()), + string_cls, nullptr); + for (size_t i = 0; i < sink.column_names.size(); ++i) { + jstring str = env->NewStringUTF(sink.column_names[i].c_str()); + env->SetObjectArrayElement(j_cols, static_cast<jsize>(i), str); + env->DeleteLocalRef(str); + } + } else { + j_cols = env->NewObjectArray(0, string_cls, nullptr); + } + + // Call Java open() + env->CallVoidMethod(_jni_writer_obj, _open_id, j_location, j_options, j_cols); + Status st = _check_jni_exception(env, "open"); + + env->DeleteLocalRef(j_location); + env->DeleteLocalRef(j_options); + env->DeleteLocalRef(j_cols); + env->DeleteLocalRef(string_cls); + + if (st.ok()) { + _opened = true; + } + return st; +} + +Status JniPaimonWriteBackend::create_writer(const std::string& partition_bytes, int32_t bucket, + 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); + *writer = std::move(jni_writer); + return Status::OK(); +} + +Status JniPaimonWriteBackend::create_committer( + std::unique_ptr<IPaimonCommitter>* committer) { + auto c = std::make_unique<JniPaimonCommitter>(_sink); + *committer = std::move(c); + return Status::OK(); +} + +// ──────────────────────────────────────────────────────────── +// JniPaimonWriter +// ──────────────────────────────────────────────────────────── + +JniPaimonWriter::JniPaimonWriter(JNIEnv* env, jobject jni_writer_obj, + jmethodID write_id, jmethodID prepare_commit_id, + jmethodID abort_id, + std::unique_ptr<ArrowMemoryPool<>> arrow_pool, + const TPaimonTableSink& sink) + : _env(env), + _jni_writer_obj(jni_writer_obj), + _write_id(write_id), + _prepare_commit_id(prepare_commit_id), + _abort_id(abort_id), + _arrow_pool(std::move(arrow_pool)), + _sink(sink) {} + +JniPaimonWriter::~JniPaimonWriter() = default; + +Status JniPaimonWriter::_write_projected_block(Block& block) { + if (block.rows() == 0) { + return Status::OK(); + } + + _row_count += block.rows(); + + // Determine output column names + if (_sink.__isset.paimon_options) { + auto it = _sink.paimon_options.find(PAIMON_OUTPUT_COLUMN_NAMES_KEY); + if (it != _sink.paimon_options.end()) { + auto output_names = _split_column_names(it->second); + DCHECK_EQ(output_names.size(), block.columns()); + for (size_t i = 0; i < output_names.size(); ++i) { + block.get_by_position(i).name = output_names[i]; + } + } + } + + // Convert Block → Arrow RecordBatch → IPC Stream + std::shared_ptr<arrow::Schema> arrow_schema; + RETURN_IF_ERROR(get_arrow_schema_from_block(block, &arrow_schema, "UTC")); + + std::shared_ptr<arrow::RecordBatch> record_batch; + RETURN_IF_ERROR(convert_to_arrow_batch(block, arrow_schema, _arrow_pool.get(), + &record_batch, cctz::utc_time_zone())); + + auto out_stream_res = arrow::io::BufferOutputStream::Create(); + if (!out_stream_res.ok()) { + return Status::InternalError("Arrow BufferOutputStream create failed: {}", + out_stream_res.status().ToString()); + } + auto out_stream = *out_stream_res; + + auto writer_res = arrow::ipc::MakeStreamWriter(out_stream, arrow_schema); + if (!writer_res.ok()) { + return Status::InternalError("Arrow StreamWriter create failed: {}", + writer_res.status().ToString()); + } + auto ipc_writer = *writer_res; + if (!ipc_writer->WriteRecordBatch(*record_batch).ok()) { + return Status::InternalError("Arrow WriteRecordBatch failed"); + } + if (!ipc_writer->Close().ok()) { + return Status::InternalError("Arrow StreamWriter close failed"); + } + + auto buffer_res = out_stream->Finish(); + if (!buffer_res.ok()) { + return Status::InternalError("Arrow output stream finish failed: {}", + buffer_res.status().ToString()); + } + std::shared_ptr<arrow::Buffer> buffer = *buffer_res; + + // Pass to Java via JNI + JNIEnv* env = nullptr; + { + JavaVM* jvm = nullptr; + jsize n_vms = 0; + JNI_GetCreatedJavaVMs(&jvm, 1, &n_vms); + jvm->GetEnv(reinterpret_cast<void**>(&env), JNI_VERSION_1_8); + if (env == nullptr) { + jvm->AttachCurrentThread(reinterpret_cast<void**>(&env), nullptr); + } + } + + auto address = reinterpret_cast<jlong>(buffer->data()); + jint length = static_cast<jint>(buffer->size()); + env->CallVoidMethod(_jni_writer_obj, _write_id, address, length); + RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg( + env, false, "JNI exception in JniPaimonWriter::write: ")); + return Status::OK(); +} + +Status JniPaimonWriter::write(RuntimeState* state, Block& block) { + return _write_projected_block(block); +} + +Status JniPaimonWriter::prepare_commit(std::vector<TPaimonCommitMessage>& messages) { + JNIEnv* env = nullptr; + { + JavaVM* jvm = nullptr; + jsize n_vms = 0; + JNI_GetCreatedJavaVMs(&jvm, 1, &n_vms); + jvm->GetEnv(reinterpret_cast<void**>(&env), JNI_VERSION_1_8); + if (env == nullptr) { + jvm->AttachCurrentThread(reinterpret_cast<void**>(&env), nullptr); + } + } + + jobject j_payloads_obj = env->CallObjectMethod(_jni_writer_obj, _prepare_commit_id); + Status st = Jni::Env::GetJniExceptionMsg( + env, false, "JNI exception in prepareCommit: "); + if (!st.ok()) { + return st; + } + + if (j_payloads_obj == nullptr) { + return Status::OK(); // no data written + } + + auto* j_payloads = static_cast<jobjectArray>(j_payloads_obj); + jsize num_payloads = env->GetArrayLength(j_payloads); + + for (jsize i = 0; i < num_payloads; ++i) { + jbyteArray j_bytes = static_cast<jbyteArray>( + env->GetObjectArrayElement(j_payloads, i)); + if (j_bytes == nullptr) { + continue; + } + jsize len = env->GetArrayLength(j_bytes); + if (len > 0) { + jbyte* bytes = env->GetByteArrayElements(j_bytes, nullptr); + if (bytes != nullptr) { + std::string payload(reinterpret_cast<char*>(bytes), + static_cast<size_t>(len)); + TPaimonCommitMessage msg; + msg.__set_payload(payload); + messages.emplace_back(std::move(msg)); + env->ReleaseByteArrayElements(j_bytes, bytes, JNI_ABORT); + } + } + env->DeleteLocalRef(j_bytes); + } + env->DeleteLocalRef(j_payloads); + return Status::OK(); +} + +Status JniPaimonWriter::compact(bool full_compaction) { + // Compaction is triggered by Java's prepareCommit, not as a separate call + // from BE. For explicit compaction, we would need an additional JNI call. + return Status::OK(); +} + +Status JniPaimonWriter::abort() { + JNIEnv* env = nullptr; + { + JavaVM* jvm = nullptr; + jsize n_vms = 0; + JNI_GetCreatedJavaVMs(&jvm, 1, &n_vms); + jvm->GetEnv(reinterpret_cast<void**>(&env), JNI_VERSION_1_8); + if (env == nullptr) { + jvm->AttachCurrentThread(reinterpret_cast<void**>(&env), nullptr); + } + } + env->CallVoidMethod(_jni_writer_obj, _abort_id); + return Jni::Env::GetJniExceptionMsg( + env, true, "JNI exception in abort: "); +} + +// ──────────────────────────────────────────────────────────── +// JniPaimonCommitter +// ──────────────────────────────────────────────────────────── + +JniPaimonCommitter::JniPaimonCommitter(const TPaimonTableSink& sink) : _sink(sink) {} + +Status JniPaimonCommitter::_commit_impl( + const std::vector<TPaimonCommitMessage>& messages, bool overwrite_mode, + const std::map<std::string, std::string>* static_partition) { + // The actual commit is performed by FE (PaimonTransaction). + // BE only collects and forwards the CommitMessages. + // The commit logic is implemented in FE because it requires + // Paimon SDK classes (StreamTableCommit) and the coordinator role. + // + // This method exists for future use when we implement + // commit-via-CommitBE pattern (see design doc Section 5.3, Option 3). + return Status::NotSupported( + "Commit is handled by FE Coordinator (PaimonTransaction); " + "JniPaimonCommitter is reserved for future Commit-BE pattern"); +} + +Status JniPaimonCommitter::commit(const std::vector<TPaimonCommitMessage>& messages) { + return _commit_impl(messages, false, nullptr); +} + +Status JniPaimonCommitter::overwrite( + const std::vector<TPaimonCommitMessage>& messages, + const std::map<std::string, std::string>& static_partition) { + return _commit_impl(messages, true, &static_partition); +} + +Status JniPaimonCommitter::truncate_table() { + return Status::NotSupported( + "truncate_table is handled by FE Coordinator"); +} + +Status JniPaimonCommitter::truncate_partitions( + const std::vector<std::map<std::string, std::string>>& partitions) { + return Status::NotSupported( + "truncate_partitions is handled by FE Coordinator"); +} + +Status JniPaimonCommitter::abort(const std::vector<TPaimonCommitMessage>& messages) { + // Abort is best-effort: delete the data files. + // In the current architecture, FE handles abort through PaimonTransaction. + return Status::NotSupported( + "abort is handled by FE Coordinator"); +} + +} // namespace doris 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 new file mode 100644 index 00000000000..19944c9cb0e --- /dev/null +++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h @@ -0,0 +1,140 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <gen_cpp/DataSinks_types.h> +#include <jni.h> + +#include <memory> +#include <string> + +#include "common/status.h" +#include "exec/sink/writer/paimon/paimon_write_backend.h" + +namespace doris { + +class RuntimeState; + +/// JNI-based Paimon write backend. +/// +/// 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. +/// +/// Data path: +/// Block → Arrow RecordBatch → IPC Stream → JNI direct buffer → +/// Java ArrowStreamReader → Paimon InternalRow → BatchTableWrite +/// +/// Commit path: +/// Java prepareCommit() → CommitMessageSerializer → byte[][] +/// → C++ TPaimonCommitMessage → RuntimeState → FE Coordinator +class JniPaimonWriteBackend final : public IPaimonWriteBackend { +public: + JniPaimonWriteBackend() = default; + ~JniPaimonWriteBackend() override; + + 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_committer(std::unique_ptr<IPaimonCommitter>* committer) override; + + PaimonBackendType type() const override { return PaimonBackendType::JNI; } + + bool supports_compaction() const override { return true; } + bool supports_lookup_changelog_producer() const override { return true; } + bool supports_full_compaction_changelog_producer() const override { return true; } + bool supports_partial_update_with_dv() const override { return true; } + bool supports_aggregation_with_dv() const override { return true; } + +private: + Status _get_jni_env(JNIEnv** env); + Status _check_jni_exception(JNIEnv* env, const std::string& method_name); + + // JNI global references + jclass _jni_writer_cls = nullptr; + jobject _jni_writer_obj = nullptr; + + // Cached method IDs for the Java PaimonJniWriter class + jmethodID _open_id = nullptr; + jmethodID _write_id = nullptr; + jmethodID _prepare_commit_id = nullptr; + jmethodID _abort_id = nullptr; + jmethodID _close_id = nullptr; + + // Arrow memory pool for Block → Arrow conversion + std::unique_ptr<ArrowMemoryPool<>> _arrow_pool; + + // Parsed sink configuration + TPaimonTableSink _sink; + + // Whether the JNI writer has been opened + bool _opened = false; +}; + +/// Per-(partition, bucket) writer backed by the shared JNI context. +class JniPaimonWriter final : public IPaimonWriter { +public: + JniPaimonWriter(JNIEnv* env, jobject jni_writer_obj, jmethodID write_id, + jmethodID prepare_commit_id, jmethodID abort_id, + std::unique_ptr<ArrowMemoryPool<>> arrow_pool, + const TPaimonTableSink& sink); + ~JniPaimonWriter() override; + + Status write(RuntimeState* state, Block& block) override; + Status prepare_commit(std::vector<TPaimonCommitMessage>& messages) override; + Status compact(bool full_compaction) override; + Status abort() override; + +private: + Status _write_projected_block(Block& block); + + JNIEnv* _env; + jobject _jni_writer_obj; + jmethodID _write_id; + jmethodID _prepare_commit_id; + jmethodID _abort_id; + std::unique_ptr<ArrowMemoryPool<>> _arrow_pool; + TPaimonTableSink _sink; + int64_t _row_count = 0; +}; + +/// Committer that delegates to the Java Paimon SDK. +class JniPaimonCommitter final : public IPaimonCommitter { +public: + explicit JniPaimonCommitter(const TPaimonTableSink& sink); + ~JniPaimonCommitter() override = default; + + Status commit(const std::vector<TPaimonCommitMessage>& messages) override; + Status overwrite(const std::vector<TPaimonCommitMessage>& messages, + const std::map<std::string, std::string>& static_partition) override; + Status truncate_table() override; + Status truncate_partitions( + const std::vector<std::map<std::string, std::string>>& partitions) override; + Status abort(const std::vector<TPaimonCommitMessage>& messages) override; + +private: + Status _commit_impl(const std::vector<TPaimonCommitMessage>& messages, bool overwrite_mode, + const std::map<std::string, std::string>* static_partition); + + TPaimonTableSink _sink; +}; + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_write_backend.h b/be/src/exec/sink/writer/paimon/paimon_write_backend.h new file mode 100644 index 00000000000..f35debf0e9a --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_write_backend.h @@ -0,0 +1,151 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <gen_cpp/DataSinks_types.h> + +#include <memory> +#include <string> +#include <vector> + +#include "common/status.h" +#include "core/block/block.h" + +namespace doris { + +class RuntimeState; + +// ──────────────────────────────────────────────────────────── +// Backend type enumeration +// ──────────────────────────────────────────────────────────── + +enum class PaimonBackendType { + JNI, // Java SDK via JNI + FFI // Rust via C FFI +}; + +// ──────────────────────────────────────────────────────────── +// IPaimonWriter — per-(partition, bucket) writer +// ──────────────────────────────────────────────────────────── + +class IPaimonWriter { +public: + virtual ~IPaimonWriter() = default; + + /// Write a Doris columnar block to this writer. + virtual Status write(RuntimeState* state, Block& block) = 0; + + /// Prepare commit: close files and produce commit messages. + /// The caller owns the returned messages and is responsible for forwarding + /// them to the commit coordinator (FE). + virtual Status prepare_commit(std::vector<TPaimonCommitMessage>& messages) = 0; + + /// Trigger compaction for this (partition, bucket). + /// Only supported by the JNI backend; FFI backend returns NotSupported. + virtual Status compact(bool full_compaction) = 0; + + /// Abort: discard all data written by this writer. + /// Best-effort cleanup; errors are logged but not propagated. + virtual Status abort() = 0; +}; + +// ──────────────────────────────────────────────────────────── +// IPaimonCommitter — snapshot commit coordinator +// ──────────────────────────────────────────────────────────── + +class IPaimonCommitter { +public: + virtual ~IPaimonCommitter() = default; + + /// Commit new files in APPEND mode. + /// All messages are committed atomically in one snapshot. + virtual Status commit(const std::vector<TPaimonCommitMessage>& messages) = 0; + + /// Commit in OVERWRITE mode, optionally scoped to static partitions. + virtual Status overwrite( + const std::vector<TPaimonCommitMessage>& messages, + const std::map<std::string, std::string>& static_partition) = 0; + + /// Truncate the entire table. + virtual Status truncate_table() = 0; + + /// Truncate specific partitions. + virtual Status truncate_partitions( + const std::vector<std::map<std::string, std::string>>& partitions) = 0; + + /// Abort: delete data files that were prepared but not yet committed. + virtual Status abort(const std::vector<TPaimonCommitMessage>& messages) = 0; +}; + +// ──────────────────────────────────────────────────────────── +// IPaimonWriteBackend — factory for writers and committers +// ──────────────────────────────────────────────────────────── +// +// Java (JNI) and Rust (FFI) are interchangeable implementations of +// this interface. Rust is a functional subset of Java. The selection +// logic prefers Rust (performance) and falls back to Java when the +// table requires features that Rust does not yet support. +// +// v1 only implements JniPaimonWriteBackend. FfiPaimonWriteBackend +// exists as a stub that returns NotSupported for all methods. + +class IPaimonWriteBackend { +public: + virtual ~IPaimonWriteBackend() = default; + + /// 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 committer. One per sink; shared across all writers. + virtual Status create_committer(std::unique_ptr<IPaimonCommitter>* committer) = 0; + + /// Which backend type is this? + virtual PaimonBackendType type() const = 0; + + // ──── Capability queries ──────────────────────────────── + + virtual bool supports_compaction() const = 0; + virtual bool supports_lookup_changelog_producer() const = 0; + virtual bool supports_full_compaction_changelog_producer() const = 0; + virtual bool supports_partial_update_with_dv() const = 0; + virtual bool supports_aggregation_with_dv() const = 0; +}; + +// ──────────────────────────────────────────────────────────── +// Factory +// ──────────────────────────────────────────────────────────── + +class PaimonWriteBackendFactory { +public: + /// Create the appropriate backend based on the sink configuration. + static Status create(const TPaimonTableSink& sink, + std::unique_ptr<IPaimonWriteBackend>* backend); + + /// Select the best backend type for the given table configuration. + /// v1: always returns JNI. + /// v2: returns FFI when the table does not require Java-only features. + static PaimonBackendType select_backend_type(const TPaimonTableSink& sink); +}; + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/paimon_write_backend_factory.cpp b/be/src/exec/sink/writer/paimon/paimon_write_backend_factory.cpp new file mode 100644 index 00000000000..d289eb71591 --- /dev/null +++ b/be/src/exec/sink/writer/paimon/paimon_write_backend_factory.cpp @@ -0,0 +1,65 @@ +// 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_write_backend.h" + +#include "exec/sink/writer/paimon/ffi_paimon_write_backend.h" +#include "exec/sink/writer/paimon/jni_paimon_write_backend.h" + +namespace doris { + +Status PaimonWriteBackendFactory::create(const TPaimonTableSink& sink, + std::unique_ptr<IPaimonWriteBackend>* backend) { + PaimonBackendType type = select_backend_type(sink); + + switch (type) { + case PaimonBackendType::JNI: { + *backend = std::make_unique<JniPaimonWriteBackend>(); + return Status::OK(); + } + case PaimonBackendType::FFI: { + *backend = std::make_unique<FfiPaimonWriteBackend>(); + return Status::OK(); + } + } + return Status::InternalError("Unknown Paimon backend type: {}", + static_cast<int>(type)); +} + +PaimonBackendType PaimonWriteBackendFactory::select_backend_type( + const TPaimonTableSink& sink) { + // v1: Always use JNI backend. + // + // v2: When Rust FFI backend is ready, use the following logic: + // if (sink.__isset.backend_type && sink.backend_type == TPaimonWriteBackendType::FFI) { + // return PaimonBackendType::FFI; + // } + // return PaimonBackendType::JNI; // default fallback + // + // The FFI backend is preferred when: + // - Table is append-only (no primary keys) + // - No Lookup/FullCompaction changelog producer + // - No DeletionVectors + PartialUpdate/Aggregation + if (sink.__isset.backend_type) { + if (sink.backend_type == TPaimonWriteBackendType::FFI) { + return PaimonBackendType::FFI; + } + } + return PaimonBackendType::JNI; +} + +} // namespace doris diff --git a/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp new file mode 100644 index 00000000000..d312821c9cc --- /dev/null +++ b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.cpp @@ -0,0 +1,253 @@ +// 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/vpaimon_table_writer.h" + +#include "common/logging.h" +#include "core/block/block.h" +#include "runtime/runtime_state.h" + +namespace doris { + +static constexpr const char* PAIMON_OUTPUT_COLUMN_NAMES_KEY = + "doris.output_column_names"; +static constexpr char PAIMON_COLUMN_NAME_SEPARATOR = '\x01'; + +VPaimonTableWriter::VPaimonTableWriter(const TDataSink& t_sink, + const VExprContextSPtrs& output_exprs, + std::shared_ptr<Dependency> dep, + std::shared_ptr<Dependency> fin_dep) + : AsyncResultWriter(output_exprs, std::move(dep), std::move(fin_dep)), + _t_sink(t_sink) { + DCHECK(_t_sink.__isset.paimon_table_sink); +} + +Status VPaimonTableWriter::open(RuntimeState* state, RuntimeProfile* profile) { + _state = state; + _operator_profile = profile; + + // Register profile counters + _written_rows_counter = ADD_COUNTER(_operator_profile, "WrittenRows", TUnit::UNIT); + _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"); + _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"); + _commit_payload_count = ADD_COUNTER(_operator_profile, "CommitPayloadCount", TUnit::UNIT); + _commit_payload_bytes_counter = + ADD_COUNTER(_operator_profile, "CommitPayloadBytes", TUnit::BYTES); + _buffer_flush_count = ADD_COUNTER(_operator_profile, "BufferFlushCount", TUnit::UNIT); + + SCOPED_TIMER(_open_timer); + + // Create the backend via factory + 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 (the actual per-partition-bucket routing is + // handled inside Paimon's Java BatchTableWrite). + RETURN_IF_ERROR(_backend->create_writer("" /* partition */, 0 /* bucket */, &_writer)); + + LOG(INFO) << "VPaimonTableWriter opened: table=" << _t_sink.paimon_table_sink.tb_name + << ", backend=" << static_cast<int>(_backend->type()); + return Status::OK(); +} + +Status VPaimonTableWriter::write(RuntimeState* state, Block& block) { + if (block.rows() == 0) { + return Status::OK(); + } + + SCOPED_TIMER(_send_data_timer); + + // Project: apply output expressions + Block output_block; + { + SCOPED_TIMER(_project_timer); + RETURN_IF_ERROR(_projection_block(block, &output_block)); + } + + // Rename columns using the output column names from the sink config + const auto& paimon_sink = _t_sink.paimon_table_sink; + if (paimon_sink.__isset.paimon_options) { + auto it = paimon_sink.paimon_options.find(PAIMON_OUTPUT_COLUMN_NAMES_KEY); + if (it != paimon_sink.paimon_options.end()) { + auto output_names = _split_column_names(it->second); + if (!output_names.empty() && output_names.size() == output_block.columns()) { + for (size_t i = 0; i < output_names.size(); ++i) { + output_block.get_by_position(i).name = output_names[i]; + } + } + } + } + + // If the block is already large enough, send it directly. + // Otherwise, buffer it to reduce JNI call overhead. + if (output_block.rows() >= BATCH_MAX_ROWS || + output_block.bytes() >= BATCH_MAX_BYTES) { + RETURN_IF_ERROR(_flush_buffer()); + return _write_projected_block(output_block); + } + + // Check if appending would overflow the buffer + if (_buffered_rows > 0 && + (_buffered_rows + output_block.rows() >= BATCH_MAX_ROWS || + _buffered_bytes + output_block.bytes() >= BATCH_MAX_BYTES)) { + RETURN_IF_ERROR(_flush_buffer()); + } + + RETURN_IF_ERROR(_append_to_buffer(output_block)); + if (_buffered_rows >= BATCH_MAX_ROWS || _buffered_bytes >= BATCH_MAX_BYTES) { + return _flush_buffer(); + } + return Status::OK(); +} + +Status VPaimonTableWriter::_append_to_buffer(const Block& block) { + if (!_buffer) { + _buffer = Block::create_unique(block.clone_empty()); + _buffered_rows = 0; + _buffered_bytes = 0; + } + auto columns = std::move(*_buffer).mutate_columns(); + const int cols = block.columns(); + const size_t rows = block.rows(); + for (int col = 0; col < cols; ++col) { + columns[col]->insert_range_from(*block.get_by_position(col).column, 0, rows); + } + _buffer->set_columns(std::move(columns)); + _buffered_rows += rows; + _buffered_bytes += block.bytes(); + return Status::OK(); +} + +Status VPaimonTableWriter::_flush_buffer() { + if (!_buffer || _buffered_rows == 0) { + return Status::OK(); + } + Status st = _write_projected_block(*_buffer); + COUNTER_UPDATE(_buffer_flush_count, 1); + _buffer.reset(); + _buffered_rows = 0; + _buffered_bytes = 0; + return st; +} + +Status VPaimonTableWriter::_write_projected_block(Block& block) { + if (block.rows() == 0) { + return Status::OK(); + } + + COUNTER_UPDATE(_written_rows_counter, block.rows()); + COUNTER_UPDATE(_written_bytes_counter, block.bytes()); + _state->update_num_rows_load_total(block.rows()); + _state->update_num_bytes_load_total(block.bytes()); + + if (_writer) { + SCOPED_TIMER(_file_store_write_timer); + RETURN_IF_ERROR(_writer->write(_state, block)); + } + _written_rows += block.rows(); + return Status::OK(); +} + +Status VPaimonTableWriter::close(Status status) { + SCOPED_TIMER(_close_timer); + + // Flush any remaining buffered data + if (status.ok()) { + Status flush_st = _flush_buffer(); + if (!flush_st.ok()) { + status = flush_st; + } + } else { + _buffer.reset(); + _buffered_rows = 0; + _buffered_bytes = 0; + } + + // Collect commit messages from the writer + std::vector<TPaimonCommitMessage> messages; + if (status.ok() && _writer) { + { + SCOPED_TIMER(_prepare_commit_timer); + Status prep_st = _writer->prepare_commit(messages); + if (!prep_st.ok()) { + status = prep_st; + } + } + + if (status.ok()) { + COUNTER_UPDATE(_commit_payload_count, static_cast<int64_t>(messages.size())); + for (const auto& msg : messages) { + if (msg.__isset.payload) { + COUNTER_UPDATE(_commit_payload_bytes_counter, + static_cast<int64_t>(msg.payload.size())); + } + } + + if (!messages.empty()) { + _state->add_paimon_commit_messages(messages); + LOG(INFO) << "Paimon writer closed with " << messages.size() + << " commit messages, total rows=" << _written_rows; + } + } + } + + // On error, abort the writer + if (!status.ok()) { + LOG(WARNING) << "Paimon writer closing with error: " << status.to_string(); + if (_writer) { + Status abort_st = _writer->abort(); + if (!abort_st.ok()) { + LOG(WARNING) << "Paimon writer abort failed: " << abort_st.to_string(); + } + } + } + + _writer.reset(); + _backend.reset(); + return status; +} + +// ──────────────────────────────────────────────────────────── +// Helper: split output column names +// ──────────────────────────────────────────────────────────── + +std::vector<std::string> split_paimon_output_column_names(const std::string& column_names) { + std::vector<std::string> names; + size_t begin = 0; + while (begin <= column_names.size()) { + size_t end = column_names.find(PAIMON_COLUMN_NAME_SEPARATOR, begin); + if (end == std::string::npos) { + names.emplace_back(column_names.substr(begin)); + break; + } + names.emplace_back(column_names.substr(begin, end - begin)); + begin = end + 1; + } + return names; +} + +} // 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 new file mode 100644 index 00000000000..3f169a22c6d --- /dev/null +++ b/be/src/exec/sink/writer/paimon/vpaimon_table_writer.h @@ -0,0 +1,125 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <gen_cpp/DataSinks_types.h> + +#include <memory> +#include <string> +#include <unordered_map> + +#include "common/status.h" +#include "core/block/block.h" +#include "exec/sink/writer/async_result_writer.h" +#include "exec/sink/writer/paimon/paimon_write_backend.h" +#include "exprs/vexpr_fwd.h" +#include "runtime/runtime_profile.h" + +namespace doris { + +class ObjectPool; +class RuntimeState; + +/// VPaimonTableWriter is the main entry point for writing data to Paimon +/// tables from Doris. It inherits AsyncResultWriter to leverage the +/// producer-consumer queue pattern for async I/O. +/// +/// Architecture: +/// +/// VPaimonTableWriter (this class) +/// │ +/// │ holds a std::unique_ptr<IPaimonWriteBackend> +/// │ +/// ├── JniPaimonWriteBackend (v1, production) +/// └── FfiPaimonWriteBackend (v2, stub in v1) +/// +/// Data flow: +/// Block → _route_block() → partition/bucket groups +/// → IPaimonWriter::write() per group +/// → (JNI: Arrow IPC → Java; FFI: Arrow C Data → Rust) +/// +/// Commit flow: +/// close() → prepare_commit() on all writers +/// → collect TPaimonCommitMessage[] +/// → RuntimeState::add_paimon_commit_messages() +/// → RPC to FE Coordinator +/// → FE PaimonTransaction.commit() +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); + + ~VPaimonTableWriter() override = default; + + Status init_properties(ObjectPool* pool) { return Status::OK(); } + + Status open(RuntimeState* state, RuntimeProfile* profile) override; + + Status write(RuntimeState* state, Block& block) override; + + Status close(Status status) override; + +private: + // Append block rows to the internal buffer. Flushes if buffer is full. + Status _append_to_buffer(const Block& block); + // Flush the buffered block through the backend. + Status _flush_buffer(); + // Write a single projected block (after projection). + Status _write_projected_block(Block& block); + + TDataSink _t_sink; + RuntimeState* _state = nullptr; + + // The backend abstraction — JNI or FFI + std::unique_ptr<IPaimonWriteBackend> _backend; + + // Active writer (one per BE worker; the actual per-partition-bucket + // routing happens inside Paimon's Java BatchTableWrite). + std::unique_ptr<IPaimonWriter> _writer; + + // Output column names, extracted from sink options + std::vector<std::string> _output_column_names; + + // Buffering: small blocks are accumulated before sending to Java + // to reduce JNI call overhead. Large blocks bypass the buffer. + std::unique_ptr<Block> _buffer; + size_t _buffered_rows = 0; + size_t _buffered_bytes = 0; + static constexpr size_t BATCH_MAX_ROWS = 32768; // 32K rows + static constexpr size_t BATCH_MAX_BYTES = 4 * 1024 * 1024; // 4 MB + + // Statistics + int64_t _written_rows = 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; + RuntimeProfile::Counter* _buffer_flush_count = nullptr; +}; + +} // namespace doris diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h index 2a58dc31fc9..ac2be22ff06 100644 --- a/be/src/runtime/runtime_state.h +++ b/be/src/runtime/runtime_state.h @@ -543,6 +543,18 @@ public: _mc_commit_datas.emplace_back(mc_commit_data); } + std::vector<TPaimonCommitMessage> paimon_commit_messages() const { + std::lock_guard<std::mutex> lock(_paimon_commit_messages_mutex); + return _paimon_commit_messages; + } + + void add_paimon_commit_messages( + const std::vector<TPaimonCommitMessage>& commit_messages) { + std::lock_guard<std::mutex> lock(_paimon_commit_messages_mutex); + _paimon_commit_messages.insert(_paimon_commit_messages.end(), + commit_messages.begin(), commit_messages.end()); + } + // local runtime filter mgr, the runtime filter do not have remote target or // not need local merge should regist here. the instance exec finish, the local // runtime filter mgr can release the memory of local runtime filter @@ -975,6 +987,9 @@ private: mutable std::mutex _mc_commit_datas_mutex; std::vector<TMCCommitData> _mc_commit_datas; + mutable std::mutex _paimon_commit_messages_mutex; + std::vector<TPaimonCommitMessage> _paimon_commit_messages; + std::vector<std::unique_ptr<doris::PipelineXLocalStateBase>> _op_id_to_local_state; std::unique_ptr<doris::PipelineXSinkLocalStateBase> _sink_local_state; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
