Gabriel39 commented on code in PR #66498: URL: https://github.com/apache/doris/pull/66498#discussion_r3728905765
########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/PaimonRowChangeCapabilities.java: ########## @@ -0,0 +1,101 @@ +// 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.datasource.paimon.PaimonWriteTarget; +import org.apache.doris.nereids.exceptions.AnalysisException; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.FileStoreTable; + +import java.util.Collection; +import java.util.Set; +import java.util.TreeSet; + +/** Validates row-change operations against the Paimon table capabilities. */ +final class PaimonRowChangeCapabilities { + private PaimonRowChangeCapabilities() { + } + + static void checkUpdate(PaimonWriteTarget target, Collection<String> updatedColumns) { + FileStoreTable table = target.getTable(); + requirePrimaryKey(table, "UPDATE"); + Set<String> primaryKeys = new TreeSet<>(String.CASE_INSENSITIVE_ORDER); + primaryKeys.addAll(table.primaryKeys()); + for (String column : updatedColumns) { Review Comment: [P1] Please also reject assignments to columns configured in `sequence.field`. On a deduplicate table, Paimon 1.3.1 orders versions by the user-defined sequence before applying `DeduplicateMergeFunction`. With the default ascending order, `UPDATE t SET seq = 9, value = "new"` against an existing `seq = 10` row can report success while the old row still wins; the inverse applies to descending order. The same issue affects MERGE UPDATE. Either prevent sequence-column assignments or provide a write path that guarantees SQL UPDATE semantics, and add ascending/descending regression cases. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/PaimonRowChangePlanBuilder.java: ########## @@ -0,0 +1,380 @@ +// 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.analyzer.UnboundAlias; +import org.apache.doris.nereids.analyzer.UnboundSlot; +import org.apache.doris.nereids.analyzer.UnboundStar; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.IsNull; +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.Count; +import org.apache.doris.nereids.trees.expressions.functions.scalar.AssertTrue; +import org.apache.doris.nereids.trees.expressions.functions.scalar.If; +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.info.PaimonRowChangeSpec; +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.logical.LogicalFilter; +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 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 Paimon changelog projections while binding against the current write target. */ +final class PaimonRowChangePlanBuilder { + private static final String BRANCH_LABEL = "__DORIS_PAIMON_MERGE_BRANCH__"; + private static final String MATCH_COUNT = "__DORIS_PAIMON_MERGE_MATCH_COUNT__"; + + private PaimonRowChangePlanBuilder() { + } + + static LogicalProject<?> build( + PaimonWriteTarget target, PaimonRowChangeSpec spec, LogicalPlan child) { + checkCapabilities(target, spec); + LogicalProject<?> project; + if (spec instanceof PaimonRowChangeSpec.Update) { + project = buildUpdate(target, + (PaimonRowChangeSpec.Update) spec, child); + } else if (spec instanceof PaimonRowChangeSpec.Delete) { + project = buildDelete(target, (PaimonRowChangeSpec.Delete) spec, child); + } else if (spec instanceof PaimonRowChangeSpec.Merge) { + project = buildMerge(target, + (PaimonRowChangeSpec.Merge) spec, child); + } else { + throw new AnalysisException("Unsupported Paimon row-change specification: " + + spec.getClass().getSimpleName()); + } + return project; + } + + private static void checkCapabilities(PaimonWriteTarget target, PaimonRowChangeSpec spec) { + if (spec instanceof PaimonRowChangeSpec.Update) { + PaimonRowChangeCapabilities.checkUpdate(target, + updatedColumns(((PaimonRowChangeSpec.Update) spec).getAssignments())); + return; + } + if (spec instanceof PaimonRowChangeSpec.Delete) { + PaimonRowChangeCapabilities.checkDelete(target); + return; + } + if (!(spec instanceof PaimonRowChangeSpec.Merge)) { + throw new AnalysisException("Unsupported Paimon row-change specification: " + + spec.getClass().getSimpleName()); + } + PaimonRowChangeSpec.Merge merge = (PaimonRowChangeSpec.Merge) spec; + 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())); + } + PaimonRowChangeCapabilities.checkMerge( + 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 LogicalProject<?> buildUpdate(PaimonWriteTarget target, + PaimonRowChangeSpec.Update update, LogicalPlan child) { + Map<String, Expression> changes = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER); + for (EqualTo assignment : update.getAssignments()) { + List<String> parts = ((UnboundSlot) assignment.left()).getNameParts(); + String column = parts.get(parts.size() - 1); + if (changes.put(column, assignment.right()) != null) { + throw new AnalysisException("Duplicate column name in Paimon UPDATE: " + column); + } + } + + String targetName = update.getTableAlias() != null + ? update.getTableAlias() + : Util.getTempTableDisplayName(target.getDorisTable().getName()); + List<NamedExpression> projects = new ArrayList<>(); + projects.add(new UnboundAlias(new TinyIntLiteral(PaimonRowChangeOperation.UPDATE), + PaimonRowChangeOperation.OPERATION_COLUMN)); + for (Column column : target.getSchema()) { + Expression value = changes.remove(column.getName()); + if (value == null) { + value = new UnboundSlot(targetName, column.getName()); + } + projects.add(value instanceof NamedExpression + ? (NamedExpression) value : new UnboundAlias(value, column.getName())); + } + if (!changes.isEmpty()) { + throw new AnalysisException( + "Unknown column in Paimon UPDATE: " + String.join(", ", changes.keySet())); + } + return new LogicalProject<>(projects, child); + } + + private static LogicalProject<?> buildDelete(PaimonWriteTarget target, + PaimonRowChangeSpec.Delete delete, LogicalPlan child) { + String targetName = delete.getTableAlias() != null + ? delete.getTableAlias() + : Util.getTempTableDisplayName(target.getDorisTable().getName()); + List<NamedExpression> projects = new ArrayList<>(); + projects.add(new UnboundAlias(new TinyIntLiteral(PaimonRowChangeOperation.DELETE), + PaimonRowChangeOperation.OPERATION_COLUMN)); + for (Column column : target.getSchema()) { + projects.add(new UnboundSlot(targetName, column.getName())); + } + return new LogicalProject<>(projects, child); + } + + private static LogicalProject<?> buildMerge(PaimonWriteTarget target, + PaimonRowChangeSpec.Merge merge, LogicalPlan child) { + LogicalPlan plan = child; + Expression targetPresent = targetPresence(target, merge.getTargetNameInPlan()); + if (!merge.getMatchedClauses().isEmpty()) { Review Comment: [P1] This cardinality check only covers rows that matched an existing target row. If two source rows have the same new primary key, both target PKs are NULL after the left join, both select a NOT MATCHED INSERT branch, and both changelog INSERTs are emitted; an insert-only MERGE does not create this window at all. Paimon deduplicate then chooses a winner based on execution/write order, so the MERGE can succeed with a nondeterministic result. Please enforce uniqueness for the projected target primary key of active NOT MATCHED INSERT branches as well, and add a duplicate-new-key regression test. -- 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]
