suxiaogang223 commented on code in PR #68320:
URL: https://github.com/apache/doris/pull/68320#discussion_r4119986204
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromUsingCommand.java:
##########
@@ -68,6 +78,12 @@ protected LogicalPlan handleCte(LogicalPlan logicalPlan) {
return logicalPlan;
}
+ @Override
+ protected RowLevelDmlArgs rowLevelDmlArgs(TableIf table) {
+ return RowLevelDmlArgs.forDelete(
+ table, nameParts, tableAlias, isTempPart, partitions,
handleCte(logicalQuery), true);
Review Comment:
Addressed in cc0bb9923d5 and refined in 88cdad9b760. DeleteFromUsingCommand
no longer assumes every instance is a real USING delete and no longer carries a
separate syntax boolean. It derives target-row fanout from
LogicalJoin/LogicalGenerate, so real USING (including UNNEST) deduplicates
while CTE, ORDER BY/LIMIT, and fallback DELETE paths retain the ordinary
contract. The focused tests cover all four forms.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSink.java:
##########
@@ -370,4 +452,56 @@ public PhysicalProperties getRequirePhysicalProperties() {
}
return PhysicalProperties.GATHER;
}
+
+ private PhysicalProperties toPhysicalProperties(ConnectorWriteDistribution
distribution) {
+ switch (distribution.getMode()) {
+ case EXECUTION_ANY:
+ return PhysicalProperties.EXECUTION_ANY;
+ case GATHER:
+ return PhysicalProperties.GATHER;
+ case HASH:
+ return PhysicalProperties.createHash(
Review Comment:
Addressed in cc0bb9923d5. Connector HASH now requires
ShuffleType.EXECUTION_BUCKETED rather than REQUIRE.
PhysicalConnectorTableSinkTest verifies both the required shuffle type and that
a NATURAL hash property cannot satisfy it.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalConnectorTableSink.java:
##########
@@ -370,4 +452,56 @@ public PhysicalProperties getRequirePhysicalProperties() {
}
return PhysicalProperties.GATHER;
}
+
+ private PhysicalProperties toPhysicalProperties(ConnectorWriteDistribution
distribution) {
+ switch (distribution.getMode()) {
+ case EXECUTION_ANY:
+ return PhysicalProperties.EXECUTION_ANY;
+ case GATHER:
+ return PhysicalProperties.GATHER;
+ case HASH:
+ return PhysicalProperties.createHash(
+ routeExprIds(distribution.getRouteColumns()),
ShuffleType.REQUIRE);
+ case EXTERNAL_UNPARTITIONED:
+ requireExternalWriterRoutingSupport();
+ return PhysicalProperties.EXTERNAL_TABLE_SINK_UNPARTITIONED;
+ case EXTERNAL_HASH:
+ requireExternalWriterRoutingSupport();
+ return new PhysicalProperties(new
DistributionSpecExternalTableSinkHashPartitioned(
+ routeExprIds(distribution.getRouteColumns()),
+ distribution.getPartitionFunction(),
+ distribution.getPartitionFunctionOptions(),
+ distribution.getWriterAssignment()));
+ default:
+ throw new IllegalStateException("Unsupported connector write
distribution: "
+ + distribution.getMode());
+ }
+ }
+
+ private void requireExternalWriterRoutingSupport() {
+ Preconditions.checkState(Config.be_exec_version
Review Comment:
Version 13 is the intended compatibility boundary. It was reserved on master
before this PR by #66809 (commit 3adeb25b8ba) specifically for strict
external-table-sink hash routing and Paimon fixed-bucket routing; that
reservation raised the supported maximum from 12 to 13 without landing the
implementation. This PR consumes that pre-reserved version, so no additional
version bump is needed.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/ConnectorChangelogPlanBuilder.java:
##########
@@ -0,0 +1,584 @@
+// 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.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.connector.spi.write.ConnectorChangelogMode;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.analyzer.Scope;
+import org.apache.doris.nereids.analyzer.UnboundAlias;
+import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.LessThanEqual;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Not;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.WindowExpression;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AnyValue;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.AssertTrue;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral;
+import org.apache.doris.nereids.trees.plans.commands.ConnectorWriteSchemaUtils;
+import
org.apache.doris.nereids.trees.plans.commands.info.ConnectorChangelogRowChangeSpec;
+import org.apache.doris.nereids.trees.plans.commands.merge.MergeMatchedClause;
+import
org.apache.doris.nereids.trees.plans.commands.merge.MergeNotMatchedClause;
+import org.apache.doris.nereids.trees.plans.commands.merge.MergeUtils;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.types.BigIntType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.TypeCoercionUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+/** Builds the operation-column plus full-row projection used by
changelog-oriented connectors. */
+public final class ConnectorChangelogPlanBuilder {
+ private static final String BRANCH_LABEL = "__DORIS_CHANGELOG_BRANCH__";
+
+ private ConnectorChangelogPlanBuilder() {
+ }
+
+ /** Builds a changelog plan for the requested connector row-level
operation. */
+ public static LogicalPlan build(List<Column> schema, ExternalTable
targetTable,
+ List<String> primaryKeys,
+ ConnectorChangelogMode mode, ConnectorChangelogRowChangeSpec spec,
+ LogicalPlan child, CascadesContext context) {
+ if (spec instanceof ConnectorChangelogRowChangeSpec.Update) {
+ return buildUpdate(schema, targetTable, mode,
+ (ConnectorChangelogRowChangeSpec.Update) spec, child,
context);
+ }
+ if (spec instanceof ConnectorChangelogRowChangeSpec.Delete) {
+ return buildDelete(schema, primaryKeys, mode,
+ (ConnectorChangelogRowChangeSpec.Delete) spec, child,
context);
+ }
+ if (spec instanceof ConnectorChangelogRowChangeSpec.Merge) {
+ return new MergeBuilder(schema, targetTable, primaryKeys, mode,
+ (ConnectorChangelogRowChangeSpec.Merge) spec, child,
context).build();
+ }
+ throw new AnalysisException("Unsupported connector changelog
specification: "
+ + spec.getClass().getSimpleName());
+ }
+
+ private static LogicalPlan buildUpdate(List<Column> schema, ExternalTable
targetTable,
+ ConnectorChangelogMode mode,
+ ConnectorChangelogRowChangeSpec.Update update, LogicalPlan child,
+ CascadesContext context) {
+ Map<String, Expression> changes =
Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ for (EqualTo assignment : update.getAssignments()) {
+ List<String> parts = ((UnboundSlot)
assignment.left()).getNameParts();
+ String name = parts.get(parts.size() - 1);
+ if (changes.put(name, assignment.right()) != null) {
+ throw new AnalysisException("Duplicate column name in
connector UPDATE: " + name);
+ }
+ }
+ ExpressionAnalyzer analyzer = analyzer(child, context);
+ List<NamedExpression> projects = new ArrayList<>();
+ projects.add(operation(mode.getOperationColumnName(),
mode.getUpdateValue()));
Review Comment:
Addressed in cc0bb9923d5. The generic changelog planner now rejects
primary-key assignments for both UPDATE and matched MERGE UPDATE, with
case-insensitive tests for both paths. The connector receives only an
after-image UPDATE, so allowing a key change would leave the old keyed row
behind.
--
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]
