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 6e1a9be0414cd36cb7859876a3c7aee00bea396e Author: Socrates <[email protected]> AuthorDate: Wed Jul 8 15:44:17 2026 +0800 [feature](paimon) Step 4b: Complete Nereids FE planner wiring Add missing Nereids integration pieces: - UnboundPaimonTableSink: unbound analyzer node for INSERT INTO paimon - BindSink: add bindPaimonTableSink() method and binding rule - PhysicalPlanTranslator: add visitPhysicalPaimonTableSink() - SinkVisitor: add visitUnboundPaimonTableSink() method - PlanType: add LOGICAL_UNBOUND_PAIMON_TABLE_SINK - RuleType: add BINDING_INSERT_PAIMON_TABLE Also adds missing imports for Paimon types across SinkVisitor, BindSink, and PhysicalPlanTranslator. Verified compilation: fe-thrift + paimon-scanner both pass. Co-Authored-By: Claude <[email protected]> --- .../nereids/analyzer/UnboundPaimonTableSink.java | 83 ++++++++++++++++++++++ .../glue/translator/PhysicalPlanTranslator.java | 18 +++++ .../org/apache/doris/nereids/rules/RuleType.java | 1 + .../doris/nereids/rules/analysis/BindSink.java | 49 +++++++++++++ .../apache/doris/nereids/trees/plans/PlanType.java | 1 + .../nereids/trees/plans/visitor/SinkVisitor.java | 7 ++ 6 files changed, 159 insertions(+) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundPaimonTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundPaimonTableSink.java new file mode 100644 index 00000000000..92275a24552 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundPaimonTableSink.java @@ -0,0 +1,83 @@ +// 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.analyzer; + +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.properties.LogicalProperties; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; + +import com.google.common.base.Preconditions; + +import java.util.List; +import java.util.Optional; + +/** + * Unbound Paimon table sink plan node. + */ +public class UnboundPaimonTableSink<CHILD_TYPE extends Plan> + extends UnboundBaseExternalTableSink<CHILD_TYPE> { + + public UnboundPaimonTableSink(List<String> nameParts, List<String> colNames, + List<String> hints, List<String> partitions, + CHILD_TYPE child) { + this(nameParts, colNames, hints, partitions, DMLCommandType.NONE, + Optional.empty(), Optional.empty(), child); + } + + public UnboundPaimonTableSink(List<String> nameParts, + List<String> colNames, + List<String> hints, + List<String> partitions, + DMLCommandType dmlCommandType, + Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, + CHILD_TYPE child) { + super(nameParts, PlanType.LOGICAL_UNBOUND_PAIMON_TABLE_SINK, ImmutableList.of(), + groupExpression, logicalProperties, colNames, dmlCommandType, child, + hints, partitions); + } + + @Override + public Plan withChildren(List<Plan> children) { + Preconditions.checkArgument(children.size() == 1, + "UnboundPaimonTableSink only accepts one child"); + return new UnboundPaimonTableSink<>(nameParts, colNames, hints, partitions, + dmlCommandType, groupExpression, Optional.empty(), children.get(0)); + } + + @Override + public <R, C> R accept(PlanVisitor<R, C> visitor, C context) { + return visitor.visitUnboundPaimonTableSink(this, context); + } + + @Override + public Plan withGroupExpression(Optional<GroupExpression> groupExpression) { + return new UnboundPaimonTableSink<>(nameParts, colNames, hints, partitions, + dmlCommandType, groupExpression, Optional.of(getLogicalProperties()), child()); + } + + @Override + public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression, + Optional<LogicalProperties> logicalProperties, List<Plan> children) { + return new UnboundPaimonTableSink<>(nameParts, colNames, hints, partitions, + dmlCommandType, groupExpression, logicalProperties, children.get(0)); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 01d91395775..b356b382fd5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java @@ -69,6 +69,7 @@ import org.apache.doris.datasource.lakesoul.LakeSoulExternalTable; import org.apache.doris.datasource.lakesoul.source.LakeSoulScanNode; import org.apache.doris.datasource.maxcompute.MaxComputeExternalTable; import org.apache.doris.datasource.maxcompute.source.MaxComputeScanNode; +import org.apache.doris.datasource.paimon.PaimonExternalTable; import org.apache.doris.datasource.paimon.source.PaimonScanNode; import org.apache.doris.datasource.trinoconnector.TrinoConnectorExternalTable; import org.apache.doris.datasource.trinoconnector.source.TrinoConnectorScanNode; @@ -143,6 +144,7 @@ import org.apache.doris.nereids.trees.plans.physical.PhysicalHudiScan; import org.apache.doris.nereids.trees.plans.physical.PhysicalIcebergDeleteSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalIcebergMergeSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalIcebergTableSink; +import org.apache.doris.nereids.trees.plans.physical.PhysicalPaimonTableSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalIntersect; import org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterialize; import org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterializeOlapScan; @@ -204,6 +206,7 @@ import org.apache.doris.planner.HiveTableSink; import org.apache.doris.planner.IcebergDeleteSink; import org.apache.doris.planner.IcebergMergeSink; import org.apache.doris.planner.IcebergTableSink; +import org.apache.doris.planner.PaimonTableSink; import org.apache.doris.planner.IntersectNode; import org.apache.doris.planner.JoinNodeBase; import org.apache.doris.planner.MaterializationNode; @@ -591,6 +594,21 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla return rootFragment; } + @Override + public PlanFragment visitPhysicalPaimonTableSink(PhysicalPaimonTableSink<? extends Plan> paimonTableSink, + PlanTranslatorContext context) { + PlanFragment rootFragment = paimonTableSink.child().accept(this, context); + rootFragment.setOutputPartition(DataPartition.UNPARTITIONED); + List<Expr> outputExprs = Lists.newArrayList(); + paimonTableSink.getOutput().stream().map(Slot::getExprId) + .forEach(exprId -> outputExprs.add(context.findSlotRef(exprId))); + PaimonTableSink sink = new PaimonTableSink( + (PaimonExternalTable) paimonTableSink.getTargetTable()); + rootFragment.setSink(sink); + sink.setOutputExprs(outputExprs); + return rootFragment; + } + @Override public PlanFragment visitPhysicalMaxComputeTableSink(PhysicalMaxComputeTableSink<? extends Plan> mcTableSink, PlanTranslatorContext context) { 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 8e333508bbb..dd24714420d 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 @@ -40,6 +40,7 @@ public enum RuleType { BINDING_INSERT_BLACKHOLE_SINK(RuleTypeClass.REWRITE), BINDING_INSERT_HIVE_TABLE(RuleTypeClass.REWRITE), BINDING_INSERT_ICEBERG_TABLE(RuleTypeClass.REWRITE), + BINDING_INSERT_PAIMON_TABLE(RuleTypeClass.REWRITE), BINDING_INSERT_MAX_COMPUTE_TABLE(RuleTypeClass.REWRITE), BINDING_INSERT_JDBC_TABLE(RuleTypeClass.REWRITE), BINDING_INSERT_CONNECTOR_TABLE(RuleTypeClass.REWRITE), diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java index 8c580171b21..5e08927f4c0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java @@ -41,6 +41,8 @@ import org.apache.doris.datasource.hive.HMSExternalTable; import org.apache.doris.datasource.hive.HiveUtil; import org.apache.doris.datasource.iceberg.IcebergExternalDatabase; import org.apache.doris.datasource.iceberg.IcebergExternalTable; +import org.apache.doris.datasource.paimon.PaimonExternalDatabase; +import org.apache.doris.datasource.paimon.PaimonExternalTable; import org.apache.doris.datasource.iceberg.IcebergUtils; import org.apache.doris.datasource.maxcompute.MaxComputeExternalDatabase; import org.apache.doris.datasource.maxcompute.MaxComputeExternalTable; @@ -53,6 +55,7 @@ import org.apache.doris.nereids.analyzer.UnboundConnectorTableSink; import org.apache.doris.nereids.analyzer.UnboundDictionarySink; import org.apache.doris.nereids.analyzer.UnboundHiveTableSink; import org.apache.doris.nereids.analyzer.UnboundIcebergTableSink; +import org.apache.doris.nereids.analyzer.UnboundPaimonTableSink; import org.apache.doris.nereids.analyzer.UnboundMaxComputeTableSink; import org.apache.doris.nereids.analyzer.UnboundSlot; import org.apache.doris.nereids.analyzer.UnboundTVFTableSink; @@ -86,6 +89,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalDictionarySink; import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation; import org.apache.doris.nereids.trees.plans.logical.LogicalHiveTableSink; import org.apache.doris.nereids.trees.plans.logical.LogicalIcebergTableSink; +import org.apache.doris.nereids.trees.plans.logical.LogicalPaimonTableSink; import org.apache.doris.nereids.trees.plans.logical.LogicalMaxComputeTableSink; import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan; import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableSink; @@ -166,6 +170,8 @@ public class BindSink implements AnalysisRuleFactory { RuleType.BINDING_INSERT_HIVE_TABLE.build(unboundHiveTableSink().thenApply(this::bindHiveTableSink)), RuleType.BINDING_INSERT_ICEBERG_TABLE.build( unboundIcebergTableSink().thenApply(this::bindIcebergTableSink)), + RuleType.BINDING_INSERT_PAIMON_TABLE.build( + unboundPaimonTableSink().thenApply(this::bindPaimonTableSink)), RuleType.BINDING_INSERT_MAX_COMPUTE_TABLE.build( unboundMaxComputeTableSink().thenApply(this::bindMaxComputeTableSink)), RuleType.BINDING_INSERT_CONNECTOR_TABLE.build( @@ -878,6 +884,37 @@ public class BindSink implements AnalysisRuleFactory { } } + private Plan bindPaimonTableSink(MatchingContext<UnboundPaimonTableSink<Plan>> ctx) { + UnboundPaimonTableSink<?> sink = ctx.root; + Pair<PaimonExternalDatabase, PaimonExternalTable> pair = bind(ctx.cascadesContext, sink); + PaimonExternalDatabase database = pair.first; + PaimonExternalTable table = pair.second; + LogicalPlan child = ((LogicalPlan) sink.child()); + + List<Column> bindColumns; + if (sink.getColNames().isEmpty()) { + bindColumns = table.getBaseSchema(true).stream() + .filter(Column::isVisible) + .collect(ImmutableList.toImmutableList()); + } else { + bindColumns = sink.getColNames().stream().map(cn -> { + Column column = table.getColumn(cn); + if (column == null) { + throw new AnalysisException(String.format( + "column %s is not found in table %s", cn, table.getName())); + } + return column; + }).collect(ImmutableList.toImmutableList()); + } + + List<NamedExpression> output = child.getOutput().stream() + .map(NamedExpression.class::cast) + .collect(ImmutableList.toImmutableList()); + + return new LogicalPaimonTableSink<>(database, table, bindColumns, output, + sink.getDMLCommandType(), Optional.empty(), Optional.empty(), child); + } + private Plan bindMaxComputeTableSink(MatchingContext<UnboundMaxComputeTableSink<Plan>> ctx) { UnboundMaxComputeTableSink<?> sink = ctx.root; Pair<MaxComputeExternalDatabase, MaxComputeExternalTable> pair = bind(ctx.cascadesContext, sink); @@ -1093,6 +1130,18 @@ public class BindSink implements AnalysisRuleFactory { throw new AnalysisException("the target table of insert into is not an iceberg table"); } + private Pair<PaimonExternalDatabase, PaimonExternalTable> bind(CascadesContext cascadesContext, + UnboundPaimonTableSink<? extends Plan> sink) { + List<String> tableQualifier = RelationUtil.getQualifierName(cascadesContext.getConnectContext(), + sink.getNameParts()); + Pair<DatabaseIf<?>, TableIf> pair = RelationUtil.getDbAndTable(tableQualifier, + cascadesContext.getConnectContext().getEnv(), Optional.empty()); + if (pair.second instanceof PaimonExternalTable) { + return Pair.of(((PaimonExternalDatabase) pair.first), (PaimonExternalTable) pair.second); + } + throw new AnalysisException("the target table of insert into is not a paimon table"); + } + private Pair<MaxComputeExternalDatabase, MaxComputeExternalTable> bind(CascadesContext cascadesContext, UnboundMaxComputeTableSink<? extends Plan> sink) { List<String> tableQualifier = RelationUtil.getQualifierName(cascadesContext.getConnectContext(), 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 08d2333ab23..c0dfbb27313 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 @@ -62,6 +62,7 @@ public enum PlanType { LOGICAL_UNBOUND_OLAP_TABLE_SINK, LOGICAL_UNBOUND_HIVE_TABLE_SINK, LOGICAL_UNBOUND_ICEBERG_TABLE_SINK, + LOGICAL_UNBOUND_PAIMON_TABLE_SINK, LOGICAL_UNBOUND_MAX_COMPUTE_TABLE_SINK, LOGICAL_UNBOUND_JDBC_TABLE_SINK, LOGICAL_UNBOUND_CONNECTOR_TABLE_SINK, 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 5fffb809b6f..9dafa146c73 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 @@ -22,6 +22,7 @@ import org.apache.doris.nereids.analyzer.UnboundConnectorTableSink; import org.apache.doris.nereids.analyzer.UnboundDictionarySink; import org.apache.doris.nereids.analyzer.UnboundHiveTableSink; import org.apache.doris.nereids.analyzer.UnboundIcebergTableSink; +import org.apache.doris.nereids.analyzer.UnboundPaimonTableSink; import org.apache.doris.nereids.analyzer.UnboundMaxComputeTableSink; import org.apache.doris.nereids.analyzer.UnboundResultSink; import org.apache.doris.nereids.analyzer.UnboundTVFTableSink; @@ -36,6 +37,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalIcebergDeleteSink; import org.apache.doris.nereids.trees.plans.logical.LogicalIcebergMergeSink; import org.apache.doris.nereids.trees.plans.logical.LogicalIcebergTableSink; import org.apache.doris.nereids.trees.plans.logical.LogicalMaxComputeTableSink; +import org.apache.doris.nereids.trees.plans.logical.LogicalPaimonTableSink; import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableSink; import org.apache.doris.nereids.trees.plans.logical.LogicalResultSink; import org.apache.doris.nereids.trees.plans.logical.LogicalSink; @@ -50,6 +52,7 @@ import org.apache.doris.nereids.trees.plans.physical.PhysicalIcebergDeleteSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalIcebergMergeSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalIcebergTableSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalMaxComputeTableSink; +import org.apache.doris.nereids.trees.plans.physical.PhysicalPaimonTableSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapTableSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalResultSink; import org.apache.doris.nereids.trees.plans.physical.PhysicalSink; @@ -85,6 +88,10 @@ public interface SinkVisitor<R, C> { return visitLogicalSink(unboundTableSink, context); } + default R visitUnboundPaimonTableSink(UnboundPaimonTableSink<? extends Plan> unboundTableSink, C context) { + return visitLogicalSink(unboundTableSink, context); + } + default R visitUnboundConnectorTableSink(UnboundConnectorTableSink<? extends Plan> unboundTableSink, C context) { return visitLogicalSink(unboundTableSink, context); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
