github-actions[bot] commented on code in PR #65381:
URL: https://github.com/apache/doris/pull/65381#discussion_r3601702151
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -565,6 +567,19 @@ ExecutorFactory selectInsertExecutorFactory(
emptyInsert, jobId
)
);
+ } else if (physicalSink instanceof PhysicalPaimonTableSink) {
+ boolean emptyInsert = childIsEmptyRelation(physicalSink);
Review Comment:
[P1] Keep empty Paimon overwrites executable
For `INSERT OVERWRITE ... SELECT ... WHERE 1 = 0`, this sets
`emptyInsert=true`. `initPlan()` then skips `beginTransaction()`/sink binding
and `runInternal()` returns, so `PaimonTransaction.withOverwrite()` never runs
and the old snapshot remains visible. FT-013 in this PR expects the count to
become zero, but this path cannot publish that empty snapshot. Compute
`paimonCtx` first and only take the empty-input fast path when
`!paimonCtx.isOverwrite()`.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2515,6 +2524,20 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
}
}
+ if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) {
+ params.__isset.paimon_commit_messages = true;
+
params.paimon_commit_messages.insert(params.paimon_commit_messages.end(),
pcm.begin(),
+ pcm.end());
+ } else if (!req.runtime_states.empty()) {
+ for (auto* rs : req.runtime_states) {
+ if (auto rs_pcm = rs->paimon_commit_messages(); !rs_pcm.empty()) {
+ params.__isset.paimon_commit_messages = true;
+
params.paimon_commit_messages.insert(params.paimon_commit_messages.end(),
Review Comment:
[P2] Send commit metadata in bounded RPCs
`PaimonCommitCodec` emits nominal 8 MiB chunks, but this loop concatenates
every chunk from every task RuntimeState into one `TReportExecStatusParams` and
sends it once. The default Thrift message limit is 100 MiB, so enough
files/partitions still make the final report fail after writers have drained
staged files; FE can abort only payloads it received. Send acknowledged,
idempotent batches as separate bounded RPCs instead of reassembling them here.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java:
##########
@@ -0,0 +1,111 @@
+// 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.nereids.trees.plans.physical;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.datasource.paimon.PaimonExternalDatabase;
+import org.apache.doris.datasource.paimon.PaimonExternalTable;
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.plans.AbstractPlan;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Physical Paimon table sink.
+ */
+public class PhysicalPaimonTableSink<CHILD_TYPE extends Plan>
+ extends PhysicalBaseExternalTableSink<CHILD_TYPE> {
+
+ public PhysicalPaimonTableSink(PaimonExternalDatabase database,
+ PaimonExternalTable targetTable,
+ List<Column> cols,
+ List<NamedExpression> outputExprs,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties,
+ CHILD_TYPE child) {
+ this(database, targetTable, cols, outputExprs, groupExpression,
logicalProperties,
+ PhysicalProperties.SINK_RANDOM_PARTITIONED, null, child);
+ }
+
+ public PhysicalPaimonTableSink(PaimonExternalDatabase database,
+ PaimonExternalTable targetTable,
+ List<Column> cols,
+ List<NamedExpression> outputExprs,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties,
+ PhysicalProperties physicalProperties,
+ Statistics statistics,
+ CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_PAIMON_TABLE_SINK, database, targetTable,
cols, outputExprs,
+ groupExpression, logicalProperties, physicalProperties,
statistics, child);
+ }
+
+ @Override
+ public Plan withChildren(List<Plan> children) {
+ return AbstractPlan.copyWithSameId(this, () -> new
PhysicalPaimonTableSink<>(
+ (PaimonExternalDatabase) database, (PaimonExternalTable)
targetTable,
+ cols, outputExprs, groupExpression,
+ getLogicalProperties(), physicalProperties, statistics,
children.get(0)));
+ }
+
+ @Override
+ public Plan withGroupExpression(Optional<GroupExpression> groupExpression)
{
+ return AbstractPlan.copyWithSameId(this, () -> new
PhysicalPaimonTableSink<>(
+ (PaimonExternalDatabase) database, (PaimonExternalTable)
targetTable,
+ cols, outputExprs, groupExpression,
+ getLogicalProperties(), physicalProperties, statistics,
child()));
+ }
+
+ @Override
+ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression>
groupExpression,
+ Optional<LogicalProperties> logicalProperties, List<Plan>
children) {
+ return AbstractPlan.copyWithSameId(this, () -> new
PhysicalPaimonTableSink<>(
+ (PaimonExternalDatabase) database, (PaimonExternalTable)
targetTable,
+ cols, outputExprs, groupExpression,
+ logicalProperties.get(), physicalProperties, statistics,
children.get(0)));
+ }
+
+ @Override
+ public PhysicalPaimonTableSink<Plan> withPhysicalPropertiesAndStats(
+ PhysicalProperties physicalProperties, Statistics stats) {
+ return AbstractPlan.copyWithSameId(this, () -> new
PhysicalPaimonTableSink<>(
+ (PaimonExternalDatabase) database, (PaimonExternalTable)
targetTable,
+ cols, outputExprs, groupExpression,
+ getLogicalProperties(), physicalProperties, stats, child()));
+ }
+
+ @Override
+ public PhysicalProperties getRequirePhysicalProperties() {
+ // Partition and bucket routing is handled by the Paimon SDK
internally.
+ // Doris does not need to shuffle data by partition/bucket keys.
+ return PhysicalProperties.SINK_RANDOM_PARTITIONED;
Review Comment:
[P1] Keep each Paimon bucket on one writer
`SINK_RANDOM_PARTITIONED` becomes `HIVE_TABLE_SINK_UNPARTITIONED`, which
round-robins and opens additional writer channels after 25 MiB. Equal primary
keys can therefore reach independent `TableWriteImpl`s. Each restores the same
bucket max sequence and allocates the next sequence locally, so the two
versions can tie and Paimon's reader order—not Doris input order—decides the
winner. Hash-distribute the required partition/bucket/primary-key identity to
one writer (or provide a global sequence domain), and cover duplicate keys
after crossing the scaling threshold.
##########
be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp:
##########
@@ -0,0 +1,363 @@
+// 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 <map>
+
+#include "common/check.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 {
+
+// ────────────────────────────────────────────────────────────
+// 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;
+ if (_get_jni_env(&env).ok()) {
+ if (_jni_writer_obj != nullptr) {
+ DCHECK(_close_id != nullptr);
+ env->CallVoidMethod(_jni_writer_obj, _close_id);
+ Status close_status = _check_jni_exception(env, "close
PaimonJniWriter");
+ if (!close_status.ok()) {
+ LOG(WARNING) << "Failed to close PaimonJniWriter: " <<
close_status.to_string();
+ }
+ env->DeleteGlobalRef(_jni_writer_obj);
+ _jni_writer_obj = nullptr;
+ }
+ if (_jni_writer_cls != nullptr) {
+ env->DeleteGlobalRef(_jni_writer_cls);
+ _jni_writer_cls = nullptr;
+ }
+ }
+ _opened = false;
+}
+
+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();
+}
+
+Status JniPaimonWriteBackend::_load_writer_class(JNIEnv* env, jclass*
writer_class) {
+ jclass loader_class = env->FindClass(SCANNER_LOADER_CLASS);
+ RETURN_IF_ERROR(_check_jni_exception(env, "find ScannerLoader"));
+
+ jmethodID loader_constructor = env->GetMethodID(loader_class, "<init>",
"()V");
+ jmethodID get_loaded_class = env->GetMethodID(loader_class,
"getLoadedClass",
+
"(Ljava/lang/String;)Ljava/lang/Class;");
+ RETURN_IF_ERROR(_check_jni_exception(env, "resolve ScannerLoader
methods"));
+
+ jobject loader = env->NewObject(loader_class, loader_constructor);
+ jstring class_name = env->NewStringUTF(PAIMON_JNI_WRITER_CLASS);
+ auto* loaded_class =
+ static_cast<jclass>(env->CallObjectMethod(loader,
get_loaded_class, class_name));
+ RETURN_IF_ERROR(_check_jni_exception(env, "load PaimonJniWriter"));
+
+ *writer_class = loaded_class;
+ env->DeleteLocalRef(class_name);
+ env->DeleteLocalRef(loader);
+ env->DeleteLocalRef(loader_class);
+ return Status::OK();
+}
+
+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;
+ DORIS_CHECK(sink.__isset.column_names);
+ DORIS_CHECK(sink.__isset.write_mode);
+ DORIS_CHECK(sink.__isset.serialized_table);
+ DORIS_CHECK(!sink.serialized_table.empty());
+ DORIS_CHECK(sink.__isset.transaction_id);
+ DORIS_CHECK(sink.transaction_id > 0);
+ DORIS_CHECK(sink.__isset.commit_user);
+ DORIS_CHECK(!sink.commit_user.empty());
+
+ JNIEnv* env = nullptr;
+ RETURN_IF_ERROR(_get_jni_env(&env));
+
+ // Step 1: Load PaimonJniWriter class through ScannerLoader (Paimon jars
are
+ // not on the default application classpath, so FindClass won't work).
+ jclass local_cls = nullptr;
+ RETURN_IF_ERROR(_load_writer_class(env, &local_cls));
+ _jni_writer_cls = static_cast<jclass>(env->NewGlobalRef(local_cls));
+ env->DeleteLocalRef(local_cls);
+
+ // Step 2: Cache JNI method IDs for write, prepareCommit, abort, close.
+ jmethodID open_id = env->GetMethodID(
+ _jni_writer_cls, "open",
+
"(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;JLjava/lang/String;Z)V");
+ _write_id = env->GetMethodID(_jni_writer_cls, "write",
"(Ljava/nio/ByteBuffer;)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"));
+
+ // Step 3: Create the Java PaimonJniWriter instance.
+ 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);
+
+ // Step 4: Build Java arguments and call PaimonJniWriter.open().
+ const std::map<std::string, std::string> empty_config;
+ jstring j_serialized_table =
env->NewStringUTF(sink.serialized_table.c_str());
+ jobject j_hadoop_config =
+ _to_java_options(env, sink.__isset.hadoop_config ?
sink.hadoop_config : empty_config);
+ jstring j_commit_user = env->NewStringUTF(sink.commit_user.c_str());
+
+ jclass string_cls = env->FindClass("java/lang/String");
+ jobjectArray 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);
+ }
+
+ env->CallVoidMethod(_jni_writer_obj, open_id, j_serialized_table,
j_hadoop_config, j_cols,
+ static_cast<jlong>(sink.transaction_id), j_commit_user,
+ static_cast<jboolean>(sink.write_mode ==
TPaimonWriteMode::OVERWRITE));
+ Status st = _check_jni_exception(env, "open");
+
+ env->DeleteLocalRef(j_serialized_table);
+ env->DeleteLocalRef(j_hadoop_config);
+ env->DeleteLocalRef(j_commit_user);
+ env->DeleteLocalRef(j_cols);
+ env->DeleteLocalRef(string_cls);
+
+ if (st.ok()) {
+ _opened = true;
+ }
+ return st;
+}
+
+// Writer creation stays non-const because the backend interface also supports
future stateful FFI
+// implementations.
+Status JniPaimonWriteBackend::create_writer( //
NOLINT(readability-make-member-function-const)
+ std::unique_ptr<IPaimonWriter>* writer) {
+ DORIS_CHECK(_opened);
+ *writer = std::make_unique<JniPaimonWriter>(_jni_writer_obj, _write_id,
_prepare_commit_id,
+ _abort_id,
std::make_unique<ArrowMemoryPool<>>(),
+ _sink);
+ return Status::OK();
+}
+
+JniPaimonWriter::JniPaimonWriter(jobject jni_writer_obj, jmethodID write_id,
+ jmethodID prepare_commit_id, jmethodID
abort_id,
+ std::unique_ptr<ArrowMemoryPool<>> arrow_pool,
+ TPaimonTableSink sink)
+ : _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(std::move(sink)) {}
+
+Status JniPaimonWriter::_write_projected_block(RuntimeState* state, Block&
block) {
+ if (block.rows() == 0) {
+ return Status::OK();
+ }
+
+ // Use Thrift column_names as the authoritative schema source for both
+ // Arrow schema construction and Java-side write type derivation.
+ DORIS_CHECK(_sink.__isset.column_names);
+ DORIS_CHECK_EQ(_sink.column_names.size(), block.columns());
+ for (size_t i = 0; i < _sink.column_names.size(); ++i) {
+ block.get_by_position(i).name = _sink.column_names[i];
+ }
+
+ // Pipeline: Doris Block → Arrow Schema → Arrow RecordBatch → IPC Stream →
JNI direct buffer
+ //
+ // Step 1: Build Arrow schema from the projected Block.
+ std::shared_ptr<arrow::Schema> arrow_schema;
+ RETURN_IF_ERROR(get_arrow_schema_from_block(block, &arrow_schema,
state->timezone()));
+
+ // Step 2: Convert Doris Block columns to an Arrow RecordBatch.
+ std::shared_ptr<arrow::RecordBatch> record_batch;
+ RETURN_IF_ERROR(convert_to_arrow_batch(block, arrow_schema,
_arrow_pool.get(), &record_batch,
+ state->timezone_obj()));
+
+ // Step 3: Serialize the RecordBatch to Arrow IPC Stream format in memory.
+ auto out_stream_res = arrow::io::BufferOutputStream::Create();
Review Comment:
[P1] Allocate the IPC stream from the tracked pool
The RecordBatch above uses `_arrow_pool`, but this overload allocates the
IPC stream from Arrow's default pool. It creates another batch-sized native
buffer while the Doris block and RecordBatch are still live, yet that memory
bypasses the task/query tracker, so the query limit cannot stop concurrent
large string/complex batches from pushing the BE into process OOM. Construct
the stream with the tracked pool (and reserve the estimated size before
allocation) so the query is rejected cleanly instead.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java:
##########
@@ -0,0 +1,327 @@
+// 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.datasource.paimon;
+
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
+import org.apache.doris.datasource.mvcc.MvccUtil;
+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.TPaimonCommitMessage;
+import org.apache.doris.transaction.Transaction;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.table.InnerTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageSerializer;
+import org.apache.paimon.table.sink.InnerTableCommit;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Paimon transaction.
+ *
+ * Lifecycle:
+ * 1. beginInsert() — record target table
+ * 2. updateCommitMessages() — called multiple times as BE reports commit
data
+ * 3. commit() — deserialize all CommitMessages, call
StreamTableCommit.filterAndCommit()
+ * 4. rollback() — abort uncommitted data files
+ *
+ * CommitMessage wire format:
+ * BE ← JNI ← Java: byte[] with DPCM header (magic + version + length) +
Paimon
+ * CommitMessageSerializer payload
+ */
+public class PaimonTransaction implements Transaction {
+ private static final Logger LOG =
LogManager.getLogger(PaimonTransaction.class);
+
+ private static final int COMMIT_HEADER_SIZE = 12;
+ private static final byte[] COMMIT_MAGIC = new byte[] {'D', 'P', 'C', 'M'};
+
+ private final PaimonMetadataOps ops;
+ private final long transactionId;
+ private final String commitUser;
+ private PaimonExternalTable table;
+ private boolean committed = false;
+ private boolean overwrite = false;
+ private Map<String, String> staticPartition = new HashMap<>();
+
+ private final List<byte[]> commitPayloads = Lists.newArrayList();
+ private final Set<String> commitPayloadSet = new HashSet<>();
+
+ public PaimonTransaction(PaimonMetadataOps ops, long transactionId) {
+ Preconditions.checkArgument(transactionId > 0, "Paimon transaction id
must be positive");
+ this.ops = ops;
+ this.transactionId = transactionId;
+ this.commitUser = commitUser(transactionId);
+ }
+
+ // ────────────────────────────────────────────────────────────
+ // Transaction lifecycle
+ // ────────────────────────────────────────────────────────────
+
+ public void beginInsert(PaimonExternalTable table,
Optional<InsertCommandContext> insertCtx) {
+ this.table = table;
+ PaimonInsertCommandContext context = (PaimonInsertCommandContext)
insertCtx.orElseThrow(
+ () -> new IllegalStateException("Missing Paimon insert command
context"));
+ this.overwrite = context.isOverwrite();
+ this.staticPartition = new HashMap<>(context.getStaticPartition());
+ }
+
+ public void finishInsert(PaimonExternalTable table,
Optional<InsertCommandContext> insertCtx) {
+ }
+
+ @Override
+ public void commit() throws UserException {
+ List<byte[]> rawPayloads = snapshotPayloads();
+ if (rawPayloads.isEmpty() && !overwrite) {
+ LOG.info("Skip empty PaimonTransaction commit, txnId={}, table={}",
+ transactionId, tableName());
+ return;
+ }
+ if (table == null) {
+ throw new UserException("Missing paimon table for transaction");
+ }
+ try {
+ List<CommitMessage> allMessages = deserializePayloads(rawPayloads);
+ LOG.info("Commit PaimonTransaction, txnId={}, table={},
payloads={}, messages={}, overwrite={}",
+ transactionId, tableName(), rawPayloads.size(),
allMessages.size(), overwrite);
+ if (allMessages.isEmpty() && !overwrite) {
+ throw new RuntimeException(
+ "Paimon commit messages are empty, payloads=" +
rawPayloads.size());
+ }
+ doCommit(allMessages);
+ synchronized (this) {
+ committed = true;
+ }
+ } catch (Exception e) {
+ throw new UserException("Failed to commit paimon transaction on
FE", e);
+ }
+ }
+
+ @Override
+ public void rollback() {
+ if (isCommitted()) {
+ LOG.info("Skip rollback for committed PaimonTransaction, txnId={},
table={}",
+ transactionId, tableName());
+ return;
+ }
+ List<byte[]> rawPayloads = snapshotPayloads();
+ if (rawPayloads.isEmpty()) {
+ LOG.info("Skip empty PaimonTransaction rollback, txnId={},
table={}",
+ transactionId, tableName());
+ return;
+ }
+ if (table == null) {
+ LOG.warn("Skip PaimonTransaction rollback, table missing,
txnId={}", transactionId);
+ return;
+ }
+ try {
+ List<CommitMessage> allMessages = deserializePayloads(rawPayloads);
+ if (allMessages.isEmpty()) {
+ LOG.info("Skip PaimonTransaction rollback with empty decoded
messages, "
+ + "txnId={}, table={}", transactionId, tableName());
+ return;
+ }
+ LOG.info("Rollback PaimonTransaction, txnId={}, table={},
payloads={}, messages={}",
+ transactionId, tableName(), rawPayloads.size(),
allMessages.size());
+ doAbort(allMessages);
Review Comment:
[P1] Do not abort an outcome-unknown commit
If `filterAndCommit()` exhausts its retries after an atomic-commit
exception, `committed` is still false and
`BaseExternalTableInsertExecutor.onFail()` immediately reaches this abort.
Paimon explicitly treats that exception as outcome-unknown because the snapshot
may already be published; `abort()` then deletes the new and compacted files
referenced by that snapshot. Reconcile the commit user/identifier against the
published snapshots before aborting, and preserve the files while the outcome
remains unknown.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java:
##########
@@ -467,6 +482,23 @@ private void
setStaticPartitionToContext(UnboundIcebergTableSink<?> sink,
}
}
+ private void setStaticPartitionToContext(UnboundPaimonTableSink<?> sink,
+ PaimonInsertCommandContext insertCtx) {
+ if (sink.hasStaticPartition()) {
+ Map<String, String> staticPartitionValues = Maps.newHashMap();
+ for (Map.Entry<String, Expression> entry
+ : sink.getStaticPartitionKeyValues().entrySet()) {
+ Expression expr = entry.getValue();
+ if (!(expr instanceof Literal)) {
+ throw new AnalysisException(
+ String.format("Static partition value must be a
literal, but got: %s", expr));
+ }
+ staticPartitionValues.put(entry.getKey(), ((Literal)
expr).getStringValue());
Review Comment:
[P2] Canonicalize static partition key spelling
Paimon binding validates static partition names with a case-insensitive map,
so `PARTITION(region='east')` is accepted for a field named `Region`; this
stores the original `region` key in the transaction context. `withOverwrite()`
later passes it to Paimon, whose `RowType.getField()` and
`fieldNames.indexOf()` lookups are exact-case, so the valid Doris statement
fails during commit. Resolve every key to the canonical `Column.getName()`
during binding and reuse that map for row construction and overwrite.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java:
##########
@@ -878,6 +884,81 @@ private void
validateStaticPartition(UnboundIcebergTableSink<?> sink, IcebergExt
}
}
+ private Plan
bindPaimonTableSink(MatchingContext<UnboundPaimonTableSink<Plan>> ctx) {
+ UnboundPaimonTableSink<?> sink = ctx.root;
+ Pair<PaimonExternalDatabase, PaimonExternalTable> pair =
bind(ctx.cascadesContext, sink);
+ PaimonExternalDatabase database = pair.first;
+ PaimonExternalTable table = pair.second;
+ LogicalPlan child = ((LogicalPlan) sink.child());
+
+ Map<String, Expression> staticPartitions =
Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ staticPartitions.putAll(sink.getStaticPartitionKeyValues());
+ Set<String> staticPartitionColNames = staticPartitions.keySet();
+ if (!staticPartitionColNames.isEmpty()) {
+ Set<String> partitionColumnNames =
Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
+
partitionColumnNames.addAll(table.getPartitionColumnNames(Optional.empty()));
+ for (String columnName : staticPartitionColNames) {
+ if (!partitionColumnNames.contains(columnName)) {
+ throw new AnalysisException(String.format(
+ "Column '%s' is not a partition column of Paimon
table '%s'",
+ columnName, table.getName()));
+ }
+ Expression partitionValue = staticPartitions.get(columnName);
+ if (!(partitionValue instanceof Literal)) {
+ throw new AnalysisException(String.format(
+ "Partition value for column '%s' must be a
literal, but got: %s",
+ columnName, partitionValue));
+ }
+ }
+ }
+
+ List<Column> bindColumns;
+ if (sink.getColNames().isEmpty()) {
+ bindColumns = table.getBaseSchema(true).stream()
+ .filter(Column::isVisible)
+ .filter(column ->
!staticPartitionColNames.contains(column.getName()))
+ .collect(ImmutableList.toImmutableList());
+ } else {
+ bindColumns = sink.getColNames().stream().map(cn -> {
+ if (staticPartitionColNames.contains(cn)) {
+ throw new AnalysisException(String.format(
+ "Static partition column '%s' must not appear in
the insert column list", cn));
+ }
+ Column column = table.getColumn(cn);
+ if (column == null) {
+ throw new AnalysisException(String.format(
+ "column %s is not found in table %s", cn,
table.getName()));
+ }
+ return column;
+ }).collect(ImmutableList.toImmutableList());
+ }
+
+ if (bindColumns.size() != child.getOutput().size()) {
+ throw new AnalysisException("insert into cols should be
corresponding to the query output");
+ }
+ Map<String, NamedExpression> columnToOutput =
getSpecifiedColumnToOutput(bindColumns, child);
Review Comment:
[P1] Reject duplicate Paimon target columns
The case-insensitive map overwrites an earlier occurrence, but the coercion
project iterates `writeColumns` and re-emits the surviving slot for every
duplicate, so the sink's column/output size check still passes. Java then maps
both names to one table index, leaves the omitted field null, and treats the
write as full solely because the list length equals the table field count; on a
PK table this bypasses the partial-write guard and can commit the wrong row.
Reject duplicate target names before building this map, and validate unique
full-schema coverage in `PaimonWriteSchema`.
--
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]
