github-actions[bot] commented on code in PR #68320:
URL: https://github.com/apache/doris/pull/68320#discussion_r4119628079
##########
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:
[P2] Only request target-row deduplication for a real USING join. The parser
also selects this command for DELETE with a CTE or ORDER BY/LIMIT and no USING.
The unconditional `true` makes the changelog builder demand a primary key,
although the equivalent plain DELETE on a keyless full-row changelog table
succeeds. Carry whether USING was present, so those forms retain the ordinary
DELETE contract.
##########
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:
[P1] Gate these exchange modes on a capability introduced with this BE code.
The FE accepts `be_exec_version >= 13`, but that version predates this PR (the
default is already 15); an older BE in a rolling upgrade can therefore receive
the new Thrift partition types 10/11 without their exchange handlers. It will
reject the enum or, if decoded, miss every send branch for a multi-destination
exchange. Please use a new capability/version and check all destination BEs
before emitting these types.
##########
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:
[P1] Enforce writer affinity for HASH when the child has only a natural scan
hash property. A stable-colocate OLAP scan over multiple range partitions can
advertise NATURAL(k), which satisfies this REQUIRE(k) and lets the optimizer
omit an exchange. Yet a scan-only sink fragment assigns tablets without bucket
awareness and may split same-k tablets from different partitions across writer
instances. Please require an execution shuffle or prove global writer affinity,
and test repeated k across source partitions.
##########
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:
[P1] Preserve the old key for key-changing updates. `UPDATE t SET id=2 WHERE
id=1` produces only an UPDATE row with id=2 here; the connector never receives
id=1, so it cannot remove the original keyed row. Matched MERGE UPDATE has the
same output shape; its saved original key is dropped after the cardinality
check. Since the default SPI validator permits key assignments, reject them in
both forms or emit a delete and insert that carries both keys.
--
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]
