github-actions[bot] commented on code in PR #65381: URL: https://github.com/apache/doris/pull/65381#discussion_r3612030697
########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java: ########## @@ -0,0 +1,359 @@ +// 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 commitOutcomeUnknown = 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()); + } + synchronized (this) { + commitOutcomeUnknown = true; Review Comment: [P1] Keep rollback enabled until the commit can actually be outcome-unknown This flag is set before any catalog/authenticator/table/committer work. If both attempts fail in `ops.getCatalog()`, `getInnerTable()`, `newCommit()`, or another path that never reaches `filterAndCommit()`, the outcome is definitely unpublished; nevertheless `onFail()` reaches `rollback()`, which now skips `abort()` and leaves every prepared data/compaction file behind. Paimon itself distinguishes and cleans preparation failures, reserving the no-cleanup case for an exception at the atomic snapshot commit. Set/retain `UNKNOWN` only after crossing that boundary (or after reconciliation cannot determine visibility), while keeping rollback enabled for definite pre-commit failures; add a failure-injection test for that path. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java: ########## @@ -0,0 +1,359 @@ +// 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 commitOutcomeUnknown = 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; Review Comment: [P1] Pin the table used to prepare files for the whole transaction The sink serializes one concrete `FileStoreTable` to BE, but this transaction retains only the mutable Doris wrapper. Commit/abort later call `getPaimonTable()` again and use the catalog's current authenticator. After the target read lock is released, a cache invalidation/expiry or concurrent catalog reset can therefore make BE prepare files for table/location A while FE commits or aborts through a reloaded table/auth context B; that can fail after prepare or address the wrong storage target. Capture the same concrete `InnerTable` and matching auth context when the write is bound, keep them valid through commit/abort, and add a test that resets or alters the catalog after prepare. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,382 @@ +// 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.paimon; + +import org.apache.doris.common.security.authentication.PreExecutionAuthenticator; +import org.apache.doris.common.security.authentication.PreExecutionAuthenticatorCache; + +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.ipc.ArrowStreamReader; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.disk.IOManagerImpl; +import org.apache.paimon.memory.HeapMemorySegmentPool; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.InnerTableCommit; +import org.apache.paimon.table.sink.TableWriteImpl; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * JNI entry point for Paimon write operations. + * + * <p>Called from C++ ({@code JniPaimonWriter}) via JNI. One instance per BE pipeline + * fragment (one per {@code VPaimonTableWriter}). Data path: + * + * <pre> + * C++ Block → Arrow IPC Stream → JNI direct ByteBuffer + * → PaimonJniWriter.write(directBuffer) + * → ArrowStreamReader → VectorSchemaRoot + * → PaimonArrowConverter (column-major typed extraction) + * → PaimonWriteSchema.tableRow() (canonical table-schema order) + * → BatchTableWrite.write(row) (SDK-owned routing and buffering) + * </pre> + * + * <p>Commit path: + * + * <pre> + * VPaimonTableWriter::close() → JNI → PaimonJniWriter.prepareCommit() + * → BatchTableWrite.prepareCommit() + * → PaimonCommitCodec.encode() → DPCM-framed byte[][] + * → C++ collects TPaimonCommitMessage[] → RPC to FE → PaimonTransaction + * </pre> + */ +public class PaimonJniWriter { + private static final Logger LOG = LoggerFactory.getLogger(PaimonJniWriter.class); + + private final BufferAllocator allocator; + private final ClassLoader classLoader; + private final PaimonArrowConverter arrowConverter = new PaimonArrowConverter(); + private final PaimonCommitCodec commitCodec = new PaimonCommitCodec(); + + private PreExecutionAuthenticator preExecutionAuthenticator; + + private PaimonWriteSchema writeSchema; + private FileStoreTable table; + private BatchTableWrite writer; + private TableWriteImpl<?> tableWrite; + private IOManager ioManager; + private long commitIdentifier; + private String commitUser; + private List<CommitMessage> preparedCommitMessages = Collections.emptyList(); + + public PaimonJniWriter() { + this.allocator = new RootAllocator(Long.MAX_VALUE); + this.classLoader = this.getClass().getClassLoader(); + } + + // ──────────────────────────────────────────────────────────── + // JNI entry points (called from C++) + // ──────────────────────────────────────────────────────────── + + /** + * Initialize the writer. Called once per BE pipeline fragment via JNI. + * + * <p>This method: + * <ol> + * <li>Deserializes the target Paimon {@link FileStoreTable} selected by FE.</li> + * <li>Creates a {@link PaimonWriteSchema} which normalizes Doris input + * columns to the table-schema row layout.</li> + * <li>Opens one Paimon SDK writer session.</li> + * </ol> + * + * @param serializedTable serialized Paimon table selected by FE + * @param hadoopConfig filesystem and authentication configuration + * @param columnNames output column names in the order produced by BE + * @param transactionId Doris external transaction identifier + * @param commitUser Paimon commit user shared with the FE committer + * @param overwrite whether this is an overwrite write + */ + public void open(String serializedTable, Map<String, String> hadoopConfig, + String[] columnNames, long transactionId, String commitUser, + boolean overwrite) throws Exception { + Thread.currentThread().setContextClassLoader(classLoader); + + this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(hadoopConfig); + preExecutionAuthenticator.execute(() -> { + try { + FileStoreTable table = PaimonUtils.deserialize(serializedTable); + LOG.info("PaimonJniWriter opening: table={}, columns={}", + table.fullName(), columnNames != null ? columnNames.length : 0); + this.commitIdentifier = transactionId; + this.table = table; + this.commitUser = commitUser; + + this.writeSchema = PaimonWriteSchema.create(table.rowType(), columnNames); + if (writeSchema.isPartial() && !table.primaryKeys().isEmpty()) { + throw new UnsupportedOperationException( + "Paimon primary-key write requires all table columns"); + } + openFileStoreWriter(table, commitUser, overwrite); + return null; + } catch (Throwable t) { + throw new RuntimeException("PaimonJniWriter open failed", t); + } + }); + } + + /** + * Write a batch of rows from an Arrow IPC Stream buffer. + * + * <p>Called from C++ {@code JniPaimonWriter::_write_projected_block()} + * once per Block. The buffer is a zero-copy direct view of the native + * Arrow IPC Stream bytes. Rows are deserialized, normalized to table-schema + * order, and handed to Paimon's high-level {@code write(row)} API. The SDK + * owns partition/bucket routing, buffering, spill, and file rolling. + * + * @param directBuffer direct view of the native Arrow IPC Stream bytes (no copy) + */ + public void write(ByteBuffer directBuffer) throws Exception { + preExecutionAuthenticator.execute(() -> { + try { + try (ArrowStreamReader reader = new ArrowStreamReader( + new DirectBufInputStream(directBuffer), allocator)) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + while (reader.loadNextBatch()) { + writeBatch(root); + } + } + return null; + } catch (Throwable t) { + throw new RuntimeException("PaimonJniWriter write failed: bytes=" + + directBuffer.capacity(), t); + } + }); + } + + /** + * Prepare commit: flush all in-memory data, close files, and serialize commit + * messages for the FE coordinator. + * + * <p>Flushes and collects Paimon {@link CommitMessage}s, then encodes them via + * {@link PaimonCommitCodec} into DPCM-framed byte chunks that are forwarded to + * FE through the BE. + * + * @return byte[][] each element is a DPCM-framed serialized CommitMessage chunk + */ + public byte[][] prepareCommit() throws Exception { + return preExecutionAuthenticator.execute(() -> { + try { + List<CommitMessage> messages = prepareCommitMessages(); + if (messages.isEmpty()) { + LOG.info("PaimonJniWriter prepareCommit: empty"); + return new byte[0][]; + } + LOG.info("PaimonJniWriter prepareCommit: {} messages", messages.size()); + return commitCodec.encode(messages); + } catch (Throwable t) { + throw new RuntimeException("PaimonJniWriter prepareCommit failed", t); + } + }); + } + + /** + * Abort: discard all written data files and close the SDK writer. + * Called from C++ when write or prepareCommit fails. + */ + public void abort() throws Exception { + try { + if (preExecutionAuthenticator != null) { + preExecutionAuthenticator.execute(() -> { + abortWriter(); + return null; + }); + } else { + abortWriter(); + } + } catch (Exception e) { + LOG.error("PaimonJniWriter abort failed", e); + throw e; + } + } + + /** + * Close: release all resources. + */ + public void close() throws Exception { + try { + if (preExecutionAuthenticator != null) { + preExecutionAuthenticator.execute(() -> { + closeResources(); + return null; + }); + } else { + closeResources(); + } + } catch (Exception e) { + LOG.warn("PaimonJniWriter close error", e); + throw e; + } + } + + // ──────────────────────────────────────────────────────────── + // Initialization helpers + // ──────────────────────────────────────────────────────────── + + private void openFileStoreWriter(FileStoreTable table, String commitUser, boolean overwrite) + throws Exception { + CoreOptions coreOptions = CoreOptions.fromMap(table.options()); + tableWrite = table.newWrite(commitUser); + if (overwrite) { + tableWrite.withIgnorePreviousFiles(true); + } + writer = tableWrite; + openSpillResources(coreOptions); + } + + private void openSpillResources(CoreOptions coreOptions) throws Exception { + if (!coreOptions.writeBufferSpillable()) { + return; + } + String spillDirectory = System.getProperty("java.io.tmpdir"); + ioManager = new IOManagerImpl(spillDirectory); + HeapMemorySegmentPool memorySegmentPool = new HeapMemorySegmentPool( + coreOptions.writeBufferSize(), coreOptions.pageSize()); + writer.withIOManager(ioManager).withMemoryPool(memorySegmentPool); + LOG.info("Paimon writer spill enabled: dir={}", spillDirectory); + } + + // ──────────────────────────────────────────────────────────── + // Data writing + // ──────────────────────────────────────────────────────────── + + private void writeBatch(VectorSchemaRoot root) throws Exception { + int rowCount = root.getRowCount(); + if (rowCount == 0) { + return; + } + // Extract values in Doris input order, then normalize every row to the + // full table schema before calling the SDK's high-level write API. + Object[][] columnValues = arrowConverter.convert(root, writeSchema.targetTypes()); + BatchTableWrite currentWriter = requireWriter(); + for (int r = 0; r < rowCount; r++) { + currentWriter.write(writeSchema.tableRow(columnValues, r)); Review Comment: [P1] Reject dynamic-bucket tables until a bucket assigner is implemented For a primary-key Paimon table configured with `bucket=-1`, this bucket-less overload fails on the first row. Paimon 1.3.1 selects `DynamicBucketRowKeyExtractor`, whose `bucket()` deliberately throws and requires the caller to assign a bucket and invoke `write(row, bucket)`. Doris currently has neither that assignment/bootstrap protocol nor an analysis/open-time rejection, so a supported Paimon table cannot be written at all. Reject dynamic-bucket sinks with a clear unsupported-mode error until the assignment plus bucket-to-writer ownership is implemented, and add a `bucket=-1` PK regression. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java: ########## @@ -467,6 +483,25 @@ private void setStaticPartitionToContext(UnboundIcebergTableSink<?> sink, } } + private void setStaticPartitionToContext(UnboundPaimonTableSink<?> sink, + PaimonExternalTable table, 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)); + } + Column partitionColumn = Preconditions.checkNotNull(table.getColumn(entry.getKey()), + "Paimon partition column was validated during analysis"); + staticPartitionValues.put(partitionColumn.getName(), ((Literal) expr).getStringValue()); Review Comment: [P1] Encode null and blank static values as Paimon's default partition `NullLiteral.getStringValue()` produces the ordinary string `"null"`, but Paimon writes an actual null (and blank/whitespace string values) under the table's configurable `partition.default-name`. On a table with `dynamic-partition-overwrite=false`, an empty `PARTITION(region=NULL)` overwrite consequently targets the literal `region='null'` partition—potentially deleting it—while leaving the real null partition visible; with replacement rows, the writer emits the default partition and Paimon's static-overwrite sanity check rejects it as outside this predicate. Build the spec from typed partition semantics and encode null/blank values with the table's actual default-partition name. Cover both a null and a literal `"null"` partition, including a custom default name. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java: ########## @@ -878,6 +884,88 @@ 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()); Review Comment: [P2] Reject case-insensitive duplicate static partition keys before normalizing The parser rejects only exact-spelling duplicates, so `PARTITION(region='east', REGION='west')` reaches this `TreeMap`. Because its comparator is case-insensitive, `putAll()` silently collapses the two entries and chooses one value, making an ambiguous statement overwrite a partition different from what one clause requests. Detect duplicates using Doris's case-insensitive identifier rules before inserting into the normalized map (and add the mixed-case duplicate case to analysis coverage). ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java: ########## @@ -0,0 +1,359 @@ +// 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 commitOutcomeUnknown = 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()); + } + synchronized (this) { + commitOutcomeUnknown = true; + } + doCommitWithReconciliation(allMessages); + synchronized (this) { + committed = true; + commitOutcomeUnknown = false; + } + } 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; + } + if (isCommitOutcomeUnknown()) { + LOG.warn("Skip rollback for PaimonTransaction with unknown commit outcome, " + + "txnId={}, table={}. Preserving data files for snapshot safety", + 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); + } catch (Exception e) { + LOG.warn("Failed to rollback PaimonTransaction, txnId={}, table={}", + transactionId, tableName(), e); + } + } + + // ──────────────────────────────────────────────────────────── + // CommitMessage collection (called from Coordinator) + // ──────────────────────────────────────────────────────────── + + public void updateCommitMessages(List<TPaimonCommitMessage> messages) { + if (messages == null || messages.isEmpty()) { + return; + } + synchronized (this) { + for (TPaimonCommitMessage msg : messages) { + addPayload(msg); + } + } + } + + private void addPayload(TPaimonCommitMessage message) { + if (message == null || !message.isSetPayload()) { + return; + } + byte[] payload = message.getPayload(); + if (payload == null || payload.length == 0) { + return; + } + String key = Base64.getEncoder().encodeToString(payload); + if (commitPayloadSet.add(key)) { + commitPayloads.add(Arrays.copyOf(payload, payload.length)); + } + } + + // ──────────────────────────────────────────────────────────── + // Transaction identity + // ──────────────────────────────────────────────────────────── + + public static String commitUser(long transactionId) { + return "doris_txn_" + transactionId; + } + + public String getCommitUser() { + return commitUser; + } + + public long getTransactionId() { + return transactionId; + } + + boolean isCommitted() { + synchronized (this) { + return committed; + } + } + + boolean isCommitOutcomeUnknown() { + synchronized (this) { + return commitOutcomeUnknown; + } + } + + int getPayloadCount() { + synchronized (this) { + return commitPayloads.size(); + } + } + + // ──────────────────────────────────────────────────────────── + // Internal commit logic + // ──────────────────────────────────────────────────────────── + + private void doCommit(List<CommitMessage> msgs) throws Exception { + ops.getCatalog(); + ExecutionAuthenticator authenticator = ops.dorisCatalog.getExecutionAuthenticator(); + authenticator.execute(() -> { + InnerTable paimonTable = getInnerTable(); + InnerTableCommit committer = paimonTable.newCommit(commitUser); + try { + if (overwrite) { + committer.withOverwrite(staticPartition); Review Comment: [P1] Disable dynamic overwrite semantics for a static partition spec `withOverwrite(staticPartition)` does not select static semantics by itself. Paimon 1.3.1 defaults `dynamic-partition-overwrite=true`; in that branch it ignores this predicate, targets only partitions present in appended files, and skips the overwrite entirely when there are no appended files. Thus an empty `PARTITION(region='east')` overwrite leaves `east` unchanged, and a partial multi-level spec such as `PARTITION(pt0=1)` leaves stale `(1,B)` data when the replacement rows contain only `(1,A)`. Configure/copy the table with dynamic partition overwrite disabled whenever Doris has a static spec (while preserving dynamic mode for all-dynamic overwrite), and cover both an empty static overwrite and a partial multi-level spec. -- 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]
