github-actions[bot] commented on code in PR #65868: URL: https://github.com/apache/doris/pull/65868#discussion_r3664730946
########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java: ########## @@ -0,0 +1,617 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.paimon; + +import org.apache.doris.common.classloader.ThreadClassLoaderContext; +import org.apache.doris.common.security.authentication.PreExecutionAuthenticator; +import org.apache.doris.common.security.authentication.PreExecutionAuthenticatorCache; + +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.ipc.ArrowStreamReader; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.crosspartition.IndexBootstrap; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.disk.IOManagerImpl; +import org.apache.paimon.index.BucketAssigner; +import org.apache.paimon.index.HashBucketAssigner; +import org.apache.paimon.index.SimpleHashBucketAssigner; +import org.apache.paimon.memory.HeapMemorySegmentPool; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.InnerTableCommit; +import org.apache.paimon.table.sink.PartitionKeyExtractor; +import org.apache.paimon.table.sink.RowPartitionKeyExtractor; +import org.apache.paimon.table.sink.SinkRecord; +import org.apache.paimon.table.sink.TableWriteImpl; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * JNI entry point for Paimon write operations. + * + * <p>Called from C++ ({@code JniPaimonWriter}) via JNI. One instance per BE pipeline + * fragment (one per {@code PaimonTableWriter}). Data path: + * + * <pre> + * C++ Block → Arrow IPC Stream → JNI direct ByteBuffer + * → PaimonJniWriter.write(directBuffer) + * → ArrowStreamReader → VectorSchemaRoot + * → PaimonArrowConverter (column-major typed extraction) + * → PaimonWriteSchema.tableRow() (canonical table-schema order) + * → Paimon SDK bucket assignment and table write + * </pre> + * + * <p>Commit path: + * + * <pre> + * PaimonTableWriter::close() → JNI → PaimonJniWriter.prepareCommit() + * → TableWriteImpl.prepareCommit() + * → PaimonCommitCodec.encode() → DPCM-framed byte[][] + * → C++ collects TPaimonCommitMessage[] → RPC to FE → PaimonTransaction + * </pre> + */ +public class PaimonJniWriter { + private static final Logger LOG = LoggerFactory.getLogger(PaimonJniWriter.class); + + private final BufferAllocator allocator; + private final ClassLoader classLoader; + private final PaimonCommitCodec commitCodec = new PaimonCommitCodec(); + + private PreExecutionAuthenticator preExecutionAuthenticator; + private PaimonArrowConverter arrowConverter; + + private PaimonWriteSchema writeSchema; + private FileStoreTable table; + private TableWriteImpl<?> writer; + private IOManager ioManager; + private long commitIdentifier; + private String commitUser; + private BucketMode bucketMode; + private BucketAssigner hashBucketAssigner; + private PartitionKeyExtractor<InternalRow> dynamicBucketExtractor; + private GlobalIndexAssigner globalIndexAssigner; + private boolean fullCompactionChangelog; + private final Set<PartitionBucket> fullCompactionBuckets = new HashSet<>(); + private List<CommitMessage> preparedCommitMessages = Collections.emptyList(); + + public PaimonJniWriter() { + this.allocator = new RootAllocator(Long.MAX_VALUE); + this.classLoader = this.getClass().getClassLoader(); + } + + // ──────────────────────────────────────────────────────────── + // JNI entry points (called from C++) + // ──────────────────────────────────────────────────────────── + + /** + * Initialize the writer. Called once per BE pipeline fragment via JNI. + * + * <p>This method: + * <ol> + * <li>Deserializes the target Paimon {@link FileStoreTable} selected by FE.</li> + * <li>Creates a {@link PaimonWriteSchema} which normalizes Doris input + * columns to the table-schema row layout.</li> + * <li>Opens one Paimon SDK writer session.</li> + * </ol> + * + * @param serializedTable serialized Paimon table selected by FE + * @param hadoopConfig filesystem and authentication configuration + * @param columnNames output column names in the order produced by BE + * @param transactionId Doris external transaction identifier + * @param commitUser Paimon commit user shared with the FE committer + * @param overwrite whether this is an overwrite write + * @param timeZone normalized Doris session timezone used for Paimon LTZ values + * @param spillDirectories Doris storage-root scoped directories for Paimon write-buffer spill + */ + public void open(String serializedTable, Map<String, String> hadoopConfig, + String[] columnNames, long transactionId, String commitUser, + boolean overwrite, String timeZone, String spillDirectories) throws Exception { + try (ThreadClassLoaderContext ignored = new ThreadClassLoaderContext(classLoader)) { + this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(hadoopConfig); + this.arrowConverter = new PaimonArrowConverter(ZoneId.of(timeZone)); Review Comment: [P1] Normalize Doris time-zone aliases before opening the writer `SET time_zone = 'EST'` is valid because FE resolves Java short IDs, but only `CST` is canonicalized before the BE/JNI bridge. The unchanged `EST` reaches this unconditional one-argument `ZoneId.of`, which throws `ZoneRulesException` before any row is written (including non-temporal rows). Please pass a canonical zone/offset or resolve aliases with the same semantics as Doris, and cover an alias other than CST. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonWriteBinding.java: ########## @@ -0,0 +1,183 @@ +// 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.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 com.google.common.base.Preconditions; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.table.FileStoreTable; + +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)))); + } catch (Exception e) { + throw new UserException("Failed to bind Paimon table for write", e); + } + Map<String, Expression> typedStaticPartition = context.getStaticPartition(); + Map<String, String> staticPartition = resolveStaticPartition(table, typedStaticPartition); + return new PaimonWriteBinding( + dorisTable, + table, + buildHadoopConfig(catalog), + context.isOverwrite(), + staticPartition); + } + + public FileStoreTable getTable() { + return table; + } + + public String getSerializedTable() { + return serializedTable; + } + + public Map<String, String> getHadoopConfig() { + return hadoopConfig; + } + + public boolean isOverwrite() { + return overwrite; + } + + public Map<String, String> getStaticPartition() { + return staticPartition; + } + + public String tableName() { + return dorisTable.getDbName() + "." + dorisTable.getName(); + } + + private static FileStoreTable requireFileStoreTable(org.apache.paimon.table.Table table) { + Preconditions.checkState(table instanceof FileStoreTable, + "Paimon write requires a file store table"); + return (FileStoreTable) table; + } + + private static Map<String, String> resolveStaticPartition(FileStoreTable table, + Map<String, Expression> typedStaticPartition) throws AnalysisException { + Map<String, String> canonicalNames = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (String partitionKey : table.partitionKeys()) { + canonicalNames.put(partitionKey, partitionKey); + } + + String defaultPartitionName = CoreOptions.fromMap(table.options()).partitionDefaultName(); + Map<String, String> resolved = new LinkedHashMap<>(); + for (Map.Entry<String, Expression> entry : typedStaticPartition.entrySet()) { + String canonicalName = canonicalNames.get(entry.getKey()); + if (canonicalName == null) { + throw new AnalysisException("Column '" + entry.getKey() + + "' is not a partition column of Paimon table"); + } + Expression value = entry.getValue(); + if (!(value instanceof Literal)) { + throw new AnalysisException("Static partition value must be a literal, but got: " + + value); + } + String partitionValue = value instanceof NullLiteral + ? defaultPartitionName : ((Literal) value).getStringValue(); Review Comment: [P1] Preserve typed identity in the static partition spec For a string partition column, SQL `NULL` and a literal equal to `partition.default-name` are distinct typed partitions (the new read-side code handles that distinction), but both become the same string here. The committer then maps that marker to null, so `PARTITION(p='<default-name>')` overwrites the null partition instead of the literal one. Please carry typed partition values through commit, or reject this collision if the current Paimon API cannot represent it. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java: ########## @@ -0,0 +1,167 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.trees.plans.physical; + +import org.apache.doris.catalog.Column; +import org.apache.doris.datasource.paimon.PaimonExternalDatabase; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType; +import org.apache.doris.nereids.properties.LogicalProperties; +import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.statistics.Statistics; + +import com.google.common.base.Preconditions; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.BucketMode; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.TreeMap; + +/** + * Physical Paimon table sink. + */ +public class PhysicalPaimonTableSink<CHILD_TYPE extends Plan> + extends PhysicalBaseExternalTableSink<CHILD_TYPE> { + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + CHILD_TYPE child) { + this(database, targetTable, cols, outputExprs, groupExpression, logicalProperties, + PhysicalProperties.SINK_RANDOM_PARTITIONED, null, child); + } + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + PhysicalProperties physicalProperties, + Statistics statistics, + CHILD_TYPE child) { + super(PlanType.PHYSICAL_PAIMON_TABLE_SINK, database, targetTable, cols, outputExprs, + groupExpression, logicalProperties, physicalProperties, statistics, child); + } + + @Override + public Plan withChildren(List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, children.get(0)); + } + + @Override + public Plan withGroupExpression(Optional<GroupExpression> groupExpression) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, statistics, child()); + } + + @Override + public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, List<Plan> children) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + logicalProperties.get(), physicalProperties, statistics, children.get(0)); + } + + @Override + public PhysicalPaimonTableSink<Plan> withPhysicalPropertiesAndStats( + PhysicalProperties physicalProperties, Statistics stats) { + return new PhysicalPaimonTableSink<>( + (PaimonExternalDatabase) database, (PaimonExternalTable) targetTable, + cols, outputExprs, groupExpression, + getLogicalProperties(), physicalProperties, stats, child()); + } + + @Override + public PhysicalProperties getRequirePhysicalProperties() { + if (requiresSingleWriter()) { + return PhysicalProperties.GATHER; + } + + PaimonExternalTable paimonExternalTable = (PaimonExternalTable) targetTable; + Table paimonTable = paimonExternalTable.getPaimonTable(Optional.empty()); + List<String> primaryKeys = paimonTable.primaryKeys(); + if (primaryKeys.isEmpty()) { + return PhysicalProperties.SINK_RANDOM_PARTITIONED; + } + + List<Slot> outputSlots = child().getOutput(); + Preconditions.checkState(cols.size() == outputSlots.size(), + "Paimon sink columns must match child output"); + Map<String, ExprId> columnExprIds = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (int i = 0; i < cols.size(); i++) { + columnExprIds.put(cols.get(i).getName(), outputSlots.get(i).getExprId()); + } + + List<ExprId> primaryKeyExprIds = new ArrayList<>(primaryKeys.size()); + for (String primaryKey : primaryKeys) { + primaryKeyExprIds.add(Preconditions.checkNotNull( + columnExprIds.get(primaryKey), + "Paimon primary-key column is missing from sink output")); + } + return PhysicalProperties.createHash(primaryKeyExprIds, ShuffleType.REQUIRE); Review Comment: [P1] Route fixed-bucket rows by Paimon bucket ownership Hashing the full primary key only co-locates identical keys. Distinct keys can share a fixed bucket—especially when `bucket-key` is a PK subset—after Doris has already split them across independent `TableWriteImpl`s. That violates Paimon's one-writer ownership for a `(partition, bucket)` group and can create sequence/compaction conflicts. Please exchange by Paimon's actual partition-and-bucket selector, or gather fixed-bucket primary-key writes until a bucket-aware exchange exists. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/GlobalIndexAssigner.java: ########## @@ -0,0 +1,167 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.paimon; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.crosspartition.BucketAssigner; +import org.apache.paimon.crosspartition.ExistingProcessor; +import org.apache.paimon.crosspartition.IndexBootstrap; +import org.apache.paimon.crosspartition.KeyPartPartitionKeyExtractor; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.PartitionKeyExtractor; +import org.apache.paimon.table.sink.RowPartitionAllPrimaryKeyExtractor; +import org.apache.paimon.utils.IDMapping; +import org.apache.paimon.utils.PositiveIntInt; +import org.apache.paimon.utils.ProjectToRowFunction; + +import java.util.HashMap; +import java.util.Map; +import java.util.function.BiConsumer; + +/** + * Assigns buckets for a key-dynamic table with a process-local global-key index. + * + * <p>Doris gathers key-dynamic writes into one writer, so a single in-memory index can preserve + * Paimon's cross-partition merge semantics without a native state backend. + */ +final class GlobalIndexAssigner implements AutoCloseable { + private final FileStoreTable table; + // TODO: After resolving rocksdbjni allocator compatibility with the Doris BE jemalloc hook, + // use a RocksDB-backed on-disk index to bound Java heap usage for large tables. + private final Map<BinaryRow, PositiveIntInt> keyIndex = new HashMap<>(); Review Comment: [P1] Bound the key-dynamic global index Every key-dynamic writer open bootstraps every existing live key into this `HashMap`, and later copies new keys into it for the writer's lifetime. There is no spill path, cap, eviction, Doris memory accounting, or early capacity check, so a tiny insert into a sufficiently large table can exhaust the embedded JVM heap and take down the BE. Please use a disk-backed/bounded index, or reject/cap this mode until one is available. ########## be/src/exec/pipeline/pipeline_fragment_context.cpp: ########## @@ -2175,6 +2184,20 @@ void PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r } } + if (auto pcm = req.runtime_state->paimon_commit_messages(); !pcm.empty()) { Review Comment: [P1] Bound the aggregate commit payload sent in one report The 8 MiB codec limit applies only to each element, but this appends every chunk from the writer—or every task-local writer—to one `TReportExecStatusParams`. Thirteen near-limit chunks already exceed Doris's default 100 MiB Thrift message limit, so a metadata-heavy write fails final reporting after `prepareCommit`; FE receives no payloads and cannot abort those prepared files. Please split payloads across transaction-scoped reports with an explicit completion boundary (or use another bounded transport), and make an undeliverable payload fail while the writer can still abort. ########## fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java: ########## @@ -392,6 +392,10 @@ private void setExternalTableAutoAnalyzePolicy(ExternalTable table, List<AlterCl private void processAlterTableForExternalTable( ExternalTable table, List<AlterClause> alterClauses) throws UserException { + if (alterClauses.size() > 1) { Review Comment: [P2] Scope the multi-clause restriction to Paimon This method is also the shared ALTER path for HMS, JDBC, Iceberg, MaxCompute, Hudi, and Trino, and it previously dispatched every compatible clause. The unconditional guard therefore rejects existing non-Paimon behavior—for example, Iceberg implements and commits multiple schema operations—even though this PR only establishes the restriction for Paimon. Please scope the check to Paimon or validate an intentional all-catalog compatibility change separately. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java: ########## @@ -150,71 +144,115 @@ public static List<InternalRow> read( return rows; } - public static PaimonPartitionInfo generatePartitionInfo(List<Column> partitionColumns, - List<Partition> paimonPartitions, boolean legacyPartitionName) { + public static PaimonPartitionInfo generatePartitionInfo(Table table, List<Column> partitionColumns, + List<PartitionEntry> partitionEntries) { - if (CollectionUtils.isEmpty(partitionColumns) || paimonPartitions.isEmpty()) { + if (CollectionUtils.isEmpty(partitionColumns) || partitionEntries.isEmpty()) { return PaimonPartitionInfo.EMPTY; } - Map<String, PartitionItem> nameToPartitionItem = Maps.newHashMap(); - Map<String, Partition> nameToPartition = Maps.newHashMap(); - PaimonPartitionInfo partitionInfo = new PaimonPartitionInfo(nameToPartitionItem, nameToPartition); + CoreOptions options = new CoreOptions(table.options()); + InternalRowPartitionComputer partitionComputer = new InternalRowPartitionComputer( + options.partitionDefaultName(), + table.rowType().project(table.partitionKeys()), + table.partitionKeys().toArray(new String[0]), + options.legacyPartitionName()); List<Type> types = partitionColumns.stream() .map(Column::getType) .collect(Collectors.toList()); + List<PaimonPartitionCandidate> candidates = Lists.newArrayListWithExpectedSize(partitionEntries.size()); + Map<String, Integer> displayNameCounts = Maps.newHashMap(); + + for (PartitionEntry partitionEntry : partitionEntries) { + Map<String, String> typedSpec = getPartitionInfoMap( + table, partitionEntry.partition(), TimeUtils.getTimeZone().getID()); Review Comment: [P1] Keep session-zoned LTZ bounds out of the global cache This turns each LTZ partition instant into civil time in the current Doris session zone, but the resulting partition info is memoized per table without a zone in the cache key. A UTC preload can therefore leave `00:00` cached for an Asia/Shanghai session that should see `08:00`, allowing its predicate to prune the matching partition. Please cache zone-independent identity, include the effective zone in the key, or treat LTZ partitions as unprunable until evaluation is session-local. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonMetadataOps.java: ########## @@ -408,6 +409,237 @@ public boolean databaseExist(String dbName) { } } + private Identifier tableIdentifier(ExternalTable dorisTable) { + return Identifier.create(dorisTable.getRemoteDbName(), dorisTable.getRemoteName()); + } + + private List<DataField> loadRemoteFields(ExternalTable dorisTable) throws UserException { + try { + return executionAuthenticator.execute( + () -> new ArrayList<>(catalog.getTable(tableIdentifier(dorisTable)).rowType().getFields())); + } catch (Exception e) { + throw new UserException("Failed to load schema for Paimon table " + dorisTable.getName() + + ": " + ExceptionUtils.getRootCauseMessage(e), e); + } + } + + private Map<String, DataField> indexFieldsByDorisName(List<DataField> fields) throws UserException { + Map<String, DataField> fieldsByLowerCase = new HashMap<>(); + for (DataField field : fields) { + DataField previous = fieldsByLowerCase.put(field.name().toLowerCase(Locale.ROOT), field); + if (previous != null) { + throw new UserException("Paimon table contains columns which differ only by case: " + + previous.name() + " and " + field.name()); + } + } + return fieldsByLowerCase; + } + + private DataField resolveRemoteField(Map<String, DataField> fieldsByDorisName, String columnName) + throws UserException { + DataField field = fieldsByDorisName.get(columnName.toLowerCase(Locale.ROOT)); + if (field == null) { + throw new UserException("Column " + columnName + " does not exist in Paimon table"); + } + return field; + } + + private DataType toPaimonColumnType(Column column) throws UserException { + try { + return toPaimonType(column.getType()).copy(column.isAllowNull()); + } catch (RuntimeException e) { + throw new UserException("Unsupported Paimon type for column " + column.getName() + + ": " + ExceptionUtils.getRootCauseMessage(e), e); + } + } + + private void registerDorisColumnName(Set<String> columnNames, String columnName) throws UserException { + if (!columnNames.add(columnName.toLowerCase(Locale.ROOT))) { + throw new UserException("Column " + columnName + + " conflicts with an existing Paimon column (case-insensitive)"); + } + } + + private void checkUnsupportedColumnAttributes(Column column) throws UserException { + if (column.isAggregated()) { + throw new UserException("Paimon column does not support aggregation method: " + column.getName()); + } + if (column.isAutoInc()) { + throw new UserException("Paimon column does not support AUTO_INCREMENT: " + column.getName()); + } + if (column.isGeneratedColumn()) { + throw new UserException("Column " + column.getName() + + " cannot be a generated column in a Paimon table"); + } + } + + private void appendAddColumnChanges(List<SchemaChange> changes, Column column, SchemaChange.Move move) + throws UserException { + changes.add(SchemaChange.addColumn( + column.getName(), toPaimonColumnType(column), column.getComment(), move)); + if (column.getDefaultValue() != null) { + changes.add(SchemaChange.updateColumnDefaultValue( + new String[] {column.getName()}, column.getDefaultValue())); + } + } + + private void alterTable(ExternalTable dorisTable, List<SchemaChange> changes, String operation, + long updateTime) throws UserException { + if (changes.isEmpty()) { + return; + } + try { + executionAuthenticator.execute(() -> { + catalog.alterTable(tableIdentifier(dorisTable), changes, false); + return null; + }); + } catch (Exception e) { + throw new UserException("Failed to " + operation + " for Paimon table " + dorisTable.getName() + + ": " + ExceptionUtils.getRootCauseMessage(e), e); + } + refreshTable(dorisTable, updateTime); + } + + private void refreshTable(ExternalTable dorisTable, long updateTime) { + Optional<ExternalDatabase<?>> db = dorisCatalog.getDbForReplay(dorisTable.getDbName()); + if (db.isPresent()) { + Optional<?> table = db.get().getTableForReplay(dorisTable.getName()); + if (table.isPresent()) { + Env.getCurrentEnv().getRefreshManager() + .refreshTableInternal(db.get(), (ExternalTable) table.get(), updateTime); + } + } + } + + @Override + public void addColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime) + throws UserException { + List<DataField> fields = loadRemoteFields(dorisTable); + Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(fields); + registerDorisColumnName(new HashSet<>(fieldsByDorisName.keySet()), column.getName()); + checkUnsupportedColumnAttributes(column); + + SchemaChange.Move move = null; + if (position != null) { + move = position.isFirst() + ? SchemaChange.Move.first(column.getName()) + : SchemaChange.Move.after(column.getName(), + resolveRemoteField(fieldsByDorisName, position.getLastCol()).name()); + } + + List<SchemaChange> changes = new ArrayList<>(); + appendAddColumnChanges(changes, column, move); + alterTable(dorisTable, changes, "add column " + column.getName(), updateTime); + } + + @Override + public void addColumns(ExternalTable dorisTable, List<Column> columns, long updateTime) throws UserException { + Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(loadRemoteFields(dorisTable)); + Set<String> columnNames = new HashSet<>(fieldsByDorisName.keySet()); + List<SchemaChange> changes = new ArrayList<>(); + for (Column column : columns) { + registerDorisColumnName(columnNames, column.getName()); + checkUnsupportedColumnAttributes(column); + appendAddColumnChanges(changes, column, null); + } + alterTable(dorisTable, changes, "add columns", updateTime); + } + + @Override + public void dropColumn(ExternalTable dorisTable, String columnName, long updateTime) throws UserException { + Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(loadRemoteFields(dorisTable)); + String remoteColumnName = resolveRemoteField(fieldsByDorisName, columnName).name(); + alterTable(dorisTable, Collections.singletonList(SchemaChange.dropColumn(remoteColumnName)), + "drop column " + remoteColumnName, updateTime); + } + + @Override + public void renameColumn(ExternalTable dorisTable, String oldName, String newName, long updateTime) + throws UserException { + Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(loadRemoteFields(dorisTable)); + DataField oldField = resolveRemoteField(fieldsByDorisName, oldName); + DataField conflictingField = fieldsByDorisName.get(newName.toLowerCase(Locale.ROOT)); + if (conflictingField != null && conflictingField != oldField) { + throw new UserException("Column " + newName + + " conflicts with an existing Paimon column (case-insensitive)"); + } + if (oldField.name().equals(newName)) { + return; + } + alterTable(dorisTable, + Collections.singletonList(SchemaChange.renameColumn(oldField.name(), newName)), + "rename column " + oldField.name() + " to " + newName, updateTime); + } + + @Override + public void modifyColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime) + throws UserException { + checkUnsupportedColumnAttributes(column); + Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(loadRemoteFields(dorisTable)); + DataField currentField = resolveRemoteField(fieldsByDorisName, column.getName()); + DataType requestedType = toPaimonColumnType(column); Review Comment: [P1] Do not rebuild the remote type from a lossy projection `column` is a Doris projection of the remote type, but the reverse visitor is not lossless: it maps every char family and VARBINARY to MAX length, resets `DATETIMEV2` to default timestamp precision, can lose binary roots/nested nullability, turns default-mapped LTZ into NTZ, and rejects mapped `TIMESTAMPTZ`. Thus a comment-, position-, default-, or nullability-only MODIFY can silently add an unrelated type change or fail. Please preserve `currentField.type()` whenever the repeated Doris type is an equivalent projection, retain all facets for genuine type changes, and reject ambiguous conversions. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonCommitCodec.java: ########## @@ -0,0 +1,123 @@ +// 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.io.DataOutputSerializer; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageSerializer; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** + * Encodes Paimon commit messages into the DPCM (Doris-Paimon Commit Message) payload + * format forwarded to FE. + * + * <h3>DPCM framing format</h3> + * Each payload is framed as: + * <pre> + * ┌──────────┬─────────────┬────────────┬──────────────────────┐ + * │ Magic (4)│ Version (4) │ Length (4) │ Serialized Messages │ + * │ "DPCM" │ big-endian │ big-endian │ (varies) │ + * └──────────┴─────────────┴────────────┴──────────────────────┘ + * </pre> + * + * <p>Messages are serialized using Paimon's {@link CommitMessageSerializer} and + * split into chunks if the serialized payload exceeds {@link #MAX_PAYLOAD_BYTES} + * (8 MiB). Chunk size starts at {@link #DEFAULT_CHUNK_SIZE} (512 messages) and + * is halved adaptively until each chunk fits within the size limit. + */ +final class PaimonCommitCodec { + static final int HEADER_BYTES = 12; + /** Maximum payload size per chunk (8 MiB) to stay under gRPC message limits. */ + static final int MAX_PAYLOAD_BYTES = 8 * 1024 * 1024; + /** Starting number of commit messages per chunk. */ + static final int DEFAULT_CHUNK_SIZE = 512; + + private final CommitMessageSerializer serializer = new CommitMessageSerializer(); + + /** + * Encode commit messages into DPCM-framed byte chunks. + * + * @param messages Paimon commit messages from {@code prepareCommit()} + * @return byte[][] where each element is a complete DPCM-framed chunk + */ + byte[][] encode(List<CommitMessage> messages) throws Exception { + if (messages.isEmpty()) { + return new byte[0][]; + } + + // Adaptive chunking: start with DEFAULT_CHUNK_SIZE messages per chunk; + // if the serialized chunk exceeds MAX_PAYLOAD_BYTES, halve the chunk + // size and retry until each chunk fits or chunkSize reaches 1. + int chunkSize = DEFAULT_CHUNK_SIZE; + List<byte[]> payloads = new ArrayList<>(); + int offset = 0; + while (offset < messages.size()) { + int end = Math.min(offset + chunkSize, messages.size()); + byte[] payload = encodeChunk(messages.subList(offset, end)); Review Comment: [P1] Enforce the chunk limit while serializing This serializes the initial slice of up to 512 messages before observing its size. `encodeChunk` keeps the growable serializer buffer live, `getCopyOfBuffer()` makes a full copy, and `frame()` makes another before line 75 checks 8 MiB and retries smaller. A metadata-heavy batch can therefore allocate several copies of hundreds of MiB and OOM the embedded JVM before adaptive splitting. Please use a size-limited/counting output or incremental bounded encoding and avoid the double copy; add a multi-message peak-allocation test. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonWriteBinding.java: ########## @@ -0,0 +1,183 @@ +// 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.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 com.google.common.base.Preconditions; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.table.FileStoreTable; + +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)))); + } catch (Exception e) { + throw new UserException("Failed to bind Paimon table for write", e); + } + Map<String, Expression> typedStaticPartition = context.getStaticPartition(); + Map<String, String> staticPartition = resolveStaticPartition(table, typedStaticPartition); + return new PaimonWriteBinding( + dorisTable, + table, + buildHadoopConfig(catalog), + context.isOverwrite(), + staticPartition); + } + + public FileStoreTable getTable() { + return table; + } + + public String getSerializedTable() { + return serializedTable; + } + + public Map<String, String> getHadoopConfig() { + return hadoopConfig; + } + + public boolean isOverwrite() { + return overwrite; + } + + public Map<String, String> getStaticPartition() { + return staticPartition; + } + + public String tableName() { + return dorisTable.getDbName() + "." + dorisTable.getName(); + } + + private static FileStoreTable requireFileStoreTable(org.apache.paimon.table.Table table) { + Preconditions.checkState(table instanceof FileStoreTable, + "Paimon write requires a file store table"); + return (FileStoreTable) table; + } + + private static Map<String, String> resolveStaticPartition(FileStoreTable table, + Map<String, Expression> typedStaticPartition) throws AnalysisException { + Map<String, String> canonicalNames = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (String partitionKey : table.partitionKeys()) { + canonicalNames.put(partitionKey, partitionKey); + } + + String defaultPartitionName = CoreOptions.fromMap(table.options()).partitionDefaultName(); + Map<String, String> resolved = new LinkedHashMap<>(); + for (Map.Entry<String, Expression> entry : typedStaticPartition.entrySet()) { + String canonicalName = canonicalNames.get(entry.getKey()); + if (canonicalName == null) { + throw new AnalysisException("Column '" + entry.getKey() + + "' is not a partition column of Paimon table"); + } + Expression value = entry.getValue(); + if (!(value instanceof Literal)) { + throw new AnalysisException("Static partition value must be a literal, but got: " + + value); + } + String partitionValue = value instanceof NullLiteral + ? defaultPartitionName : ((Literal) value).getStringValue(); Review Comment: [P1] Build the overwrite spec from the value actually written `BindSink` writes the target-type cast of a static literal, but this serializes the original expression. For example, an INT partition with `p=TRUE` writes `1` while `withOverwrite` receives `"true"`; a BOOL partition with `p=1` writes true while the spec remains `"1"`. The LTZ case also writes using the Doris session zone but reparses this unqualified text in the FE JVM default zone. Commit can therefore fail or target a different partition from the row being written. Please resolve the literal once with the target Paimon type and Doris zone, then use that same typed/canonical value for both paths (or reject it before files are prepared). -- 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]
