suxiaogang223 commented on code in PR #65868: URL: https://github.com/apache/doris/pull/65868#discussion_r3670992072
########## 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: Not changed to add a process-local lease. Such a lease cannot coordinate FE failover, another FE process, or external Flink/Spark writers, so it would provide a false correctness guarantee. The current implementation gathers HASH_DYNAMIC/KEY_DYNAMIC to one writer within each INSERT and now explicitly documents that concurrent INSERTs into a dynamic-bucket table are unsupported. Proper support requires catalog/distributed locking or fencing and is deferred to a separate feature. ########## 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: Fixed in 7418f8cae7. HASH_FIXED now requires a single writer whenever automatic compaction may run, including append-only tables; write-only append tables remain parallel because they do not restore/replace compaction inputs. PhysicalPaimonTableSinkTest covers both cases as well as fixed PK and dynamic bucket modes. ########## 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: Fixed in 7418f8cae7. The new getPaimonTableForWrite() always loads the latest remote target handle and deliberately bypasses the statement read MVCC snapshot. PaimonWriteBinding, target column types, partition keys, and sink physical-property planning now use that same latest handle, while a historical self-insert source remains pinned only on the read side. ########## 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: Fixed in 7418f8cae7. Struct conversion now binds children by schema position and validates equal field count/order with case-insensitive field-name checks, removing the exact-name lookup that rejected legal MixedCase children while still rejecting ambiguous schema mismatches. PaimonArrowConverterTest adds mixed-case success and ordering-mismatch coverage. -- 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]
