suxiaogang223 commented on code in PR #65868:
URL: https://github.com/apache/doris/pull/65868#discussion_r3671691917
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java:
##########
@@ -859,6 +873,89 @@ 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());
+ Set<String> staticPartitionColNames = staticPartitions.keySet();
+ if (!staticPartitionColNames.isEmpty()) {
+ Set<String> partitionColumnNames =
Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
+ partitionColumnNames.addAll(table.getWritePartitionColumnNames());
+ for (String columnName : staticPartitionColNames) {
+ if (!partitionColumnNames.contains(columnName)) {
+ throw new AnalysisException(String.format(
+ "Column '%s' is not a partition column of Paimon
table '%s'",
+ columnName, table.getName()));
+ }
+ Expression partitionValue = staticPartitions.get(columnName);
+ if (!(partitionValue instanceof Literal)) {
+ throw new AnalysisException(String.format(
+ "Partition value for column '%s' must be a
literal, but got: %s",
Review Comment:
Fixed in 7418f8cae7. Paimon target columns now come from the immutable
PaimonWriteTarget captured from the latest remote table, rather than
getBaseSchema/getColumn on the read-side table snapshot. The same target is
carried through binding, physical planning, and the sink.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonArrowConverter.java:
##########
@@ -0,0 +1,373 @@
+// 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.complex.ListVector;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.complex.StructVector;
+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.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+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 {
+
+ private final ZoneId sessionTimeZone;
+
+ PaimonArrowConverter(ZoneId sessionTimeZone) {
+ this.sessionTimeZone = sessionTimeZone;
+ }
+
+ RowReader rows(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");
+ }
+ return new RowReader(fields, vectors, targetTypes);
+ }
+
+ /** Bound view of one Arrow batch which converts only the requested row. */
+ final class RowReader {
+ private final List<Field> fields;
+ private final List<FieldVector> vectors;
+ private final DataType[] targetTypes;
+
+ private RowReader(
+ List<Field> fields, List<FieldVector> vectors, DataType[]
targetTypes) {
+ this.fields = fields;
+ this.vectors = vectors;
+ this.targetTypes = targetTypes;
+ }
+
+ Object[] values(int rowIndex) {
+ Object[] values = new Object[vectors.size()];
+ for (int column = 0; column < vectors.size(); column++) {
+ values[column] = convertVectorValue(
+ vectors.get(column), rowIndex, fields.get(column),
targetTypes[column]);
+ }
+ return values;
+ }
+ }
+
+ private Object convertVectorValue(
+ FieldVector vector, int index, Field arrowField, DataType
targetType) {
+ if (vector.isNull(index)) {
+ return null;
+ }
+ if (vector instanceof StructVector && targetType instanceof RowType) {
+ return convertStructVector((StructVector) vector, index, (RowType)
targetType);
+ }
+ if (vector instanceof MapVector && targetType instanceof MapType) {
+ return convertMapVector((MapVector) vector, index, (MapType)
targetType);
+ }
+ if (vector instanceof ListVector && targetType instanceof ArrayType) {
+ return convertArrayVector((ListVector) vector, index, (ArrayType)
targetType);
+ }
+ if (vector instanceof IntVector) {
+ return ((IntVector) vector).get(index);
+ }
+ if (vector instanceof BigIntVector) {
+ return ((BigIntVector) vector).get(index);
+ }
+ if (vector instanceof SmallIntVector) {
+ return ((SmallIntVector) vector).get(index);
+ }
+ if (vector instanceof TinyIntVector) {
+ return ((TinyIntVector) vector).get(index);
+ }
+ if (vector instanceof Float4Vector) {
+ return ((Float4Vector) vector).get(index);
+ }
+ if (vector instanceof Float8Vector) {
+ return ((Float8Vector) vector).get(index);
+ }
+ if (vector instanceof BitVector) {
+ return ((BitVector) vector).get(index) == 1;
+ }
+ if (vector instanceof DateDayVector) {
+ return ((DateDayVector) vector).get(index);
+ }
+ if (vector instanceof VarCharVector) {
Review Comment:
Fixed in 7418f8cae7. PaimonArrowConverter now handles VariantType explicitly
and converts Doris UTF-8 JSON to GenericVariant.fromJson. Unit coverage
includes scalar, array, and object JSON values.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonTransaction.java:
##########
@@ -0,0 +1,448 @@
+// 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.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<CommitPayloadKey> 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;
+ }
+ // Treat the Thrift payload as immutable after report handling and
adopt it directly. The
+ // report is not reused, so copying it would only create a second full
representation.
+ if (commitPayloadSet.add(new CommitPayloadKey(payload))) {
+ commitPayloads.add(payload);
+ }
+ }
+
+ // ────────────────────────────────────────────────────────────
+ // Transaction identity
+ // ────────────────────────────────────────────────────────────
+
+ public static String commitUser(long transactionId) {
Review Comment:
Fixed in 7418f8cae7. The commit user is now namespaced as
doris_cluster_<clusterId>_txn_<transactionId>. It remains stable for retries
while preventing independent Doris clusters from sharing the same Paimon stream
commit identity.
--
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]
