github-actions[bot] commented on code in PR #65868:
URL: https://github.com/apache/doris/pull/65868#discussion_r3668682153
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java:
##########
@@ -150,71 +145,125 @@ 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());
+ RowType partitionType = table.rowType().project(table.partitionKeys());
+ if (partitionType.getFields().stream().anyMatch(field ->
+ field.type().getTypeRoot() ==
DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
+ // This metadata is cached by table snapshot, but LTZ values are
represented as
+ // session-local civil times in Doris. Caching those bounds would
let one session
+ // reuse pruning metadata produced in another time zone. Keep scan
correctness by
+ // delegating pruning to Paimon until this cache can carry a
time-zone-independent
+ // typed representation.
+ return PaimonPartitionInfo.UNPRUNABLE;
Review Comment:
[P1] Preserve dictionary freshness for unprunable LTZ partitions
This singleton also clears `nameToPartition`, but Paimon dictionary loading
records the positive snapshot ID from `getTableSnapshot()`. On the next
freshness check, `getNewestUpdateVersionOrTime()` reads this empty map and
returns 0, so `Dictionary.hasNewerSourceVersion()` throws because 0 is smaller
than the saved version. Please keep non-pruning freshness metadata here, or
make both dictionary paths use the snapshot ID consistently; otherwise every
dictionary backed by an LTZ-partitioned Paimon table breaks after its first
successful load.
##########
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<>();
+
+ private int bucketIndex;
+ private int targetBucketRowNumber;
+ private int assignId;
+ private int numAssigners;
+ private boolean bootstrapping;
+ private BiConsumer<InternalRow, Integer> collector;
+ private PartitionKeyExtractor<InternalRow> extractor;
+ private PartitionKeyExtractor<InternalRow> bootstrapExtractor;
+ private IDMapping<BinaryRow> partitionMapping;
+ private BucketAssigner bucketAssigner;
+ private ExistingProcessor existingProcessor;
+
+ GlobalIndexAssigner(FileStoreTable table) {
+ this.table = table;
+ }
+
+ void open(
+ int numAssigners,
+ int assignId,
+ BiConsumer<InternalRow, Integer> collector) {
+ this.numAssigners = numAssigners;
+ this.assignId = assignId;
+ this.collector = collector;
+
+ CoreOptions coreOptions = table.coreOptions();
+ this.bucketIndex =
IndexBootstrap.bootstrapType(table.schema()).getFieldCount() - 1;
+ this.targetBucketRowNumber = (int)
coreOptions.dynamicBucketTargetRowNum();
Review Comment:
[P2] Validate the long dynamic-bucket target before narrowing it
Paimon exposes `dynamic-bucket.target-row-num` as a `long` with no
`Integer.MAX_VALUE` ceiling, so a valid value such as 2147483648 wraps to a
negative number here. `BucketAssigner` only reuses a bucket when `count <
maxCount`; with the wrapped value that is never true, and every new key creates
another bucket. Please preserve the long threshold or reject/saturate
out-of-range values before opening this writer, and cover the int boundary.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -0,0 +1,656 @@
+// 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.MemoryPoolFactory;
+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 (row-at-a-time 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 ClassLoader classLoader;
+ private final PaimonCommitCodec commitCodec = new PaimonCommitCodec();
+
+ private BufferAllocator allocator;
+ 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();
+ private boolean sdkCloseFailed;
+
+ public PaimonJniWriter() {
+ this.allocator = new RootAllocator(Long.MAX_VALUE);
Review Comment:
[P1] Include Arrow decoding in the writer memory budget
`ArrowStreamReader` allocates its loaded vectors from this allocator, so
`Long.MAX_VALUE` gives every sink writer an effectively unbounded off-heap
decoder outside the Doris query tracker. The new native limit only covers
`DorisMemorySegmentPool`, and the C++ projected block plus IPC buffer are still
live during this decode. A wide block or several parallel writers can therefore
exhaust BE direct memory without tripping the advertised writer/query cap.
Please use a bounded allocator charged to the same per-writer budget, and fail
the query before process-level exhaustion.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java:
##########
@@ -0,0 +1,173 @@
+// 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);
+ }
+
+ /**
+ * Whether this sink must use one writer to preserve Paimon write
semantics.
+ */
+ public boolean requiresSingleWriter() {
+ Table paimonTable = ((PaimonExternalTable) targetTable)
+ .getPaimonTable(Optional.empty());
+ Preconditions.checkState(paimonTable instanceof FileStoreTable,
+ "Paimon write requires a file store table");
+ FileStoreTable fileStoreTable = (FileStoreTable) paimonTable;
+ BucketMode bucketMode = fileStoreTable.bucketMode();
+ if (bucketMode == BucketMode.HASH_DYNAMIC
+ || bucketMode == BucketMode.KEY_DYNAMIC
+ // Until Doris has a Paimon bucket-aware exchange, a
fixed-bucket
+ // primary-key table must not let independent writers own the
same
+ // partition/bucket. Hashing the complete primary key is
insufficient
+ // when bucket-key is a subset and can also split compaction
ownership.
+ || (bucketMode == BucketMode.HASH_FIXED
+ && !paimonTable.primaryKeys().isEmpty())) {
+ return true;
+ }
+
+ CoreOptions coreOptions = CoreOptions.fromMap(paimonTable.options());
Review Comment:
[P1] Keep one writer per fixed append bucket during compaction
An ordinary `HASH_FIXED` append table still reaches random sink distribution
here, even though each local Paimon writer restores the existing files for
every bucket it touches and may compact them. Once a bucket has enough old
files, two local writers can select the same compaction inputs; FE then
combines both replacement sets, and Paimon's manifest conflict check rejects
the duplicate deletes, failing an otherwise valid INSERT. Please gather these
writes while auto-compaction is enabled, or exchange by the exact Paimon
partition and bucket so each bucket has one owner.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -0,0 +1,656 @@
+// 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.MemoryPoolFactory;
+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 (row-at-a-time 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 ClassLoader classLoader;
+ private final PaimonCommitCodec commitCodec = new PaimonCommitCodec();
+
+ private BufferAllocator allocator;
+ 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();
+ private boolean sdkCloseFailed;
+
+ 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
+ * @param memoryPoolLimitBytes maximum Doris-managed Paimon write-buffer
memory
+ * @param nativeMemoryManager opaque BE manager used to allocate tracked
native pages
+ */
+ public void open(String serializedTable, Map<String, String> hadoopConfig,
+ String[] columnNames, long transactionId, String
commitUser,
+ boolean overwrite, String timeZone, String
spillDirectories,
+ long memoryPoolLimitBytes, long nativeMemoryManager)
throws Exception {
+ try (ThreadClassLoaderContext ignored = new
ThreadClassLoaderContext(classLoader)) {
+ if (memoryPoolLimitBytes <= 0) {
+ throw new IllegalArgumentException(
+ "PaimonJniWriter requires a positive memory pool
limit");
+ }
+ if (nativeMemoryManager == 0) {
+ throw new IllegalArgumentException(
+ "PaimonJniWriter requires a native memory manager");
+ }
+ this.preExecutionAuthenticator =
PreExecutionAuthenticatorCache.getAuthenticator(hadoopConfig);
+ this.arrowConverter = new
PaimonArrowConverter(ZoneId.of(timeZone));
+ preExecutionAuthenticator.execute(() -> {
+ try {
+ FileStoreTable table =
PaimonUtils.deserialize(serializedTable);
+ LOG.info("PaimonJniWriter opening: table={}, columns={}",
+ table.fullName(), columnNames != null ?
columnNames.length : 0);
+ this.commitIdentifier = transactionId;
+ this.table = table;
+ this.commitUser = commitUser;
+ this.bucketMode = table.bucketMode();
+
+ CoreOptions coreOptions =
CoreOptions.fromMap(table.options());
+ this.writeSchema =
PaimonWriteSchema.create(table.rowType(), columnNames);
+ validateWriteColumnsForMergeEngine(columnNames.length,
coreOptions);
+ this.fullCompactionChangelog =
+ !coreOptions.writeOnly()
+ && coreOptions.changelogProducer()
+ ==
CoreOptions.ChangelogProducer.FULL_COMPACTION;
+ openFileStoreWriter(
+ table,
+ commitUser,
+ overwrite,
+ spillDirectories,
+ coreOptions,
+ memoryPoolLimitBytes,
+ nativeMemoryManager);
+ return null;
+ } catch (Throwable t) {
+ try {
+ closeResources();
+ } catch (Throwable closeFailure) {
+ t.addSuppressed(closeFailure);
+ }
+ 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 writer and bucket assigner APIs. The SDK
+ * owns partition/bucket semantics, 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 {
+ try (ThreadClassLoaderContext ignored = new
ThreadClassLoaderContext(classLoader)) {
+ 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 {
+ try (ThreadClassLoaderContext ignored = new
ThreadClassLoaderContext(classLoader)) {
+ 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 (ThreadClassLoaderContext ignored = new
ThreadClassLoaderContext(classLoader)) {
+ 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 (ThreadClassLoaderContext ignored = new
ThreadClassLoaderContext(classLoader)) {
+ 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,
+ String spillDirectories, CoreOptions coreOptions, long
memoryPoolLimitBytes,
+ long nativeMemoryManager) throws Exception {
+ writer = table.newWrite(commitUser);
+ if (overwrite) {
+ writer.withIgnorePreviousFiles(true);
+ }
+ openMemoryResources(
+ coreOptions, spillDirectories, memoryPoolLimitBytes,
nativeMemoryManager);
+ openDynamicBucketAssigner(table, commitUser, overwrite, coreOptions);
+ }
+
+ private void validateWriteColumnsForMergeEngine(int writeColumnCount,
CoreOptions coreOptions) {
+ if (writeColumnCount == table.rowType().getFieldCount() ||
table.primaryKeys().isEmpty()) {
+ return;
+ }
+
+ CoreOptions.MergeEngine mergeEngine = coreOptions.mergeEngine();
+ if (mergeEngine != CoreOptions.MergeEngine.PARTIAL_UPDATE) {
+ throw new UnsupportedOperationException(
+ "Paimon primary-key partial-column write requires "
+ + "merge-engine=partial-update, but table uses
merge-engine="
+ + mergeEngine);
+ }
+ }
+
+ private void openMemoryResources(
+ CoreOptions coreOptions,
+ String spillDirectories,
+ long memoryPoolLimitBytes,
+ long nativeMemoryManager) throws Exception {
+ int pageSize = coreOptions.pageSize();
+ long effectivePoolLimit = Math.min(coreOptions.writeBufferSize(),
memoryPoolLimitBytes);
+ DorisMemorySegmentPool memorySegmentPool =
+ new DorisMemorySegmentPool(effectivePoolLimit, pageSize,
nativeMemoryManager);
+ MemoryPoolFactory memoryPoolFactory = new
MemoryPoolFactory(memorySegmentPool);
+ writer.withMemoryPoolFactory(memoryPoolFactory);
+ LOG.info("Paimon writer uses Doris-managed memory pool: limit={}
bytes, pageSize={}",
+ memoryPoolFactory.totalBufferSize(), pageSize);
+
+ if (!coreOptions.writeBufferSpillable()) {
+ return;
+ }
+
+ String[] splitDirectories = IOManagerImpl.splitPaths(spillDirectories);
+ for (String directory : splitDirectories) {
+ Files.createDirectories(Paths.get(directory));
+ }
+ ioManager = IOManager.create(splitDirectories);
+ writer.withIOManager(ioManager);
+ LOG.info("Paimon writer spill enabled: dirs={}", spillDirectories);
+ }
+
+ private void openDynamicBucketAssigner(FileStoreTable table, String
commitUser,
Review Comment:
[P1] Prevent overlapping dynamic-bucket write transactions
Gather only reduces one statement to one local writer. Two concurrent Doris
INSERTs still open independent hash assigners or bootstrap independent global
indexes for the same table/partition, with no table-scoped guard through
commit. Paimon explicitly supports only one dynamic-bucket write job per
partition because overlapping jobs can create duplicate data. Please serialize
or reject overlapping HASH_DYNAMIC/KEY_DYNAMIC transactions through
commit/rollback, and add a two-session concurrent INSERT test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonWriteBinding.java:
##########
@@ -0,0 +1,243 @@
+// 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.AnalysisException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.datasource.mvcc.MvccUtil;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import
org.apache.doris.nereids.trees.plans.commands.insert.PaimonInsertCommandContext;
+import org.apache.doris.nereids.types.DataType;
+
+import com.google.common.base.Preconditions;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypeRoot;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * Transaction-scoped binding between one Doris insert and one concrete Paimon
table handle.
+ *
+ * <p>The table and writer configuration are captured exactly once while the
sink is finalized.
+ * BE preparation, FE commit, abort and outcome reconciliation therefore all
refer to the same
+ * Paimon table handle.
+ */
+public class PaimonWriteBinding {
+ private final PaimonExternalTable dorisTable;
+ private final FileStoreTable table;
+ private final String serializedTable;
+ private final Map<String, String> hadoopConfig;
+ private final boolean overwrite;
+ private final Map<String, String> staticPartition;
+
+ private PaimonWriteBinding(PaimonExternalTable dorisTable, FileStoreTable
table,
+ Map<String, String> hadoopConfig, boolean overwrite,
+ Map<String, String> staticPartition) {
+ this.dorisTable = dorisTable;
+ this.table = table;
+ this.serializedTable = PaimonUtil.encodeObjectToString(table);
+ this.hadoopConfig = Collections.unmodifiableMap(new
HashMap<>(hadoopConfig));
+ this.overwrite = overwrite;
+ this.staticPartition = Collections.unmodifiableMap(new
LinkedHashMap<>(staticPartition));
+ }
+
+ public static PaimonWriteBinding create(PaimonExternalTable dorisTable,
+ PaimonInsertCommandContext context) throws UserException {
+ PaimonExternalCatalog catalog = (PaimonExternalCatalog)
dorisTable.getCatalog();
+ FileStoreTable table;
+ try {
+ table = catalog.getExecutionAuthenticator().execute(() ->
requireFileStoreTable(
+
dorisTable.getPaimonTable(MvccUtil.getSnapshotFromContext(dorisTable))));
Review Comment:
[P1] Do not bind the write target to a source time-travel snapshot
`StatementContext` stores MVCC snapshots by table identity, so in a
self-insert such as `INSERT INTO t SELECT ... FROM t FOR VERSION AS OF ...`,
this lookup returns the source's historical Paimon table and serializes it as
the target writer. Target columns and physical planning were derived from the
latest table; after schema evolution the BE therefore opens an old row type and
can reject valid target columns (or apply stale write-mode assumptions). Please
pin a separate latest write-target handle instead of reusing the source read
snapshot, and add a schema-evolved time-travel self-insert test.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonArrowConverter.java:
##########
@@ -0,0 +1,348 @@
+// 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) {
+ byte[] value = ((VarCharVector) vector).get(index);
+ return targetType instanceof BinaryType || targetType instanceof
VarBinaryType
+ ? value : BinaryString.fromBytes(value);
+ }
+ if (vector instanceof VarBinaryVector) {
+ return ((VarBinaryVector) vector).get(index);
+ }
+ if (vector instanceof TimeStampVector) {
+ ArrowType.Timestamp timestampType = (ArrowType.Timestamp)
arrowField.getType();
+ return toPaimonTimestamp(
+ arrowTimestampToMicros(((TimeStampVector)
vector).get(index), timestampType),
+ timestampType, targetType);
+ }
+ if (vector instanceof DecimalVector) {
+ DecimalVector decimalVector = (DecimalVector) vector;
+ int precision = decimalVector.getPrecision();
+ int scale = decimalVector.getScale();
+ BigDecimal decimal = getBigDecimalFromArrowBuf(
+ decimalVector.getDataBuffer(), index, scale,
DecimalVector.TYPE_WIDTH);
+ return Decimal.fromBigDecimal(decimal, precision, scale);
+ }
+ return convertToPaimonType(vector.getObject(index), arrowField,
targetType);
+ }
+
+ 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 toPaimonTimestamp((LocalDateTime) value, targetType);
+ }
+ 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());
+ }
+ return value;
+ }
+
+ private GenericRow convertStructVector(
+ StructVector vector, int index, RowType rowType) {
+ List<DataField> childFields = rowType.getFields();
+ GenericRow row = new GenericRow(childFields.size());
+ for (int i = 0; i < childFields.size(); i++) {
+ DataField childField = childFields.get(i);
+ FieldVector childVector = findChildVector(vector,
childField.name());
Review Comment:
[P1] Preserve nested Paimon field names across the Arrow bridge
Doris `StructField` lowercases every nested name, so a legal Paimon child
such as `MixedCase` reaches this Arrow struct as `mixedcase`. This lookup then
uses the original Paimon name with exact equality and throws `Arrow struct does
not contain Paimon field MixedCase` for every write. Please carry the canonical
Paimon child names into the write schema, or use ambiguity-checked
case-insensitive matching, and cover mixed-case ROW children (including nested
array/map cases).
--
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]
