Gabriel39 commented on code in PR #65381: URL: https://github.com/apache/doris/pull/65381#discussion_r3601570604
########## 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: [P1] The stated default-value behavior is not true for the current writer setup. Missing fields are expanded to null in a full-schema `GenericRow`, and `openFileStoreWriter()` does not call `withWriteType(partialRowType)`. In Paimon 1.3.1, `TableWriteImpl.writeAndReturn()` calls `checkNullability(row)` before `wrapDefaultValue(row)`. Thus an omitted `NOT NULL DEFAULT` field throws `Cannot write null to non-null column(...)` instead of applying its default. Please preserve the distinction between an omitted field and an explicit NULL, then either use Paimon's partial write-type contract or materialize defaults before constructing the full row. A regression case such as `INSERT INTO t(id) VALUES (1)` with another `NOT NULL DEFAULT` column is needed. ########## 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: [P1] This does not discard already-written data files; it only closes the writer. On Paimon 1.3.1, `TableWriteImpl.close()` delegates to `FileStoreWrite.close()`. The concrete close path closes record writers and clears in-memory maps, while file cleanup is exposed separately as `TableCommit.abort(List<CommitMessage>)`. The C++ error path calls this method without first preparing or forwarding commit messages, and FE rollback skips when its payload list is empty. Therefore, if a file has already rolled/spilled and a later row fails or the query is cancelled, that file can remain orphaned even though no snapshot is published. Please collect cleanup commit messages and invoke `InnerTableCommit.abort(messages)` in an authenticated context, or forward cleanup messages to FE on the error path. Add a failure-injection test that verifies the object-store files, not only table rows and snapshot count. ########## 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); Review Comment: [P1] Each pipeline LocalState creates one of these writers, so `Long.MAX_VALUE` makes every Java Arrow allocator effectively unbounded and independent of Doris query/workload-group memory accounting. For one block the peak includes the native Block, Arrow IPC buffer, Java Arrow vectors, a fully boxed `Object[][]`, per-row `GenericRow` objects, and the Paimon write buffer. `parallel_pipeline_task_num` multiplies this by the number of local writers; spill only covers the Paimon buffer, not Arrow decoding or boxed copies. Please derive a hard allocator/batch limit from the query budget and bridge Java/Paimon reserve/release to Doris memory accounting. The full-batch `Object[][]` materialization should also be avoided or bounded. -- 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]
