This is an automated email from the ASF dual-hosted git repository. suxiaogang223 pushed a commit to branch feature/paimon-jni-write-v1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 1f7b10dc79952683a4c329390e6a4e323977f81c Author: Socrates <[email protected]> AuthorDate: Wed Jul 8 15:20:48 2026 +0800 [feature](paimon) Step 4: FE Planner - Nereids integration (partial) Add Nereids plan nodes and planner support for Paimon write: - LogicalPaimonTableSink and PhysicalPaimonTableSink - LogicalPaimonTableSinkToPhysicalPaimonTableSink implementation rule - PaimonTableSink planner (generates TPaimonTableSink Thrift) - PlanType, SinkVisitor, RuleType enum entries NOTE: FE compilation could not be verified due to Maven repository connectivity issues. The following wiring remains for a complete build: - BindSink.java: add bindPaimonTableSink method - PhysicalPlanTranslator.java: add visitPhysicalPaimonTableSink - UnboundPaimonTableSink.java: create stub/analyzer node - PaimonTransaction.java: FE-side commit coordinator Co-Authored-By: Claude <[email protected]> --- .../org/apache/doris/nereids/rules/RuleType.java | 1 + ...alPaimonTableSinkToPhysicalPaimonTableSink.java | 48 +++++ .../apache/doris/nereids/trees/plans/PlanType.java | 2 + .../plans/logical/LogicalPaimonTableSink.java | 149 ++++++++++++++ .../plans/physical/PhysicalPaimonTableSink.java | 117 +++++++++++ .../nereids/trees/plans/visitor/SinkVisitor.java | 8 + .../org/apache/doris/planner/PaimonTableSink.java | 227 +++++++++++++++++++++ 7 files changed, 552 insertions(+) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java index 1e4f38fd005..8e333508bbb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java @@ -561,6 +561,7 @@ public enum RuleType { LOGICAL_OLAP_TABLE_SINK_TO_PHYSICAL_OLAP_TABLE_SINK_RULE(RuleTypeClass.IMPLEMENTATION), LOGICAL_HIVE_TABLE_SINK_TO_PHYSICAL_HIVE_TABLE_SINK_RULE(RuleTypeClass.IMPLEMENTATION), LOGICAL_ICEBERG_TABLE_SINK_TO_PHYSICAL_ICEBERG_TABLE_SINK_RULE(RuleTypeClass.IMPLEMENTATION), + LOGICAL_PAIMON_TABLE_SINK_TO_PHYSICAL_PAIMON_TABLE_SINK_RULE(RuleTypeClass.IMPLEMENTATION), LOGICAL_MAX_COMPUTE_TABLE_SINK_TO_PHYSICAL_MAX_COMPUTE_TABLE_SINK_RULE(RuleTypeClass.IMPLEMENTATION), LOGICAL_ICEBERG_DELETE_SINK_TO_PHYSICAL_ICEBERG_DELETE_SINK_RULE(RuleTypeClass.IMPLEMENTATION), LOGICAL_ICEBERG_MERGE_SINK_TO_PHYSICAL_ICEBERG_MERGE_SINK_RULE(RuleTypeClass.IMPLEMENTATION), diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalPaimonTableSinkToPhysicalPaimonTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalPaimonTableSinkToPhysicalPaimonTableSink.java new file mode 100644 index 00000000000..02438c24628 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalPaimonTableSinkToPhysicalPaimonTableSink.java @@ -0,0 +1,48 @@ +// 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.rules.implementation; + +import org.apache.doris.nereids.rules.Rule; +import org.apache.doris.nereids.rules.RuleType; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalPaimonTableSink; +import org.apache.doris.nereids.trees.plans.physical.PhysicalPaimonTableSink; + +import java.util.Optional; + +/** + * Implementation rule that converts LogicalPaimonTableSink to PhysicalPaimonTableSink. + */ +public class LogicalPaimonTableSinkToPhysicalPaimonTableSink extends OneImplementationRuleFactory { + @Override + public Rule build() { + return logicalPaimonTableSink().thenApply(ctx -> { + LogicalPaimonTableSink<? extends Plan> sink = ctx.root; + return new PhysicalPaimonTableSink<>( + sink.getDatabase(), + sink.getTargetTable(), + sink.getCols(), + sink.getOutputExprs(), + Optional.empty(), + sink.getLogicalProperties(), + null, + null, + sink.child()); + }).toRule(RuleType.LOGICAL_PAIMON_TABLE_SINK_TO_PHYSICAL_PAIMON_TABLE_SINK_RULE); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java index 2f5dcbb7cfc..08d2333ab23 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java @@ -50,6 +50,7 @@ public enum PlanType { LOGICAL_OLAP_TABLE_SINK, LOGICAL_HIVE_TABLE_SINK, LOGICAL_ICEBERG_TABLE_SINK, + LOGICAL_PAIMON_TABLE_SINK, LOGICAL_MAX_COMPUTE_TABLE_SINK, LOGICAL_ICEBERG_DELETE_SINK, LOGICAL_ICEBERG_MERGE_SINK, @@ -125,6 +126,7 @@ public enum PlanType { PHYSICAL_OLAP_TABLE_SINK, PHYSICAL_HIVE_TABLE_SINK, PHYSICAL_ICEBERG_TABLE_SINK, + PHYSICAL_PAIMON_TABLE_SINK, PHYSICAL_MAX_COMPUTE_TABLE_SINK, PHYSICAL_ICEBERG_DELETE_SINK, PHYSICAL_ICEBERG_MERGE_SINK, diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalPaimonTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalPaimonTableSink.java new file mode 100644 index 00000000000..0bcc586a7b1 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalPaimonTableSink.java @@ -0,0 +1,149 @@ +// 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.logical; + +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.LogicalProperties; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.plans.AbstractPlan; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.algebra.Sink; +import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.nereids.util.Utils; + +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; + +import java.util.List; +import java.util.Objects; +import java.util.Optional; + +/** + * Logical Paimon table sink for INSERT INTO paimon_table. + */ +public class LogicalPaimonTableSink<CHILD_TYPE extends Plan> extends LogicalTableSink<CHILD_TYPE> + implements Sink { + + private final PaimonExternalDatabase database; + private final PaimonExternalTable targetTable; + private final DMLCommandType dmlCommandType; + + public LogicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + DMLCommandType dmlCommandType, + Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, + CHILD_TYPE child) { + super(PlanType.LOGICAL_PAIMON_TABLE_SINK, outputExprs, groupExpression, logicalProperties, + cols, child); + this.database = Objects.requireNonNull(database, "database != null"); + this.targetTable = Objects.requireNonNull(targetTable, "targetTable != null"); + this.dmlCommandType = dmlCommandType; + } + + public Plan withChildAndUpdateOutput(Plan child) { + List<NamedExpression> output = child.getOutput().stream() + .map(NamedExpression.class::cast) + .collect(ImmutableList.toImmutableList()); + return AbstractPlan.copyWithSameId(this, () -> + new LogicalPaimonTableSink<>(database, targetTable, cols, output, + dmlCommandType, Optional.empty(), Optional.empty(), child)); + } + + @Override + public Plan withChildren(List<Plan> children) { + Preconditions.checkArgument(children.size() == 1, + "LogicalPaimonTableSink only accepts one child"); + return AbstractPlan.copyWithSameId(this, () -> + new LogicalPaimonTableSink<>(database, targetTable, cols, outputExprs, + dmlCommandType, Optional.empty(), Optional.empty(), children.get(0))); + } + + public LogicalPaimonTableSink<CHILD_TYPE> withOutputExprs(List<NamedExpression> outputExprs) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalPaimonTableSink<>(database, targetTable, cols, outputExprs, + dmlCommandType, Optional.empty(), Optional.empty(), child())); + } + + public PaimonExternalDatabase getDatabase() { + return database; + } + + public PaimonExternalTable getTargetTable() { + return targetTable; + } + + public DMLCommandType getDmlCommandType() { + return dmlCommandType; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + if (!super.equals(o)) return false; + LogicalPaimonTableSink<?> that = (LogicalPaimonTableSink<?>) o; + return dmlCommandType == that.dmlCommandType + && Objects.equals(database, that.database) + && Objects.equals(targetTable, that.targetTable) + && Objects.equals(cols, that.cols); + } + + @Override + public int hashCode() { + return Objects.hash(super.hashCode(), database, targetTable, cols, dmlCommandType); + } + + @Override + public String toString() { + return Utils.toSqlString("LogicalPaimonTableSink[" + id.asInt() + "]", + "outputExprs", outputExprs, + "database", database.getFullName(), + "targetTable", targetTable.getName(), + "cols", cols, + "dmlCommandType", dmlCommandType); + } + + @Override + public <R, C> R accept(PlanVisitor<R, C> visitor, C context) { + return visitor.visitLogicalPaimonTableSink(this, context); + } + + @Override + public Plan withGroupExpression(Optional<GroupExpression> groupExpression) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalPaimonTableSink<>(database, targetTable, cols, outputExprs, + dmlCommandType, groupExpression, + Optional.of(getLogicalProperties()), child())); + } + + @Override + public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, List<Plan> children) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalPaimonTableSink<>(database, targetTable, cols, outputExprs, + dmlCommandType, groupExpression, logicalProperties, children.get(0))); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java new file mode 100644 index 00000000000..4eae65bee82 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalPaimonTableSink.java @@ -0,0 +1,117 @@ +// 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.LogicalProperties; +import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.plans.AbstractPlan; +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.qe.ConnectContext; + +import java.util.List; +import java.util.Optional; + +/** + * 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.GATHER, null, child); + } + + public PhysicalPaimonTableSink(PaimonExternalDatabase database, + PaimonExternalTable targetTable, + List<Column> cols, + List<NamedExpression> outputExprs, + Optional<GroupExpression> groupExpression, + LogicalProperties logicalProperties, + PhysicalProperties physicalProperties, + ConnectContext connectContext, + CHILD_TYPE child) { + super(PlanType.PHYSICAL_PAIMON_TABLE_SINK, outputExprs, groupExpression, + logicalProperties, physicalProperties, connectContext, cols, child); + this.database = database; + this.targetTable = targetTable; + } + + private final PaimonExternalDatabase database; + private final PaimonExternalTable targetTable; + + public PaimonExternalDatabase getDatabase() { + return database; + } + + public PaimonExternalTable getTargetTable() { + return targetTable; + } + + @Override + public Plan withChildren(List<Plan> children) { + return AbstractPlan.copyWithSameId(this, () -> + new PhysicalPaimonTableSink<>(database, targetTable, cols, outputExprs, + Optional.empty(), getLogicalProperties(), + physicalProperties, connectContext, children.get(0))); + } + + @Override + public Plan withGroupExpression(Optional<GroupExpression> groupExpression) { + return AbstractPlan.copyWithSameId(this, () -> + new PhysicalPaimonTableSink<>(database, targetTable, cols, outputExprs, + groupExpression, getLogicalProperties(), + physicalProperties, connectContext, child())); + } + + @Override + public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, List<Plan> children) { + return AbstractPlan.copyWithSameId(this, () -> + new PhysicalPaimonTableSink<>(database, targetTable, cols, outputExprs, + groupExpression, logicalProperties.get(), + physicalProperties, connectContext, children.get(0))); + } + + @Override + public PhysicalPaimonTableSink<Plan> withPhysicalPropertiesAndStats( + PhysicalProperties physicalProperties, org.apache.doris.statistics.Statistics stats) { + return AbstractPlan.copyWithSameId(this, () -> + new PhysicalPaimonTableSink<>(database, targetTable, cols, outputExprs, + groupExpression, getLogicalProperties(), + physicalProperties, connectContext, child())); + } + + @Override + public <R, C> R accept(PlanVisitor<R, C> visitor, C context) { + return visitor.visitPhysicalPaimonTableSink(this, context); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/SinkVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/SinkVisitor.java index dcc6f715c9e..5fffb809b6f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/SinkVisitor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/SinkVisitor.java @@ -133,6 +133,10 @@ public interface SinkVisitor<R, C> { return visitLogicalTableSink(icebergTableSink, context); } + default R visitLogicalPaimonTableSink(LogicalPaimonTableSink<? extends Plan> paimonTableSink, C context) { + return visitLogicalTableSink(paimonTableSink, context); + } + default R visitLogicalMaxComputeTableSink(LogicalMaxComputeTableSink<? extends Plan> mcTableSink, C context) { return visitLogicalTableSink(mcTableSink, context); } @@ -197,6 +201,10 @@ public interface SinkVisitor<R, C> { return visitPhysicalTableSink(icebergTableSink, context); } + default R visitPhysicalPaimonTableSink(PhysicalPaimonTableSink<? extends Plan> paimonTableSink, C context) { + return visitPhysicalTableSink(paimonTableSink, context); + } + default R visitPhysicalMaxComputeTableSink(PhysicalMaxComputeTableSink<? extends Plan> mcTableSink, C context) { return visitPhysicalTableSink(mcTableSink, context); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java new file mode 100644 index 00000000000..55bdead55f0 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PaimonTableSink.java @@ -0,0 +1,227 @@ +// 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.planner; + +import org.apache.doris.analysis.Expr; +import org.apache.doris.catalog.Column; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.datasource.mvcc.MvccUtil; +import org.apache.doris.datasource.paimon.PaimonExternalCatalog; +import org.apache.doris.datasource.paimon.PaimonExternalTable; +import org.apache.doris.nereids.trees.plans.commands.insert.BaseExternalTableInsertCommandContext; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertCommandContext; +import org.apache.doris.thrift.TDataSink; +import org.apache.doris.thrift.TDataSinkType; +import org.apache.doris.thrift.TExplainLevel; +import org.apache.doris.thrift.TFileFormatType; +import org.apache.doris.thrift.TPaimonTableSink; +import org.apache.doris.thrift.TPaimonWriteBackendType; +import org.apache.doris.thrift.TPaimonWriteMode; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.utils.InstantiationUtil; + +import java.util.ArrayList; +import java.util.Base64; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** + * Paimon table sink. + * + * Generates TPaimonTableSink payload consumed by BE, including table location, + * Paimon options, Hadoop config, bucket metadata, and sink column names. + * + * v1: always sets backend_type = JNI. + */ +public class PaimonTableSink extends BaseExternalTableDataSink { + private static final Logger LOG = LogManager.getLogger(PaimonTableSink.class); + private static final String OUTPUT_COLUMN_NAMES_KEY = "doris.output_column_names"; + private static final String COLUMN_NAME_SEPARATOR = ""; + private static final Base64.Encoder BASE64_ENCODER = + java.util.Base64.getUrlEncoder().withoutPadding(); + + private final PaimonExternalTable targetTable; + private List<Expr> outputExprs; + private List<Column> cols; + + private static final HashSet<TFileFormatType> supportedTypes = new HashSet<TFileFormatType>() {{ + add(TFileFormatType.FORMAT_ORC); + add(TFileFormatType.FORMAT_PARQUET); + }}; + + public PaimonTableSink(PaimonExternalTable targetTable) { + super(); + this.targetTable = targetTable; + } + + public void setCols(List<Column> cols) { + this.cols = cols; + } + + public void setOutputExprs(List<Expr> outputExprs) { + this.outputExprs = outputExprs; + } + + @Override + protected Set<TFileFormatType> supportedFileFormatTypes() { + return supportedTypes; + } + + @Override + public String getExplainString(String prefix, TExplainLevel explainLevel) { + StringBuilder strBuilder = new StringBuilder(); + strBuilder.append(prefix).append("PAIMON TABLE SINK\n"); + if (explainLevel == TExplainLevel.BRIEF) { + return strBuilder.toString(); + } + strBuilder.append(prefix).append(" table: ").append(targetTable.getName()).append("\n"); + return strBuilder.toString(); + } + + @Override + public void bindDataSink(Optional<InsertCommandContext> insertCtx) throws AnalysisException { + TPaimonTableSink tSink = new TPaimonTableSink(); + + tSink.setDbName(targetTable.getDbName()); + tSink.setTbName(targetTable.getName()); + + Map<String, String> hadoopConfig = new HashMap<>( + targetTable.getCatalog().getCatalogProperty().getHadoopProperties()); + Map<String, String> paimonOptions = new HashMap<>(); + + String warehouse = ((PaimonExternalCatalog) targetTable.getCatalog()) + .getPaimonOptionsMap().get(CatalogOptions.WAREHOUSE.key()); + String defaultFs = resolveDefaultFsName(warehouse); + if (defaultFs != null && !defaultFs.isEmpty()) { + String current = hadoopConfig.get("fs.defaultFS"); + if (current == null || current.isEmpty() || current.startsWith("file:/")) { + hadoopConfig.put("fs.defaultFS", defaultFs); + } + } + + // Transaction context: commit identifier and user + if (insertCtx.isPresent() && insertCtx.get() instanceof BaseExternalTableInsertCommandContext) { + BaseExternalTableInsertCommandContext ctx = + (BaseExternalTableInsertCommandContext) insertCtx.get(); + if (ctx.getTxnId() > 0) { + paimonOptions.put("doris.commit_identifier", String.valueOf(ctx.getTxnId())); + } + if (ctx.getCommitUser() != null && !ctx.getCommitUser().isEmpty()) { + paimonOptions.put("doris.commit_user", ctx.getCommitUser()); + } + } + + // Column names (BE-side column name preservation) + paimonOptions.put(OUTPUT_COLUMN_NAMES_KEY, String.join(COLUMN_NAME_SEPARATOR, outputColumnNames())); + + // Hadoop user + String hadoopUser = hadoopConfig.get("hadoop.username"); + if (hadoopUser == null || hadoopUser.isEmpty()) { + hadoopUser = hadoopConfig.get("hadoop.user.name"); + } + if (hadoopUser == null || hadoopUser.isEmpty()) { + hadoopUser = "hadoop"; + } + hadoopConfig.put("hadoop.user.name", hadoopUser); + + // Table location + String tableLocation = null; + org.apache.paimon.table.Table paimonTable = + targetTable.getPaimonTable(MvccUtil.getSnapshotFromContext(targetTable)); + if (paimonTable instanceof org.apache.paimon.table.FileStoreTable) { + tableLocation = ((org.apache.paimon.table.FileStoreTable) paimonTable).location().toString(); + } + if (tableLocation != null && !tableLocation.isEmpty()) { + tSink.setTableLocation(tableLocation); + } + + // Serialize table for BE-side fast loading + if (paimonTable != null) { + tSink.setSerializedTable(encodeObjectToString(paimonTable)); + } + + // Bucket info + if (paimonTable instanceof org.apache.paimon.table.FileStoreTable) { + int bucketNum = ((org.apache.paimon.table.FileStoreTable) paimonTable).schema().numBuckets(); + if (bucketNum > 0) { + tSink.setBucketNum(bucketNum); + } + } + + // v1: always JNI backend, APPEND mode + tSink.setBackendType(TPaimonWriteBackendType.JNI); + tSink.setWriteMode(TPaimonWriteMode.APPEND); + + tSink.setPaimonOptions(paimonOptions); + tSink.setHadoopConfig(hadoopConfig); + + List<String> columnNames = new ArrayList<>(); + for (Column col : cols) { + columnNames.add(col.getName()); + } + tSink.setColumnNames(columnNames); + + tDataSink = new TDataSink(TDataSinkType.PAIMON_TABLE_SINK); + tDataSink.setPaimonTableSink(tSink); + } + + private List<String> outputColumnNames() throws AnalysisException { + List<Column> fullSchema = targetTable.getFullSchema(); + if (fullSchema.size() != outputExprs.size()) { + throw new AnalysisException("Paimon sink output column size mismatch, schema=" + + fullSchema.size() + ", exprs=" + outputExprs.size()); + } + List<String> names = new ArrayList<>(fullSchema.size()); + for (Column col : fullSchema) { + names.add(col.getName()); + } + return names; + } + + private static String encodeObjectToString(Object obj) { + try { + byte[] bytes = InstantiationUtil.serializeObject(obj); + return new String(BASE64_ENCODER.encode(bytes), + java.nio.charset.StandardCharsets.UTF_8); + } catch (Exception e) { + throw new RuntimeException("Failed to serialize Paimon table", e); + } + } + + private static String resolveDefaultFsName(String warehouse) { + if (warehouse == null || warehouse.isEmpty()) return null; + try { + java.net.URI uri = java.net.URI.create(warehouse); + String scheme = uri.getScheme(); + String authority = uri.getAuthority(); + if (scheme != null && !scheme.isEmpty() && authority != null && !authority.isEmpty()) { + return scheme + "://" + authority; + } + } catch (Exception e) { + LOG.warn("paimon: invalid warehouse uri {}", warehouse); + } + return null; + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
