suxiaogang223 commented on code in PR #65868: URL: https://github.com/apache/doris/pull/65868#discussion_r3671693113
########## be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp: ########## @@ -0,0 +1,430 @@ +// 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 <algorithm> +#include <map> +#include <string_view> +#include <vector> + +#include "common/check.h" +#include "common/logging.h" +#include "exec/sink/writer/paimon/paimon_jni_memory_manager.h" +#include "format/arrow/arrow_block_convertor.h" +#include "format/arrow/arrow_row_batch.h" +#include "runtime/exec_env.h" +#include "runtime/runtime_state.h" +#include "util/jni-util.h" +#include "util/pretty_printer.h" +#include "util/string_util.h" + +namespace doris { + +namespace { +constexpr std::string_view PAIMON_JNI_WRITER_IO_TMP_DIR = "paimon_jni_writer_io_tmp"; +} // namespace + +// ──────────────────────────────────────────────────────────── +// JNI helpers — JVM attachment and class loading +// ──────────────────────────────────────────────────────────── + +static constexpr const char* PAIMON_JNI_WRITER_CLASS = "org/apache/doris/paimon/PaimonJniWriter"; +static constexpr const char* SCANNER_LOADER_CLASS = + "org/apache/doris/common/classloader/ScannerLoader"; + +/// Attach the current native thread to the JVM if not already attached, +/// and return a valid JNIEnv pointer. +static Status _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(); +} + +JniPaimonWriteBackend::~JniPaimonWriteBackend() { + JNIEnv* env = nullptr; + bool java_users_stopped = _jni_writer_obj == nullptr; + Status env_status = _get_jni_env(&env); + if (env_status.ok()) { + if (_jni_writer_obj != nullptr) { + _refresh_memory_profile(); + DCHECK(_close_id != nullptr); + env->CallVoidMethod(_jni_writer_obj, _close_id); + Status close_status = _check_jni_exception(env, "close PaimonJniWriter"); Review Comment: Fixed in de8a71f281. Backend shutdown is now an explicit status-bearing step before prepared commit messages are published. A Java close failure fails the INSERT so FE aborts the transaction; native pages are retained only when Java shutdown cannot be confirmed, and the BE rejects subsequent JNI Paimon writers after that unsafe state. ########## fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java: ########## @@ -0,0 +1,152 @@ +// 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. + +package org.apache.doris.planner; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.catalog.Column; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.UserException; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.datasource.paimon.PaimonTransaction; +import org.apache.doris.datasource.paimon.PaimonWriteBinding; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertCommandContext; +import org.apache.doris.nereids.trees.plans.commands.insert.PaimonInsertCommandContext; +import org.apache.doris.thrift.TDataSink; +import org.apache.doris.thrift.TDataSinkType; +import org.apache.doris.thrift.TExplainLevel; +import org.apache.doris.thrift.TFileFormatType; +import org.apache.doris.thrift.TPaimonTableSink; +import org.apache.doris.thrift.TPaimonWriteBackendType; +import org.apache.doris.thrift.TPaimonWriteMode; + +import com.google.common.base.Preconditions; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +/** + * Paimon table sink. + * + * Generates TPaimonTableSink payload consumed by BE, including serialized table + * metadata, Hadoop authentication config, transaction identity, write mode, + * and sink column names. + * + * v1: single-writer architecture; partition/bucket routing delegated to SDK. + */ +public class PaimonTableSink extends BaseExternalTableDataSink { + private final PaimonExternalTable targetTable; + private List<Expr> outputExprs; + private List<Column> cols; + + private static final HashSet<TFileFormatType> supportedTypes = new HashSet<TFileFormatType>() {{ + add(TFileFormatType.FORMAT_ORC); + add(TFileFormatType.FORMAT_PARQUET); + }}; + + public PaimonTableSink(PaimonExternalTable targetTable) { + super(); + this.targetTable = targetTable; + } + + public void setCols(List<Column> cols) { + this.cols = cols; + } + + public void setOutputExprs(List<Expr> outputExprs) { + this.outputExprs = outputExprs; + } + + @Override + protected Set<TFileFormatType> supportedFileFormatTypes() { + return supportedTypes; + } + + @Override + public String getExplainString(String prefix, TExplainLevel explainLevel) { + StringBuilder strBuilder = new StringBuilder(); + strBuilder.append(prefix).append("PAIMON TABLE SINK\n"); + if (explainLevel == TExplainLevel.BRIEF) { + return strBuilder.toString(); + } + strBuilder.append(prefix).append(" table: ").append(targetTable.getName()).append("\n"); + return strBuilder.toString(); + } + + @Override + public void bindDataSink(Optional<InsertCommandContext> insertCtx) throws AnalysisException { + TPaimonTableSink tSink = new TPaimonTableSink(); + PaimonInsertCommandContext ctx = (PaimonInsertCommandContext) insertCtx.get(); + Preconditions.checkState(ctx.getTxnId() > 0, + "Paimon transaction must begin before sink binding"); + + PaimonTransaction transaction; + PaimonWriteBinding binding; + try { + transaction = (PaimonTransaction) targetTable.getCatalog() + .getTransactionManager().getTransaction(ctx.getTxnId()); + binding = PaimonWriteBinding.create(targetTable, ctx); Review Comment: Fixed in 7418f8cae7. PaimonWriteTarget captures one latest FileStoreTable generation at sink binding. That immutable target supplies schema binding, distribution decisions, PaimonWriteBinding, and the serialized table, so PaimonTableSink no longer reloads metadata during binding. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
