github-actions[bot] commented on code in PR #66498:
URL: https://github.com/apache/doris/pull/66498#discussion_r3756061932
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromUsingCommand.java:
##########
@@ -55,6 +58,17 @@ public void run(ConnectContext ctx, StmtExecutor executor)
throws Exception {
+ " Please check the following session variables: "
+ ctx.getSessionVariable().printDebugModeVariables());
}
+ TableIf table =
RelationUtil.getTable(RelationUtil.getQualifierName(ctx, nameParts),
+ ctx.getEnv(), Optional.empty());
+ if (table instanceof PaimonExternalTable) {
+ if (isTempPart || !partitions.isEmpty()) {
+ throw new AnalysisException(
+ "Paimon DELETE does not support partition name lists;
use a WHERE predicate");
+ }
+ new PaimonDeleteCommand(nameParts, tableAlias,
handleCte(logicalQuery))
Review Comment:
[P1] Deduplicate DELETE USING rows before writing an input changelog. This
branch sends the raw target/source join to Paimon, so if two USING rows match
the same target key, buildDelete emits two DELETE records.
PaimonRowChangeCapabilities still allows changelog-producer=input, and that
producer persists every input record; an incremental consumer therefore sees
two retractions for the one row this SQL statement deleted and can undercount
aggregates even though the current table looks correct. The existing
input-producer UPDATE fence does not cover this distinct multiplicity path.
Collapse the joined input to one row per target primary key (or reject this
shape for input-producer tables), and add a duplicate-USING incremental-read
regression.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/PaimonRowChangeCapabilities.java:
##########
@@ -0,0 +1,211 @@
+// 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.analysis.UserIdentity;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.paimon.PaimonExternalTable;
+import org.apache.doris.datasource.paimon.PaimonWriteTarget;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.plans.commands.info.PaimonRowChangeSpec;
+import org.apache.doris.nereids.trees.plans.commands.merge.MergeMatchedClause;
+import org.apache.doris.qe.ConnectContext;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+/** Validates row-change operations against the Paimon table capabilities. */
+final class PaimonRowChangeCapabilities {
+ private PaimonRowChangeCapabilities() {
+ }
+
+ static void check(PaimonWriteTarget target, PaimonRowChangeSpec spec,
+ CascadesContext cascadesContext) {
+ requireNoDataMask(target, spec, cascadesContext);
+ if (spec instanceof PaimonRowChangeSpec.Update) {
+ checkUpdate(target,
+ updatedColumns(((PaimonRowChangeSpec.Update)
spec).getAssignments()));
+ } else if (spec instanceof PaimonRowChangeSpec.Delete) {
+ checkDelete(target);
+ } else if (spec instanceof PaimonRowChangeSpec.Merge) {
+ checkMerge(target, (PaimonRowChangeSpec.Merge) spec);
+ } else {
+ throw new AnalysisException("Unsupported Paimon row-change
specification: "
+ + spec.getClass().getSimpleName());
+ }
+ }
+
+ private static void checkMerge(PaimonWriteTarget target,
PaimonRowChangeSpec.Merge merge) {
+ Set<String> updatedColumns = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ boolean containsUpdate = false;
+ boolean containsDelete = false;
+ for (MergeMatchedClause clause : merge.getMatchedClauses()) {
+ containsDelete |= clause.isDelete();
+ containsUpdate |= !clause.isDelete();
+ updatedColumns.addAll(updatedColumns(clause.getAssignments()));
+ }
+ checkMergeCapabilities(target, updatedColumns, containsUpdate,
containsDelete);
+ }
+
+ private static Set<String> updatedColumns(List<EqualTo> assignments) {
+ Set<String> columns = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ for (EqualTo assignment : assignments) {
+ List<String> parts = ((UnboundSlot)
assignment.left()).getNameParts();
+ columns.add(parts.get(parts.size() - 1));
+ }
+ return columns;
+ }
+
+ private static void checkUpdate(PaimonWriteTarget target,
Collection<String> updatedColumns) {
+ FileStoreTable table = target.getTable();
+ requirePrimaryKey(table, "UPDATE");
+ CoreOptions options = CoreOptions.fromMap(table.options());
Review Comment:
[P1] Fence row-change DML when the key-dynamic index is truncated. A table
can set cross-partition-upsert.index-ttl while bucket=-1; Paimon then applies
that TTL during index initialization, so the target scan can find an old row
that this sink's IndexBootstrap omits. GlobalIndexAssigner treats the missing
key as new, assigns a fresh bucket/partition, and never addresses the scanned
row's old location. DELETE can therefore leave the row intact, while
UPDATE/MERGE can create a duplicate, with no concurrent writer involved. Reject
row-change DML when this TTL is configured, or carry the scanned row's physical
location/use a complete bootstrap, and add an expired-key final-state
regression.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/PaimonRowChangeCapabilities.java:
##########
@@ -0,0 +1,211 @@
+// 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.analysis.UserIdentity;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.paimon.PaimonExternalTable;
+import org.apache.doris.datasource.paimon.PaimonWriteTarget;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.plans.commands.info.PaimonRowChangeSpec;
+import org.apache.doris.nereids.trees.plans.commands.merge.MergeMatchedClause;
+import org.apache.doris.qe.ConnectContext;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+/** Validates row-change operations against the Paimon table capabilities. */
+final class PaimonRowChangeCapabilities {
+ private PaimonRowChangeCapabilities() {
+ }
+
+ static void check(PaimonWriteTarget target, PaimonRowChangeSpec spec,
+ CascadesContext cascadesContext) {
+ requireNoDataMask(target, spec, cascadesContext);
+ if (spec instanceof PaimonRowChangeSpec.Update) {
+ checkUpdate(target,
+ updatedColumns(((PaimonRowChangeSpec.Update)
spec).getAssignments()));
+ } else if (spec instanceof PaimonRowChangeSpec.Delete) {
+ checkDelete(target);
+ } else if (spec instanceof PaimonRowChangeSpec.Merge) {
+ checkMerge(target, (PaimonRowChangeSpec.Merge) spec);
+ } else {
+ throw new AnalysisException("Unsupported Paimon row-change
specification: "
+ + spec.getClass().getSimpleName());
+ }
+ }
+
+ private static void checkMerge(PaimonWriteTarget target,
PaimonRowChangeSpec.Merge merge) {
+ Set<String> updatedColumns = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ boolean containsUpdate = false;
+ boolean containsDelete = false;
+ for (MergeMatchedClause clause : merge.getMatchedClauses()) {
+ containsDelete |= clause.isDelete();
+ containsUpdate |= !clause.isDelete();
+ updatedColumns.addAll(updatedColumns(clause.getAssignments()));
+ }
+ checkMergeCapabilities(target, updatedColumns, containsUpdate,
containsDelete);
+ }
+
+ private static Set<String> updatedColumns(List<EqualTo> assignments) {
+ Set<String> columns = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ for (EqualTo assignment : assignments) {
+ List<String> parts = ((UnboundSlot)
assignment.left()).getNameParts();
+ columns.add(parts.get(parts.size() - 1));
+ }
+ return columns;
+ }
+
+ private static void checkUpdate(PaimonWriteTarget target,
Collection<String> updatedColumns) {
+ FileStoreTable table = target.getTable();
+ requirePrimaryKey(table, "UPDATE");
+ CoreOptions options = CoreOptions.fromMap(table.options());
+ requireNoRowKindField(options, "UPDATE");
+ if (options.changelogProducer() ==
CoreOptions.ChangelogProducer.INPUT) {
+ throw new AnalysisException("Paimon UPDATE is not supported when "
+ + "changelog-producer=input because both UPDATE_BEFORE and
UPDATE_AFTER "
+ + "records are required");
+ }
+ Set<String> primaryKeys = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ primaryKeys.addAll(table.primaryKeys());
+ Set<String> partitionKeys = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ partitionKeys.addAll(table.partitionKeys());
+ Set<String> sequenceFields = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ sequenceFields.addAll(options.sequenceField());
+ for (String column : updatedColumns) {
+ if (primaryKeys.contains(column)) {
+ throw new AnalysisException("Paimon UPDATE cannot modify
primary-key column '"
+ + column + "'");
+ }
+ if (sequenceFields.contains(column)) {
+ throw new AnalysisException("Paimon UPDATE cannot modify
sequence-field column '"
+ + column + "'");
+ }
+ if (partitionKeys.contains(column) && options.bucket() != -1) {
Review Comment:
[P1] Also reject key-dynamic partition moves when ignore-delete=true. Moving
a key between partitions in bucket=-1 mode relies on the global-index path
emitting a DELETE to the old partition before writing the replacement. This
option deliberately discards that retract, but UPDATE-only and MERGE-UPDATE
statements never call checkDelete, so this loop still permits the move and the
same logical key remains visible in both partitions. Fence partition-column
assignments on ignore-delete (or use a removal that the option cannot suppress)
and cover the combined option/move case.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/PaimonRowChangePlanBuilder.java:
##########
@@ -0,0 +1,137 @@
+// 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.common.util.Util;
+import org.apache.doris.datasource.paimon.PaimonRowChangeOperation;
+import org.apache.doris.datasource.paimon.PaimonWriteTarget;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.analyzer.Scope;
+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.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
+import org.apache.doris.nereids.trees.plans.commands.info.PaimonRowChangeSpec;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.util.TypeCoercionUtils;
+
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** Builds a Paimon changelog projection against the current write target. */
+final class PaimonRowChangePlanBuilder {
+ private PaimonRowChangePlanBuilder() {
+ }
+
+ static LogicalProject<?> build(
+ PaimonWriteTarget target, PaimonRowChangeSpec spec, LogicalPlan
child,
+ CascadesContext cascadesContext) {
+ PaimonRowChangeCapabilities.check(target, spec, cascadesContext);
+ if (spec instanceof PaimonRowChangeSpec.Update) {
+ return buildUpdate(target, (PaimonRowChangeSpec.Update) spec,
+ child, cascadesContext);
+ }
+ if (spec instanceof PaimonRowChangeSpec.Delete) {
+ return buildDelete(target, (PaimonRowChangeSpec.Delete) spec,
+ child, cascadesContext);
+ }
+ if (spec instanceof PaimonRowChangeSpec.Merge) {
+ return PaimonMergePlanner.build(target,
(PaimonRowChangeSpec.Merge) spec,
+ child, cascadesContext);
+ }
+ throw new AnalysisException("Unsupported Paimon row-change
specification: "
+ + spec.getClass().getSimpleName());
+ }
+
+ private static LogicalProject<?> buildUpdate(PaimonWriteTarget target,
+ PaimonRowChangeSpec.Update update, LogicalPlan child,
+ CascadesContext cascadesContext) {
+ Map<String, Expression> changes =
Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ for (EqualTo assignment : update.getAssignments()) {
+ List<String> parts = ((UnboundSlot)
assignment.left()).getNameParts();
+ String columnName = parts.get(parts.size() - 1);
+ if (changes.put(columnName, assignment.right()) != null) {
+ throw new AnalysisException(
+ "Duplicate column name in Paimon UPDATE: " +
columnName);
+ }
+ }
+
+ String targetName = update.getTableAlias() != null
+ ? update.getTableAlias()
+ :
Util.getTempTableDisplayName(target.getDorisTable().getName());
Review Comment:
[P2] Preserve the resolved target qualifier when synthesizing row images.
Without an alias this reduces the target to table.column; bindSingleSlotByTable
then matches only the last qualifier component, so UPDATE FROM or DELETE USING
against another catalog/database's same-named table fails as ambiguous even
when the user's predicates are fully qualified. MERGE's suffix lookup has the
same problem when its target was written unqualified. Carry the resolved target
slot identities/full qualifier through the row-change spec instead of
rebuilding a short name, and add same-basename joined-table regressions.
--
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]
