Gabriel39 commented on code in PR #65381: URL: https://github.com/apache/doris/pull/65381#discussion_r3601623621
########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonCommitCodec.java: ########## @@ -0,0 +1,114 @@ +// 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.paimon.io.DataOutputSerializer; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageSerializer; + +import java.util.ArrayList; +import java.util.List; + +/** + * Encodes Paimon commit messages into the DPCM (Doris-Paimon Commit Message) payload + * format forwarded to FE. + * + * <h3>DPCM framing format</h3> + * Each payload is framed as: + * <pre> + * ┌──────────┬─────────────┬────────────┬──────────────────────┐ + * │ Magic (4)│ Version (4) │ Length (4) │ Serialized Messages │ + * │ "DPCM" │ big-endian │ big-endian │ (varies) │ + * └──────────┴─────────────┴────────────┴──────────────────────┘ + * </pre> + * + * <p>Messages are serialized using Paimon's {@link CommitMessageSerializer} and + * split into chunks if the serialized payload exceeds {@link #MAX_PAYLOAD_BYTES} + * (8 MiB). Chunk size starts at {@link #DEFAULT_CHUNK_SIZE} (512 messages) and + * is halved adaptively until each chunk fits within the size limit. + */ +final class PaimonCommitCodec { + static final int HEADER_BYTES = 12; + /** Maximum payload size per chunk (8 MiB) to stay under gRPC message limits. */ + static final int MAX_PAYLOAD_BYTES = 8 * 1024 * 1024; + /** Starting number of commit messages per chunk. */ + static final int DEFAULT_CHUNK_SIZE = 512; + + private final CommitMessageSerializer serializer = new CommitMessageSerializer(); + + /** + * Encode commit messages into DPCM-framed byte chunks. + * + * @param messages Paimon commit messages from {@code prepareCommit()} + * @return byte[][] where each element is a complete DPCM-framed chunk + */ + byte[][] encode(List<CommitMessage> messages) throws Exception { + if (messages.isEmpty()) { + return new byte[0][]; + } + + // Adaptive chunking: start with DEFAULT_CHUNK_SIZE messages per chunk; + // if the serialized chunk exceeds MAX_PAYLOAD_BYTES, halve the chunk + // size and retry until each chunk fits or chunkSize reaches 1. + int chunkSize = DEFAULT_CHUNK_SIZE; + List<byte[]> payloads = new ArrayList<>(); + int offset = 0; + while (offset < messages.size()) { + int end = Math.min(offset + chunkSize, messages.size()); + byte[] payload = encodeChunk(messages.subList(offset, end)); + if (payload.length > MAX_PAYLOAD_BYTES && chunkSize > 1) { Review Comment: [P2] The 8 MiB invariant is not enforced when a single serialized `CommitMessage` is already larger than the limit. Once `chunkSize` reaches 1, this condition is false, the oversized payload is appended, and it is forwarded through the BE-to-FE report path anyway. A single bucket/writer can accumulate enough file additions or deletions to hit this case. Please either define a protocol that can segment one large commit message safely, or fail explicitly before RPC with a clear size diagnostic and make sure the error path aborts the prepared files. Add a test where one encoded message itself exceeds `MAX_PAYLOAD_BYTES`; the current tests only cover framing and an empty list. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,370 @@ +// 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.TableWriteImpl; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.InputStream; +import java.nio.ByteBuffer; +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 BatchTableWrite writer; + private TableWriteImpl<?> tableWrite; + private IOManager ioManager; + private long commitIdentifier; + + public PaimonJniWriter() { + this.allocator = new RootAllocator(Long.MAX_VALUE); + this.classLoader = this.getClass().getClassLoader(); + configureLogLevels(); Review Comment: [P2] Constructing a writer mutates the logger level for the entire `org.apache.paimon` namespace. This is a process/classloader-wide side effect, not writer-local configuration, so opening one write can suppress INFO diagnostics for Paimon scans and other writers; doing the reflective reconfiguration for every LocalState is also unnecessary. Please move this to the connector's static logging configuration (or a single guarded initialization) and avoid changing the whole Paimon namespace from a per-query object. ########## 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(); Review Comment: [P2] All recovery-critical state—target table, overwrite/static-partition mode, and the commit payloads—is held only in FE memory. `AbstractExternalTransactionManager` and `GlobalExternalTransactionInfoMgr` are also in-memory maps. If the coordinating FE fails after BEs have prepared files (or after some payloads have arrived), the new leader cannot reconstruct this transaction to commit or abort it, leaving prepared files orphaned. Please define the supported failover semantics for Paimon writes and add a recovery/cleanup mechanism: persist the transaction metadata and payloads, or make BE-side cleanup/retry independent of the original FE. At minimum, add a failover test around the post-prepare/pre-commit window and document any intentional limitation. -- 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]
