suxiaogang223 commented on code in PR #65381: URL: https://github.com/apache/doris/pull/65381#discussion_r3611908447
########## 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(); + } + + // ──────────────────────────────────────────────────────────── + // 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.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(() -> { + closeWriter(); Review Comment: Fixed in bc98f94c3e8. `PaimonJniWriter` now retains the authenticated table/user context and prepared commit messages, prepares cleanup messages on the abort path when necessary, and invokes `InnerTableCommit.abort(messages)` inside the authenticated context before closing the writer. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java: ########## @@ -0,0 +1,102 @@ +// 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.data.GenericRow; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.RowType; + +/** + * Immutable input-column metadata shared by all batches of one writer. + * + * <p>Doris may produce a subset or permutation of the Paimon table columns. The + * SDK {@code write(row)} API, including its partition and bucket extractors, + * expects rows in table-schema order. This class resolves the input columns once + * and expands every Arrow row to that canonical layout. + * + * <h3>Partial-write support</h3> + * Missing columns remain null in the expanded row. Paimon then applies its normal Review Comment: Fixed in bc98f94c3e8. Omitted columns are now distinguished from explicit NULLs and Paimon defaults are materialized with `DefaultValueUtils` before constructing the full row. Added a focused `PaimonWriteSchemaTest` for an omitted `NOT NULL DEFAULT` column. ########## 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: Fixed in bc98f94c3e8. `PaimonCommitCodec` now validates the final encoded chunk and fails explicitly with `IOException` when one DPCM payload exceeds `MAX_PAYLOAD_BYTES`. The test uses a real oversized `CommitMessageImpl` through the full codec path, and the writer abort path now cleans prepared messages. ########## 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: Fixed in bc98f94c3e8. Removed the per-writer mutation of the global `org.apache.paimon` logger level, so constructing a write no longer changes process-wide logging behavior. -- 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]
