suxiaogang223 commented on code in PR #65868: URL: https://github.com/apache/doris/pull/65868#discussion_r3666332032
########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,617 @@ +// 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.classloader.ThreadClassLoaderContext; +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.crosspartition.IndexBootstrap; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.disk.IOManagerImpl; +import org.apache.paimon.index.BucketAssigner; +import org.apache.paimon.index.HashBucketAssigner; +import org.apache.paimon.index.SimpleHashBucketAssigner; +import org.apache.paimon.memory.HeapMemorySegmentPool; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.InnerTableCommit; +import org.apache.paimon.table.sink.PartitionKeyExtractor; +import org.apache.paimon.table.sink.RowPartitionKeyExtractor; +import org.apache.paimon.table.sink.SinkRecord; +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.nio.file.Files; +import java.nio.file.Paths; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * JNI entry point for Paimon write operations. + * + * <p>Called from C++ ({@code JniPaimonWriter}) via JNI. One instance per BE pipeline + * fragment (one per {@code PaimonTableWriter}). 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) + * → Paimon SDK bucket assignment and table write + * </pre> + * + * <p>Commit path: + * + * <pre> + * PaimonTableWriter::close() → JNI → PaimonJniWriter.prepareCommit() + * → TableWriteImpl.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 PaimonCommitCodec commitCodec = new PaimonCommitCodec(); + + private PreExecutionAuthenticator preExecutionAuthenticator; + private PaimonArrowConverter arrowConverter; + + private PaimonWriteSchema writeSchema; + private FileStoreTable table; + private TableWriteImpl<?> writer; + private IOManager ioManager; + private long commitIdentifier; + private String commitUser; + private BucketMode bucketMode; + private BucketAssigner hashBucketAssigner; + private PartitionKeyExtractor<InternalRow> dynamicBucketExtractor; + private GlobalIndexAssigner globalIndexAssigner; + private boolean fullCompactionChangelog; + private final Set<PartitionBucket> fullCompactionBuckets = new HashSet<>(); + 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 + * @param timeZone normalized Doris session timezone used for Paimon LTZ values + * @param spillDirectories Doris storage-root scoped directories for Paimon write-buffer spill + */ + public void open(String serializedTable, Map<String, String> hadoopConfig, + String[] columnNames, long transactionId, String commitUser, + boolean overwrite, String timeZone, String spillDirectories) throws Exception { + try (ThreadClassLoaderContext ignored = new ThreadClassLoaderContext(classLoader)) { + this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(hadoopConfig); + this.arrowConverter = new PaimonArrowConverter(ZoneId.of(timeZone)); Review Comment: Fixed in 2932001a30. The fix is applied at the FE execution boundary rather than in PaimonJniWriter: TimeUtils.getCanonicalTimeZone resolves Doris aliases with the existing alias map, and Coordinator, CoordinatorContext, PointQueryExecutor, and BE constant folding now all send canonical IDs. The UT covers EST in addition to CST. This keeps the JNI writer on standard ZoneId input and avoids duplicating Doris alias semantics in the connector. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java: ########## @@ -0,0 +1,167 @@ +// 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.nereids.trees.plans.physical; + +import org.apache.doris.catalog.Column; +import org.apache.doris.datasource.paimon.PaimonExternalDatabase; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType; +import org.apache.doris.nereids.properties.LogicalProperties; +import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.statistics.Statistics; + +import com.google.common.base.Preconditions; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.TreeMap; + +/** + * Physical Paimon table sink. + */ +public class PhysicalPaimonTableSink<CHILD_TYPE extends Plan> + extends PhysicalBaseExternalTableSink<CHILD_TYPE> { + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + CHILD_TYPE child) { + this(database, targetTable, cols, outputExprs, groupExpression, logicalProperties, + PhysicalProperties.SINK_RANDOM_PARTITIONED, null, child); + } + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + PhysicalProperties physicalProperties, + Statistics statistics, + CHILD_TYPE child) { + super(PlanType.PHYSICAL_PAIMON_TABLE_SINK, database, targetTable, cols, outputExprs, + groupExpression, logicalProperties, physicalProperties, statistics, child); + } + + @Override + public Plan withChildren(List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, children.get(0)); + } + + @Override + public Plan withGroupExpression(Optional<GroupExpression> groupExpression) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, child()); + } + + @Override + public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + logicalProperties.get(), physicalProperties, statistics, children.get(0)); + } + + @Override + public PhysicalPaimonTableSink<Plan> withPhysicalPropertiesAndStats( + PhysicalProperties physicalProperties, Statistics stats) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, stats, child()); + } + + @Override + public PhysicalProperties getRequirePhysicalProperties() { + if (requiresSingleWriter()) { + return PhysicalProperties.GATHER; + } + + PaimonExternalTable paimonExternalTable = (PaimonExternalTable) targetTable; + Table paimonTable = paimonExternalTable.getPaimonTable(Optional.empty()); + List<String> primaryKeys = paimonTable.primaryKeys(); + if (primaryKeys.isEmpty()) { + return PhysicalProperties.SINK_RANDOM_PARTITIONED; + } + + List<Slot> outputSlots = child().getOutput(); + Preconditions.checkState(cols.size() == outputSlots.size(), + "Paimon sink columns must match child output"); + Map<String, ExprId> columnExprIds = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (int i = 0; i < cols.size(); i++) { + columnExprIds.put(cols.get(i).getName(), outputSlots.get(i).getExprId()); + } + + List<ExprId> primaryKeyExprIds = new ArrayList<>(primaryKeys.size()); + for (String primaryKey : primaryKeys) { + primaryKeyExprIds.add(Preconditions.checkNotNull( + columnExprIds.get(primaryKey), + "Paimon primary-key column is missing from sink output")); + } + return PhysicalProperties.createHash(primaryKeyExprIds, ShuffleType.REQUIRE); Review Comment: Fixed in 2932001a30. HASH_FIXED primary-key tables now require a single writer through Gather until a Paimon bucket-aware exchange exists. This avoids splitting one partition/bucket ownership across TableWriteImpl instances, including bucket-key subset cases. The regression plan expectation is updated from hash exchange to gather. ########## be/src/exec/pipeline/pipeline_fragment_context.cpp: ########## @@ -2175,6 +2184,20 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r } } + if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) { Review Comment: We intentionally do not add a total report limit in this PR. The existing protocol has one final TReportExecStatusParams completion boundary; safely splitting prepared Paimon commit data across reports would require transaction-scoped accumulation, ordering and completion semantics, retries, and abort handling in FE and BE. A local aggregate cap would only turn large valid writes into deterministic failures and would not provide a safe split. This PR therefore bounds each serialized chunk during encoding to avoid JVM allocation spikes and leaves multi-report transport as a separate protocol-level enhancement. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java: ########## @@ -150,71 +144,115 @@ public static List<InternalRow> read( return rows; } - public static PaimonPartitionInfo generatePartitionInfo(List<Column> partitionColumns, - List<Partition> paimonPartitions, boolean legacyPartitionName) { + public static PaimonPartitionInfo generatePartitionInfo(Table table, List<Column> partitionColumns, + List<PartitionEntry> partitionEntries) { - if (CollectionUtils.isEmpty(partitionColumns) || paimonPartitions.isEmpty()) { + if (CollectionUtils.isEmpty(partitionColumns) || partitionEntries.isEmpty()) { return PaimonPartitionInfo.EMPTY; } - Map<String, PartitionItem> nameToPartitionItem = Maps.newHashMap(); - Map<String, Partition> nameToPartition = Maps.newHashMap(); - PaimonPartitionInfo partitionInfo = new PaimonPartitionInfo(nameToPartitionItem, nameToPartition); + CoreOptions options = new CoreOptions(table.options()); + InternalRowPartitionComputer partitionComputer = new InternalRowPartitionComputer( + options.partitionDefaultName(), + table.rowType().project(table.partitionKeys()), + table.partitionKeys().toArray(new String[0]), + options.legacyPartitionName()); List<Type> types = partitionColumns.stream() .map(Column::getType) .collect(Collectors.toList()); + List<PaimonPartitionCandidate> candidates = Lists.newArrayListWithExpectedSize(partitionEntries.size()); + Map<String, Integer> displayNameCounts = Maps.newHashMap(); + + for (PartitionEntry partitionEntry : partitionEntries) { + Map<String, String> typedSpec = getPartitionInfoMap( + table, partitionEntry.partition(), TimeUtils.getTimeZone().getID()); Review Comment: Fixed in 2932001a30. LTZ partition metadata now returns PaimonPartitionInfo.UNPRUNABLE before building cacheable bounds. This intentionally delegates pruning to Paimon and prevents any session-local civil time from entering the table-level cache. A unit test covers this state. We can remove this fallback only after the cache uses a time-zone-independent typed identity. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java: ########## @@ -0,0 +1,167 @@ +// 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.nereids.trees.plans.physical; + +import org.apache.doris.catalog.Column; +import org.apache.doris.datasource.paimon.PaimonExternalDatabase; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType; +import org.apache.doris.nereids.properties.LogicalProperties; +import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.statistics.Statistics; + +import com.google.common.base.Preconditions; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.TreeMap; + +/** + * Physical Paimon table sink. + */ +public class PhysicalPaimonTableSink<CHILD_TYPE extends Plan> + extends PhysicalBaseExternalTableSink<CHILD_TYPE> { + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + CHILD_TYPE child) { + this(database, targetTable, cols, outputExprs, groupExpression, logicalProperties, + PhysicalProperties.SINK_RANDOM_PARTITIONED, null, child); + } + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + PhysicalProperties physicalProperties, + Statistics statistics, + CHILD_TYPE child) { + super(PlanType.PHYSICAL_PAIMON_TABLE_SINK, database, targetTable, cols, outputExprs, + groupExpression, logicalProperties, physicalProperties, statistics, child); + } + + @Override + public Plan withChildren(List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, children.get(0)); + } + + @Override + public Plan withGroupExpression(Optional<GroupExpression> groupExpression) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, child()); + } + + @Override + public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + logicalProperties.get(), physicalProperties, statistics, children.get(0)); + } + + @Override + public PhysicalPaimonTableSink<Plan> withPhysicalPropertiesAndStats( + PhysicalProperties physicalProperties, Statistics stats) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, stats, child()); + } + + @Override + public PhysicalProperties getRequirePhysicalProperties() { + if (requiresSingleWriter()) { + return PhysicalProperties.GATHER; + } + + PaimonExternalTable paimonExternalTable = (PaimonExternalTable) targetTable; + Table paimonTable = paimonExternalTable.getPaimonTable(Optional.empty()); + List<String> primaryKeys = paimonTable.primaryKeys(); + if (primaryKeys.isEmpty()) { + return PhysicalProperties.SINK_RANDOM_PARTITIONED; + } + + List<Slot> outputSlots = child().getOutput(); + Preconditions.checkState(cols.size() == outputSlots.size(), + "Paimon sink columns must match child output"); + Map<String, ExprId> columnExprIds = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (int i = 0; i < cols.size(); i++) { + columnExprIds.put(cols.get(i).getName(), outputSlots.get(i).getExprId()); + } + + List<ExprId> primaryKeyExprIds = new ArrayList<>(primaryKeys.size()); + for (String primaryKey : primaryKeys) { + primaryKeyExprIds.add(Preconditions.checkNotNull( + columnExprIds.get(primaryKey), + "Paimon primary-key column is missing from sink output")); + } + return PhysicalProperties.createHash(primaryKeyExprIds, ShuffleType.REQUIRE); Review Comment: Fixed in 2932001a30. HASH_FIXED primary-key tables now require a single writer through Gather until a Paimon bucket-aware exchange exists. This avoids splitting one partition/bucket ownership across TableWriteImpl instances, including bucket-key subset cases. The regression plan expectation is updated from hash exchange to gather. ########## be/src/exec/pipeline/pipeline_fragment_context.cpp: ########## @@ -2175,6 +2184,20 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r } } + if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) { Review Comment: We intentionally do not add a total report limit in this PR. The existing protocol has one final TReportExecStatusParams completion boundary; safely splitting prepared Paimon commit data across reports would require transaction-scoped accumulation, ordering and completion semantics, retries, and abort handling in FE and BE. A local aggregate cap would only turn large valid writes into deterministic failures and would not provide a safe split. This PR therefore bounds each serialized chunk during encoding to avoid JVM allocation spikes and leaves multi-report transport as a separate protocol-level enhancement. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java: ########## @@ -150,71 +144,115 @@ public static List<InternalRow> read( return rows; } - public static PaimonPartitionInfo generatePartitionInfo(List<Column> partitionColumns, - List<Partition> paimonPartitions, boolean legacyPartitionName) { + public static PaimonPartitionInfo generatePartitionInfo(Table table, List<Column> partitionColumns, + List<PartitionEntry> partitionEntries) { - if (CollectionUtils.isEmpty(partitionColumns) || paimonPartitions.isEmpty()) { + if (CollectionUtils.isEmpty(partitionColumns) || partitionEntries.isEmpty()) { return PaimonPartitionInfo.EMPTY; } - Map<String, PartitionItem> nameToPartitionItem = Maps.newHashMap(); - Map<String, Partition> nameToPartition = Maps.newHashMap(); - PaimonPartitionInfo partitionInfo = new PaimonPartitionInfo(nameToPartitionItem, nameToPartition); + CoreOptions options = new CoreOptions(table.options()); + InternalRowPartitionComputer partitionComputer = new InternalRowPartitionComputer( + options.partitionDefaultName(), + table.rowType().project(table.partitionKeys()), + table.partitionKeys().toArray(new String[0]), + options.legacyPartitionName()); List<Type> types = partitionColumns.stream() .map(Column::getType) .collect(Collectors.toList()); + List<PaimonPartitionCandidate> candidates = Lists.newArrayListWithExpectedSize(partitionEntries.size()); + Map<String, Integer> displayNameCounts = Maps.newHashMap(); + + for (PartitionEntry partitionEntry : partitionEntries) { + Map<String, String> typedSpec = getPartitionInfoMap( + table, partitionEntry.partition(), TimeUtils.getTimeZone().getID()); Review Comment: Fixed in 2932001a30. LTZ partition metadata now returns PaimonPartitionInfo.UNPRUNABLE before building cacheable bounds. This intentionally delegates pruning to Paimon and prevents any session-local civil time from entering the table-level cache. A unit test covers this state. We can remove this fallback only after the cache uses a time-zone-independent typed identity. ########## be/src/exec/pipeline/pipeline_fragment_context.cpp: ########## @@ -2175,6 +2184,20 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r } } + if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) { Review Comment: We intentionally do not add a total report limit in this PR. The existing protocol has one final TReportExecStatusParams completion boundary; safely splitting prepared Paimon commit data across reports would require transaction-scoped accumulation, ordering and completion semantics, retries, and abort handling in FE and BE. A local aggregate cap would only turn large valid writes into deterministic failures and would not provide a safe split. This PR therefore bounds each serialized chunk during encoding to avoid JVM allocation spikes and leaves multi-report transport as a separate protocol-level enhancement. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java: ########## @@ -150,71 +144,115 @@ public static List<InternalRow> read( return rows; } - public static PaimonPartitionInfo generatePartitionInfo(List<Column> partitionColumns, - List<Partition> paimonPartitions, boolean legacyPartitionName) { + public static PaimonPartitionInfo generatePartitionInfo(Table table, List<Column> partitionColumns, + List<PartitionEntry> partitionEntries) { - if (CollectionUtils.isEmpty(partitionColumns) || paimonPartitions.isEmpty()) { + if (CollectionUtils.isEmpty(partitionColumns) || partitionEntries.isEmpty()) { return PaimonPartitionInfo.EMPTY; } - Map<String, PartitionItem> nameToPartitionItem = Maps.newHashMap(); - Map<String, Partition> nameToPartition = Maps.newHashMap(); - PaimonPartitionInfo partitionInfo = new PaimonPartitionInfo(nameToPartitionItem, nameToPartition); + CoreOptions options = new CoreOptions(table.options()); + InternalRowPartitionComputer partitionComputer = new InternalRowPartitionComputer( + options.partitionDefaultName(), + table.rowType().project(table.partitionKeys()), + table.partitionKeys().toArray(new String[0]), + options.legacyPartitionName()); List<Type> types = partitionColumns.stream() .map(Column::getType) .collect(Collectors.toList()); + List<PaimonPartitionCandidate> candidates = Lists.newArrayListWithExpectedSize(partitionEntries.size()); + Map<String, Integer> displayNameCounts = Maps.newHashMap(); + + for (PartitionEntry partitionEntry : partitionEntries) { + Map<String, String> typedSpec = getPartitionInfoMap( + table, partitionEntry.partition(), TimeUtils.getTimeZone().getID()); Review Comment: Fixed in 2932001a30. LTZ partition metadata now returns PaimonPartitionInfo.UNPRUNABLE before building cacheable bounds. This intentionally delegates pruning to Paimon and prevents any session-local civil time from entering the table-level cache. A unit test covers this state. We can remove this fallback only after the cache uses a time-zone-independent typed identity. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,617 @@ +// 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.classloader.ThreadClassLoaderContext; +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.crosspartition.IndexBootstrap; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.disk.IOManagerImpl; +import org.apache.paimon.index.BucketAssigner; +import org.apache.paimon.index.HashBucketAssigner; +import org.apache.paimon.index.SimpleHashBucketAssigner; +import org.apache.paimon.memory.HeapMemorySegmentPool; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.InnerTableCommit; +import org.apache.paimon.table.sink.PartitionKeyExtractor; +import org.apache.paimon.table.sink.RowPartitionKeyExtractor; +import org.apache.paimon.table.sink.SinkRecord; +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.nio.file.Files; +import java.nio.file.Paths; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * JNI entry point for Paimon write operations. + * + * <p>Called from C++ ({@code JniPaimonWriter}) via JNI. One instance per BE pipeline + * fragment (one per {@code PaimonTableWriter}). 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) + * → Paimon SDK bucket assignment and table write + * </pre> + * + * <p>Commit path: + * + * <pre> + * PaimonTableWriter::close() → JNI → PaimonJniWriter.prepareCommit() + * → TableWriteImpl.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 PaimonCommitCodec commitCodec = new PaimonCommitCodec(); + + private PreExecutionAuthenticator preExecutionAuthenticator; + private PaimonArrowConverter arrowConverter; + + private PaimonWriteSchema writeSchema; + private FileStoreTable table; + private TableWriteImpl<?> writer; + private IOManager ioManager; + private long commitIdentifier; + private String commitUser; + private BucketMode bucketMode; + private BucketAssigner hashBucketAssigner; + private PartitionKeyExtractor<InternalRow> dynamicBucketExtractor; + private GlobalIndexAssigner globalIndexAssigner; + private boolean fullCompactionChangelog; + private final Set<PartitionBucket> fullCompactionBuckets = new HashSet<>(); + 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 + * @param timeZone normalized Doris session timezone used for Paimon LTZ values + * @param spillDirectories Doris storage-root scoped directories for Paimon write-buffer spill + */ + public void open(String serializedTable, Map<String, String> hadoopConfig, + String[] columnNames, long transactionId, String commitUser, + boolean overwrite, String timeZone, String spillDirectories) throws Exception { + try (ThreadClassLoaderContext ignored = new ThreadClassLoaderContext(classLoader)) { + this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(hadoopConfig); + this.arrowConverter = new PaimonArrowConverter(ZoneId.of(timeZone)); Review Comment: Fixed in 2932001a30. The fix is applied at the FE execution boundary rather than in PaimonJniWriter: TimeUtils.getCanonicalTimeZone resolves Doris aliases with the existing alias map, and Coordinator, CoordinatorContext, PointQueryExecutor, and BE constant folding now all send canonical IDs. The UT covers EST in addition to CST. This keeps the JNI writer on standard ZoneId input and avoids duplicating Doris alias semantics in the connector. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java: ########## @@ -0,0 +1,167 @@ +// 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.nereids.trees.plans.physical; + +import org.apache.doris.catalog.Column; +import org.apache.doris.datasource.paimon.PaimonExternalDatabase; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType; +import org.apache.doris.nereids.properties.LogicalProperties; +import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.statistics.Statistics; + +import com.google.common.base.Preconditions; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.TreeMap; + +/** + * Physical Paimon table sink. + */ +public class PhysicalPaimonTableSink<CHILD_TYPE extends Plan> + extends PhysicalBaseExternalTableSink<CHILD_TYPE> { + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + CHILD_TYPE child) { + this(database, targetTable, cols, outputExprs, groupExpression, logicalProperties, + PhysicalProperties.SINK_RANDOM_PARTITIONED, null, child); + } + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + PhysicalProperties physicalProperties, + Statistics statistics, + CHILD_TYPE child) { + super(PlanType.PHYSICAL_PAIMON_TABLE_SINK, database, targetTable, cols, outputExprs, + groupExpression, logicalProperties, physicalProperties, statistics, child); + } + + @Override + public Plan withChildren(List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, children.get(0)); + } + + @Override + public Plan withGroupExpression(Optional<GroupExpression> groupExpression) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, child()); + } + + @Override + public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + logicalProperties.get(), physicalProperties, statistics, children.get(0)); + } + + @Override + public PhysicalPaimonTableSink<Plan> withPhysicalPropertiesAndStats( + PhysicalProperties physicalProperties, Statistics stats) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, stats, child()); + } + + @Override + public PhysicalProperties getRequirePhysicalProperties() { + if (requiresSingleWriter()) { + return PhysicalProperties.GATHER; + } + + PaimonExternalTable paimonExternalTable = (PaimonExternalTable) targetTable; + Table paimonTable = paimonExternalTable.getPaimonTable(Optional.empty()); + List<String> primaryKeys = paimonTable.primaryKeys(); + if (primaryKeys.isEmpty()) { + return PhysicalProperties.SINK_RANDOM_PARTITIONED; + } + + List<Slot> outputSlots = child().getOutput(); + Preconditions.checkState(cols.size() == outputSlots.size(), + "Paimon sink columns must match child output"); + Map<String, ExprId> columnExprIds = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (int i = 0; i < cols.size(); i++) { + columnExprIds.put(cols.get(i).getName(), outputSlots.get(i).getExprId()); + } + + List<ExprId> primaryKeyExprIds = new ArrayList<>(primaryKeys.size()); + for (String primaryKey : primaryKeys) { + primaryKeyExprIds.add(Preconditions.checkNotNull( + columnExprIds.get(primaryKey), + "Paimon primary-key column is missing from sink output")); + } + return PhysicalProperties.createHash(primaryKeyExprIds, ShuffleType.REQUIRE); Review Comment: Fixed in 2932001a30. HASH_FIXED primary-key tables now require a single writer through Gather until a Paimon bucket-aware exchange exists. This avoids splitting one partition/bucket ownership across TableWriteImpl instances, including bucket-key subset cases. The regression plan expectation is updated from hash exchange to gather. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,617 @@ +// 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.classloader.ThreadClassLoaderContext; +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.crosspartition.IndexBootstrap; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.disk.IOManagerImpl; +import org.apache.paimon.index.BucketAssigner; +import org.apache.paimon.index.HashBucketAssigner; +import org.apache.paimon.index.SimpleHashBucketAssigner; +import org.apache.paimon.memory.HeapMemorySegmentPool; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.InnerTableCommit; +import org.apache.paimon.table.sink.PartitionKeyExtractor; +import org.apache.paimon.table.sink.RowPartitionKeyExtractor; +import org.apache.paimon.table.sink.SinkRecord; +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.nio.file.Files; +import java.nio.file.Paths; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * JNI entry point for Paimon write operations. + * + * <p>Called from C++ ({@code JniPaimonWriter}) via JNI. One instance per BE pipeline + * fragment (one per {@code PaimonTableWriter}). 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) + * → Paimon SDK bucket assignment and table write + * </pre> + * + * <p>Commit path: + * + * <pre> + * PaimonTableWriter::close() → JNI → PaimonJniWriter.prepareCommit() + * → TableWriteImpl.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 PaimonCommitCodec commitCodec = new PaimonCommitCodec(); + + private PreExecutionAuthenticator preExecutionAuthenticator; + private PaimonArrowConverter arrowConverter; + + private PaimonWriteSchema writeSchema; + private FileStoreTable table; + private TableWriteImpl<?> writer; + private IOManager ioManager; + private long commitIdentifier; + private String commitUser; + private BucketMode bucketMode; + private BucketAssigner hashBucketAssigner; + private PartitionKeyExtractor<InternalRow> dynamicBucketExtractor; + private GlobalIndexAssigner globalIndexAssigner; + private boolean fullCompactionChangelog; + private final Set<PartitionBucket> fullCompactionBuckets = new HashSet<>(); + 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 + * @param timeZone normalized Doris session timezone used for Paimon LTZ values + * @param spillDirectories Doris storage-root scoped directories for Paimon write-buffer spill + */ + public void open(String serializedTable, Map<String, String> hadoopConfig, + String[] columnNames, long transactionId, String commitUser, + boolean overwrite, String timeZone, String spillDirectories) throws Exception { + try (ThreadClassLoaderContext ignored = new ThreadClassLoaderContext(classLoader)) { + this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(hadoopConfig); + this.arrowConverter = new PaimonArrowConverter(ZoneId.of(timeZone)); Review Comment: Fixed in 2932001a30. The fix is applied at the FE execution boundary rather than in PaimonJniWriter: TimeUtils.getCanonicalTimeZone resolves Doris aliases with the existing alias map, and Coordinator, CoordinatorContext, PointQueryExecutor, and BE constant folding now all send canonical IDs. The UT covers EST in addition to CST. This keeps the JNI writer on standard ZoneId input and avoids duplicating Doris alias semantics in the connector. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,617 @@ +// 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.classloader.ThreadClassLoaderContext; +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.crosspartition.IndexBootstrap; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.disk.IOManagerImpl; +import org.apache.paimon.index.BucketAssigner; +import org.apache.paimon.index.HashBucketAssigner; +import org.apache.paimon.index.SimpleHashBucketAssigner; +import org.apache.paimon.memory.HeapMemorySegmentPool; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.InnerTableCommit; +import org.apache.paimon.table.sink.PartitionKeyExtractor; +import org.apache.paimon.table.sink.RowPartitionKeyExtractor; +import org.apache.paimon.table.sink.SinkRecord; +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.nio.file.Files; +import java.nio.file.Paths; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * JNI entry point for Paimon write operations. + * + * <p>Called from C++ ({@code JniPaimonWriter}) via JNI. One instance per BE pipeline + * fragment (one per {@code PaimonTableWriter}). 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) + * → Paimon SDK bucket assignment and table write + * </pre> + * + * <p>Commit path: + * + * <pre> + * PaimonTableWriter::close() → JNI → PaimonJniWriter.prepareCommit() + * → TableWriteImpl.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 PaimonCommitCodec commitCodec = new PaimonCommitCodec(); + + private PreExecutionAuthenticator preExecutionAuthenticator; + private PaimonArrowConverter arrowConverter; + + private PaimonWriteSchema writeSchema; + private FileStoreTable table; + private TableWriteImpl<?> writer; + private IOManager ioManager; + private long commitIdentifier; + private String commitUser; + private BucketMode bucketMode; + private BucketAssigner hashBucketAssigner; + private PartitionKeyExtractor<InternalRow> dynamicBucketExtractor; + private GlobalIndexAssigner globalIndexAssigner; + private boolean fullCompactionChangelog; + private final Set<PartitionBucket> fullCompactionBuckets = new HashSet<>(); + 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 + * @param timeZone normalized Doris session timezone used for Paimon LTZ values + * @param spillDirectories Doris storage-root scoped directories for Paimon write-buffer spill + */ + public void open(String serializedTable, Map<String, String> hadoopConfig, + String[] columnNames, long transactionId, String commitUser, + boolean overwrite, String timeZone, String spillDirectories) throws Exception { + try (ThreadClassLoaderContext ignored = new ThreadClassLoaderContext(classLoader)) { + this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(hadoopConfig); + this.arrowConverter = new PaimonArrowConverter(ZoneId.of(timeZone)); Review Comment: Fixed in 2932001a30. The fix is applied at the FE execution boundary rather than in PaimonJniWriter: TimeUtils.getCanonicalTimeZone resolves Doris aliases with the existing alias map, and Coordinator, CoordinatorContext, PointQueryExecutor, and BE constant folding now all send canonical IDs. The UT covers EST in addition to CST. This keeps the JNI writer on standard ZoneId input and avoids duplicating Doris alias semantics in the connector. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java: ########## @@ -150,71 +144,115 @@ public static List<InternalRow> read( return rows; } - public static PaimonPartitionInfo generatePartitionInfo(List<Column> partitionColumns, - List<Partition> paimonPartitions, boolean legacyPartitionName) { + public static PaimonPartitionInfo generatePartitionInfo(Table table, List<Column> partitionColumns, + List<PartitionEntry> partitionEntries) { - if (CollectionUtils.isEmpty(partitionColumns) || paimonPartitions.isEmpty()) { + if (CollectionUtils.isEmpty(partitionColumns) || partitionEntries.isEmpty()) { return PaimonPartitionInfo.EMPTY; } - Map<String, PartitionItem> nameToPartitionItem = Maps.newHashMap(); - Map<String, Partition> nameToPartition = Maps.newHashMap(); - PaimonPartitionInfo partitionInfo = new PaimonPartitionInfo(nameToPartitionItem, nameToPartition); + CoreOptions options = new CoreOptions(table.options()); + InternalRowPartitionComputer partitionComputer = new InternalRowPartitionComputer( + options.partitionDefaultName(), + table.rowType().project(table.partitionKeys()), + table.partitionKeys().toArray(new String[0]), + options.legacyPartitionName()); List<Type> types = partitionColumns.stream() .map(Column::getType) .collect(Collectors.toList()); + List<PaimonPartitionCandidate> candidates = Lists.newArrayListWithExpectedSize(partitionEntries.size()); + Map<String, Integer> displayNameCounts = Maps.newHashMap(); + + for (PartitionEntry partitionEntry : partitionEntries) { + Map<String, String> typedSpec = getPartitionInfoMap( + table, partitionEntry.partition(), TimeUtils.getTimeZone().getID()); Review Comment: Fixed in 2932001a30. LTZ partition metadata now returns PaimonPartitionInfo.UNPRUNABLE before building cacheable bounds. This intentionally delegates pruning to Paimon and prevents any session-local civil time from entering the table-level cache. A unit test covers this state. We can remove this fallback only after the cache uses a time-zone-independent typed identity. ########## be/src/exec/pipeline/pipeline_fragment_context.cpp: ########## @@ -2175,6 +2184,20 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r } } + if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) { Review Comment: We intentionally do not add a total report limit in this PR. The existing protocol has one final TReportExecStatusParams completion boundary; safely splitting prepared Paimon commit data across reports would require transaction-scoped accumulation, ordering and completion semantics, retries, and abort handling in FE and BE. A local aggregate cap would only turn large valid writes into deterministic failures and would not provide a safe split. This PR therefore bounds each serialized chunk during encoding to avoid JVM allocation spikes and leaves multi-report transport as a separate protocol-level enhancement. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/GlobalIndexAssigner.java: ########## @@ -0,0 +1,167 @@ +// 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.CoreOptions; +import org.apache.paimon.crosspartition.BucketAssigner; +import org.apache.paimon.crosspartition.ExistingProcessor; +import org.apache.paimon.crosspartition.IndexBootstrap; +import org.apache.paimon.crosspartition.KeyPartPartitionKeyExtractor; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.PartitionKeyExtractor; +import org.apache.paimon.table.sink.RowPartitionAllPrimaryKeyExtractor; +import org.apache.paimon.utils.IDMapping; +import org.apache.paimon.utils.PositiveIntInt; +import org.apache.paimon.utils.ProjectToRowFunction; + +import java.util.HashMap; +import java.util.Map; +import java.util.function.BiConsumer; + +/** + * Assigns buckets for a key-dynamic table with a process-local global-key index. + * + * <p>Doris gathers key-dynamic writes into one writer, so a single in-memory index can preserve + * Paimon's cross-partition merge semantics without a native state backend. + */ +final class GlobalIndexAssigner implements AutoCloseable { + private final FileStoreTable table; + // TODO: After resolving rocksdbjni allocator compatibility with the Doris BE jemalloc hook, + // use a RocksDB-backed on-disk index to bound Java heap usage for large tables. + private final Map<BinaryRow, PositiveIntInt> keyIndex = new HashMap<>(); Review Comment: Accepted as a documented correctness-first limitation for this PR. KEY_DYNAMIC requires a complete global primary-key to partition/bucket index; Paimon 1.3 offers the Flink state path but no connector-neutral bounded implementation that we can safely reuse in the embedded JVM. We tried rocksdbjni, but its native allocator conflicts with the Doris BE jemalloc hook in-process. The code has an explicit TODO to move this index to RocksDB-backed disk state after allocator compatibility is solved, and the roadmap documents heap growth. We intentionally avoid an approximate cap because rejecting after partial bootstrap or evicting keys would break correctness. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java: ########## @@ -0,0 +1,167 @@ +// 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.nereids.trees.plans.physical; + +import org.apache.doris.catalog.Column; +import org.apache.doris.datasource.paimon.PaimonExternalDatabase; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType; +import org.apache.doris.nereids.properties.LogicalProperties; +import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.statistics.Statistics; + +import com.google.common.base.Preconditions; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.TreeMap; + +/** + * Physical Paimon table sink. + */ +public class PhysicalPaimonTableSink<CHILD_TYPE extends Plan> + extends PhysicalBaseExternalTableSink<CHILD_TYPE> { + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + CHILD_TYPE child) { + this(database, targetTable, cols, outputExprs, groupExpression, logicalProperties, + PhysicalProperties.SINK_RANDOM_PARTITIONED, null, child); + } + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + PhysicalProperties physicalProperties, + Statistics statistics, + CHILD_TYPE child) { + super(PlanType.PHYSICAL_PAIMON_TABLE_SINK, database, targetTable, cols, outputExprs, + groupExpression, logicalProperties, physicalProperties, statistics, child); + } + + @Override + public Plan withChildren(List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, children.get(0)); + } + + @Override + public Plan withGroupExpression(Optional<GroupExpression> groupExpression) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, child()); + } + + @Override + public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + logicalProperties.get(), physicalProperties, statistics, children.get(0)); + } + + @Override + public PhysicalPaimonTableSink<Plan> withPhysicalPropertiesAndStats( + PhysicalProperties physicalProperties, Statistics stats) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, stats, child()); + } + + @Override + public PhysicalProperties getRequirePhysicalProperties() { + if (requiresSingleWriter()) { + return PhysicalProperties.GATHER; + } + + PaimonExternalTable paimonExternalTable = (PaimonExternalTable) targetTable; + Table paimonTable = paimonExternalTable.getPaimonTable(Optional.empty()); + List<String> primaryKeys = paimonTable.primaryKeys(); + if (primaryKeys.isEmpty()) { + return PhysicalProperties.SINK_RANDOM_PARTITIONED; + } + + List<Slot> outputSlots = child().getOutput(); + Preconditions.checkState(cols.size() == outputSlots.size(), + "Paimon sink columns must match child output"); + Map<String, ExprId> columnExprIds = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (int i = 0; i < cols.size(); i++) { + columnExprIds.put(cols.get(i).getName(), outputSlots.get(i).getExprId()); + } + + List<ExprId> primaryKeyExprIds = new ArrayList<>(primaryKeys.size()); + for (String primaryKey : primaryKeys) { + primaryKeyExprIds.add(Preconditions.checkNotNull( + columnExprIds.get(primaryKey), + "Paimon primary-key column is missing from sink output")); + } + return PhysicalProperties.createHash(primaryKeyExprIds, ShuffleType.REQUIRE); Review Comment: Fixed in 2932001a30. HASH_FIXED primary-key tables now require a single writer through Gather until a Paimon bucket-aware exchange exists. This avoids splitting one partition/bucket ownership across TableWriteImpl instances, including bucket-key subset cases. The regression plan expectation is updated from hash exchange to gather. -- 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]
