suxiaogang223 commented on code in PR #65868:
URL: https://github.com/apache/doris/pull/65868#discussion_r3670991350
##########
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:
Fixed in 7418f8cae7. PaimonExternalTable.getNewestUpdateVersionOrTime() now
returns the latest Paimon snapshot ID, which is the same monotonic version
recorded by getTableSnapshot(). It no longer derives freshness from the Doris
partition-pruning map, so an UNPRUNABLE LTZ partition map cannot make the
version fall back to 0.
##########
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:
Not changed in this PR. A finite standalone RootAllocator would create a
second independent limit instead of charging Arrow vectors and Paimon pages to
one budget, so their combined allocation could still exceed the advertised
writer/query limit. The correct fix requires shared reserve/release accounting
across JNI with the Doris native memory manager. An explicit TODO was added at
PaimonJniWriter construction; this is kept as a follow-up rather than adding a
partial and misleading cap.
##########
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:
Fixed in 7418f8cae7. GlobalIndexAssigner now validates
dynamic-bucket.target-row-num before narrowing: values must be in [1,
Integer.MAX_VALUE], otherwise writer construction fails with a clear error.
GlobalIndexAssignerTest covers 1, Integer.MAX_VALUE, zero, and
Integer.MAX_VALUE + 1.
--
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]
