github-actions[bot] commented on code in PR #65381: URL: https://github.com/apache/doris/pull/65381#discussion_r3613980870
########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonArrowConverter.java: ########## @@ -0,0 +1,369 @@ +// 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.arrow.vector.BigIntVector; +import org.apache.arrow.vector.BitVector; +import org.apache.arrow.vector.DateDayVector; +import org.apache.arrow.vector.DecimalVector; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.Float4Vector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.IntVector; +import org.apache.arrow.vector.SmallIntVector; +import org.apache.arrow.vector.TimeStampVector; +import org.apache.arrow.vector.TinyIntVector; +import org.apache.arrow.vector.VarBinaryVector; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Decimal; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericMap; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.Timestamp; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.BinaryType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.LocalZonedTimestampType; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.RowType; +import org.apache.paimon.types.TimestampType; +import org.apache.paimon.types.VarBinaryType; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Converts Arrow columns into Paimon internal values without owning writer state. */ +final class PaimonArrowConverter { + + Object[][] convert(VectorSchemaRoot root, DataType[] targetTypes) { + List<Field> fields = root.getSchema().getFields(); + List<FieldVector> vectors = root.getFieldVectors(); + if (fields.size() != targetTypes.length) { + throw new IllegalArgumentException("Arrow column count does not match Paimon write type"); + } + + Object[][] columnValues = new Object[fields.size()][]; + for (int column = 0; column < fields.size(); column++) { + columnValues[column] = extractColumnValues( + vectors.get(column), fields.get(column), targetTypes[column], root.getRowCount()); + } + return columnValues; + } + + private Object[] extractColumnValues(FieldVector vector, Field arrowField, + DataType targetType, int rowCount) { + Object[] values = new Object[rowCount]; + + if (vector instanceof IntVector) { + IntVector intVector = (IntVector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = intVector.isNull(i) ? null : intVector.get(i); + } + return values; + } + if (vector instanceof BigIntVector) { + BigIntVector bigIntVector = (BigIntVector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = bigIntVector.isNull(i) ? null : bigIntVector.get(i); + } + return values; + } + if (vector instanceof SmallIntVector) { + SmallIntVector smallIntVector = (SmallIntVector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = smallIntVector.isNull(i) ? null : smallIntVector.get(i); + } + return values; + } + if (vector instanceof TinyIntVector) { + TinyIntVector tinyIntVector = (TinyIntVector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = tinyIntVector.isNull(i) ? null : tinyIntVector.get(i); + } + return values; + } + if (vector instanceof Float4Vector) { + Float4Vector floatVector = (Float4Vector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = floatVector.isNull(i) ? null : floatVector.get(i); + } + return values; + } + if (vector instanceof Float8Vector) { + Float8Vector doubleVector = (Float8Vector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = doubleVector.isNull(i) ? null : doubleVector.get(i); + } + return values; + } + if (vector instanceof BitVector) { + BitVector bitVector = (BitVector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = bitVector.isNull(i) ? null : bitVector.get(i) == 1; + } + return values; + } + if (vector instanceof DateDayVector) { + DateDayVector dateVector = (DateDayVector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = dateVector.isNull(i) ? null : dateVector.get(i); + } + return values; + } + if (vector instanceof VarCharVector) { + VarCharVector stringVector = (VarCharVector) vector; + boolean binary = targetType instanceof BinaryType || targetType instanceof VarBinaryType; + for (int i = 0; i < rowCount; i++) { + values[i] = stringVector.isNull(i) ? null + : binary ? stringVector.get(i) : BinaryString.fromBytes(stringVector.get(i)); + } + return values; + } + if (vector instanceof VarBinaryVector) { + VarBinaryVector binaryVector = (VarBinaryVector) vector; + for (int i = 0; i < rowCount; i++) { + values[i] = binaryVector.isNull(i) ? null : binaryVector.get(i); + } + return values; + } + if (vector instanceof TimeStampVector) { + TimeStampVector timestampVector = (TimeStampVector) vector; + ArrowType.Timestamp timestampType = (ArrowType.Timestamp) arrowField.getType(); + for (int i = 0; i < rowCount; i++) { + values[i] = timestampVector.isNull(i) ? null : toPaimonTimestamp( + arrowTimestampToMicros(timestampVector.get(i), timestampType), + timestampType, targetType); + } + return values; + } + if (vector instanceof DecimalVector) { + DecimalVector decimalVector = (DecimalVector) vector; + int precision = decimalVector.getPrecision(); + int scale = decimalVector.getScale(); + for (int i = 0; i < rowCount; i++) { + if (decimalVector.isNull(i)) { + values[i] = null; + } else { + BigDecimal decimal = getBigDecimalFromArrowBuf( + decimalVector.getDataBuffer(), i, scale, DecimalVector.TYPE_WIDTH); + values[i] = Decimal.fromBigDecimal(decimal, precision, scale); + } + } + return values; + } + + for (int i = 0; i < rowCount; i++) { + values[i] = vector.isNull(i) ? null + : convertToPaimonType(vector.getObject(i), arrowField, targetType); + } + return values; + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private Object convertToPaimonType(Object value, Field arrowField, DataType targetType) { + if (value == null) { + return null; + } + if (targetType instanceof BinaryType || targetType instanceof VarBinaryType) { + if (value instanceof byte[]) { + return value; + } + if (value instanceof BinaryString) { + return ((BinaryString) value).toBytes(); + } + if (value instanceof org.apache.arrow.vector.util.Text) { + return ((org.apache.arrow.vector.util.Text) value).copyBytes(); + } + if (value instanceof String) { + return ((String) value).getBytes(StandardCharsets.UTF_8); + } + return value.toString().getBytes(StandardCharsets.UTF_8); + } + if (value instanceof BinaryString) { + return value; + } + if (value instanceof byte[]) { + return BinaryString.fromBytes((byte[]) value); + } + if (value instanceof org.apache.arrow.vector.util.Text) { + return BinaryString.fromBytes(((org.apache.arrow.vector.util.Text) value).copyBytes()); + } + if (value instanceof org.apache.hadoop.io.Text) { + org.apache.hadoop.io.Text text = (org.apache.hadoop.io.Text) value; + return BinaryString.fromBytes(text.getBytes(), 0, text.getLength()); + } + if (value instanceof CharSequence) { + return BinaryString.fromString(value.toString()); + } + + ArrowType.ArrowTypeID typeId = arrowField == null + ? null : arrowField.getType().getTypeID(); + if (value instanceof LocalDateTime) { + return Timestamp.fromLocalDateTime((LocalDateTime) value); + } + if (value instanceof Long && typeId == ArrowType.ArrowTypeID.Timestamp) { + ArrowType.Timestamp timestampType = (ArrowType.Timestamp) arrowField.getType(); + return toPaimonTimestamp( + arrowTimestampToMicros((Long) value, timestampType), timestampType, targetType); + } + if (value instanceof Integer && typeId == ArrowType.ArrowTypeID.Date) { + return value; + } + if (value instanceof java.time.LocalDate) { + return (int) ((java.time.LocalDate) value).toEpochDay(); + } + if (value instanceof BigDecimal) { + BigDecimal decimal = (BigDecimal) value; + return Decimal.fromBigDecimal(decimal, decimal.precision(), decimal.scale()); + } + if (targetType instanceof RowType && value instanceof Map) { + return convertStruct((Map<?, ?>) value, (RowType) targetType, arrowField); + } + if (targetType instanceof MapType && value instanceof List) { + return convertMap((List<?>) value, (MapType) targetType, arrowField); + } + if (targetType instanceof ArrayType && value instanceof List) { + return convertArray((List<?>) value, (ArrayType) targetType, arrowField); + } + return value; + } + + private GenericRow convertStruct(Map<?, ?> mapValue, RowType rowType, Field arrowField) { + List<DataField> childFields = rowType.getFields(); + GenericRow row = new GenericRow(childFields.size()); + for (int i = 0; i < childFields.size(); i++) { + DataField childField = childFields.get(i); + row.setField(i, convertToPaimonType(mapValue.get(childField.name()), + findChildField(arrowField, childField.name()), childField.type())); + } + return row; + } + + private GenericMap convertMap(List<?> values, MapType mapType, Field arrowField) { + Field keyField = null; + Field valueField = null; + if (arrowField != null && !arrowField.getChildren().isEmpty()) { + Field entries = arrowField.getChildren().get(0); + if (entries.getChildren().size() >= 2) { + keyField = entries.getChildren().get(0); + valueField = entries.getChildren().get(1); + } + } + String keyName = keyField == null ? "key" : keyField.getName(); + String valueName = valueField == null ? "value" : valueField.getName(); + Map<Object, Object> converted = new HashMap<>(); + for (Object element : values) { + if (element instanceof Map) { + Map<?, ?> entry = (Map<?, ?>) element; + converted.put( + convertToPaimonType(entry.get(keyName), keyField, mapType.getKeyType()), + convertToPaimonType(entry.get(valueName), valueField, + mapType.getValueType())); + } + } + return new GenericMap(converted); + } + + private GenericArray convertArray(List<?> values, ArrayType arrayType, Field arrowField) { + Object[] converted = new Object[values.size()]; + Field elementField = arrowField == null || arrowField.getChildren().isEmpty() + ? null : arrowField.getChildren().get(0); + for (int i = 0; i < values.size(); i++) { + converted[i] = convertToPaimonType( + values.get(i), elementField, arrayType.getElementType()); + } + return new GenericArray(converted); + } + + private static Field findChildField(Field parent, String name) { + if (parent == null || parent.getChildren() == null) { + return null; + } + for (Field child : parent.getChildren()) { + if (child.getName().equals(name)) { + return child; + } + } + return null; + } + + private static BigDecimal getBigDecimalFromArrowBuf( + org.apache.arrow.memory.ArrowBuf buffer, int index, int scale, int byteWidth) { + byte[] value = new byte[byteWidth]; + buffer.getBytes((long) index * byteWidth, value, 0, byteWidth); + if (ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN) { + for (int i = 0; i < byteWidth / 2; i++) { + byte temporary = value[i]; + int opposite = byteWidth - 1 - i; + value[i] = value[opposite]; + value[opposite] = temporary; + } + } + return new BigDecimal(new BigInteger(value), scale); + } + + private static long arrowTimestampToMicros( + long value, ArrowType.Timestamp timestampType) { + switch (timestampType.getUnit()) { + case SECOND: + return Math.multiplyExact(value, 1_000_000L); + case MILLISECOND: + return Math.multiplyExact(value, 1_000L); + case MICROSECOND: + return value; + case NANOSECOND: + return Math.floorDiv(value, 1_000L); + default: + throw new IllegalArgumentException( + "Unsupported Arrow timestamp unit: " + timestampType.getUnit()); + } + } + + static Timestamp toPaimonTimestamp(long micros, ArrowType.Timestamp arrowType, + DataType targetType) { + if (targetType instanceof LocalZonedTimestampType) { + return Timestamp.fromMicros(micros); + } + if (targetType instanceof TimestampType) { + String timezone = arrowType.getTimezone(); + if (timezone == null || timezone.isEmpty()) { + return Timestamp.fromMicros(micros); + } + long epochSecond = Math.floorDiv(micros, 1_000_000L); + long microsOfSecond = Math.floorMod(micros, 1_000_000L); + Instant instant = Instant.ofEpochSecond(epochSecond, microsOfSecond * 1_000L); + return Timestamp.fromLocalDateTime( + LocalDateTime.ofInstant(instant, ZoneId.of(timezone))); Review Comment: [P1] Preserve NTZ values without an instant round-trip BE serializes every `DATETIMEV2` as a zone-bearing Arrow epoch, and this branch converts that epoch back to a Paimon `TIMESTAMP WITHOUT TIME ZONE`. That mapping is not invertible: with `America/Los_Angeles`, the valid NTZ value `2024-03-10 02:30:00` is in the DST gap, so cctz maps it to the transition instant and this code commits `03:00:00` instead. It also calls bare `ZoneId.of(timezone)`, so a Doris-supported retained alias such as `CST` fails here. Preserve NTZ as civil fields (and reserve instant semantics for `LocalZonedTimestampType`), then cover a DST-gap value and a short-zone alias. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,387 @@ +// 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.BucketMode; +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); Review Comment: [P2] Restore the pooled worker's context ClassLoader This runs on a reusable `FragmentMgrAsyncWorkThreadPool` worker and permanently replaces that Java thread's context ClassLoader. After Paimon close, the native/JVM thread returns to the shared pool with the connector loader still installed, so a later JNI task can resolve service providers or resources through the wrong extension. Save the previous loader and restore it unconditionally after the whole writer lifecycle, including open/close failures; `ThreadClassLoaderContext` and `JdbcJniWriter.open()` show the expected restoration pattern. ########## 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); Review Comment: [P1] Reject partition-list forms instead of dropping them `UnboundPaimonTableSink` still carries a bare `PARTITION (name, ...)` list, but this binder consumes only the separate key/value static map and the bound sink no longer carries that list. The transaction consequently calls `withOverwrite(emptyMap)`. On a table with `dynamic-partition-overwrite=false`, Paimon interprets that as a full-table overwrite, so an explicitly dynamic-partitioned statement can delete every other partition; the creator also drops the `TEMPORARY` flag, causing that form to target the real table. Translate these forms to explicit Paimon semantics or reject them before planning, and add a two-partition regression proving no wider snapshot is committed. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java: ########## @@ -0,0 +1,425 @@ +// 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.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.CoreOptions; +import org.apache.paimon.io.DataInputDeserializer; +import org.apache.paimon.table.FileStoreTable; +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.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Paimon transaction. + * + * Lifecycle: + * 1. bind() — pin the concrete table and writer configuration + * 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); + + enum CommitState { + PREPARED, + COMMITTING, + COMMITTED, + OUTCOME_UNKNOWN + } + + 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 PaimonWriteBinding binding; + private CommitState state = CommitState.PREPARED; + + private final List<byte[]> commitPayloads = Lists.newArrayList(); + private final Set<String> commitPayloadSet = new HashSet<>(); + + public PaimonTransaction(PaimonMetadataOps ops, long transactionId) { + this.ops = Preconditions.checkNotNull(ops, "Paimon metadata ops must not be null"); + Preconditions.checkArgument(transactionId > 0, "Paimon transaction id must be positive"); + this.transactionId = transactionId; + this.commitUser = commitUser(transactionId); + } + + // ──────────────────────────────────────────────────────────── + // Transaction lifecycle + // ──────────────────────────────────────────────────────────── + + public synchronized void bind(PaimonWriteBinding binding) { + Preconditions.checkNotNull(binding, "Paimon write binding must not be null"); + Preconditions.checkState(this.binding == null, "Paimon transaction is already bound"); + Preconditions.checkState(state == CommitState.PREPARED, + "Paimon transaction can only be bound while prepared"); + this.binding = binding; + } + + @Override + public void commit() throws UserException { + PaimonWriteBinding writeBinding = requireBinding(); + List<byte[]> rawPayloads = snapshotPayloads(); + if (rawPayloads.isEmpty() && !writeBinding.isOverwrite()) { + LOG.info("Skip empty PaimonTransaction commit, txnId={}, table={}", + transactionId, tableName()); + markPreparedTransactionCommitted(); + return; + } + try { + List<CommitMessage> allMessages = deserializePayloads(rawPayloads); + LOG.info("Commit PaimonTransaction, txnId={}, table={}, payloads={}, messages={}, overwrite={}", + transactionId, tableName(), rawPayloads.size(), allMessages.size(), + writeBinding.isOverwrite()); + if (allMessages.isEmpty() && !writeBinding.isOverwrite()) { + throw new RuntimeException( + "Paimon commit messages are empty, payloads=" + rawPayloads.size()); + } + doCommitWithReconciliation(writeBinding, allMessages); + } catch (Exception e) { + throw new UserException("Failed to commit paimon transaction on FE", e); + } + } + + @Override + public void rollback() { + CommitState currentState = getState(); + if (currentState == CommitState.COMMITTED) { + LOG.info("Skip rollback for committed PaimonTransaction, txnId={}, table={}", + transactionId, tableName()); + return; + } + if (currentState == CommitState.COMMITTING + || currentState == CommitState.OUTCOME_UNKNOWN) { + LOG.warn("Skip rollback for PaimonTransaction in state {}, txnId={}, table={}. " + + "Preserving data files for snapshot safety", + currentState, transactionId, tableName()); + return; + } + List<byte[]> rawPayloads = snapshotPayloads(); + if (rawPayloads.isEmpty()) { + LOG.info("Skip empty PaimonTransaction rollback, txnId={}, table={}", + transactionId, tableName()); + return; + } + try { + PaimonWriteBinding writeBinding = requireBinding(); + 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(writeBinding, 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); Review Comment: [P1] Bound FE commit-metadata memory Every backend's accepted report is accumulated in this one transaction. Even when each report stays below Thrift's 100 MiB limit, this Base64 dedup key retains at least another 4/3 of the payload while `commitPayloads` keeps a byte-array copy; `commit()` then clones the complete aggregate and builds the decoded message graph without releasing either retained representation. For example, ten valid 80 MiB reports require about 1.87 GiB before commit and 2.67 GiB during the snapshot copy, before decoded objects, so a large distributed write can OOM the FE leader. The existing single-RPC issue does not bound this aggregate. Use a fixed-size digest/sequence for deduplication, enforce a transaction-scoped budget or managed spool, and avoid the full snapshot copy; cover many individually bounded backend reports. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,387 @@ +// 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.BucketMode; +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); + if (table.bucketMode() == BucketMode.HASH_DYNAMIC) { + throw new UnsupportedOperationException( + "Paimon dynamic-bucket tables are not supported for writes"); + } + 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"); Review Comment: [P1] Put write-buffer spill on Doris-managed disks Paimon 1.3.1 enables `write-buffer-spillable` by default, so each LocalState installs an IO manager rooted at the JVM-wide `java.io.tmpdir` (normally the host `/tmp`). Under memory pressure, concurrent writers can therefore fill the OS/root filesystem even while Doris's configured storage paths and disk-capacity controls remain healthy; Paimon's default spill disk limit is also unbounded. The Paimon reader already receives ExecEnv-managed roots for this purpose. Pass managed spill directories and the required reservation/accounting context through the sink, and test path selection plus normal and cancelled cleanup. -- 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]
