github-actions[bot] commented on code in PR #66498: URL: https://github.com/apache/doris/pull/66498#discussion_r3744279464
########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/PaimonRowChangeCapabilities.java: ########## @@ -0,0 +1,160 @@ +// 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.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.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) { + 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"); + Set<String> primaryKeys = new TreeSet<>(String.CASE_INSENSITIVE_ORDER); + primaryKeys.addAll(table.primaryKeys()); + 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 + "'"); + } + } + CoreOptions.MergeEngine engine = options.mergeEngine(); + if (engine != CoreOptions.MergeEngine.DEDUPLICATE) { + throw new AnalysisException("Paimon UPDATE only supports merge-engine=deduplicate; " + + "merge-engine=" + engine + " cannot preserve SQL UPDATE semantics"); + } + } + + private static void checkDelete(PaimonWriteTarget target) { + FileStoreTable table = target.getTable(); + requirePrimaryKey(table, "DELETE"); + requireNoRowKindField(CoreOptions.fromMap(table.options()), "DELETE"); + Options options = Options.fromMap(table.options()); + if (options.get(CoreOptions.IGNORE_DELETE)) { + throw new AnalysisException("Paimon DELETE is not supported when ignore-delete=true " + + "because the delete record would be ignored"); + } + CoreOptions.MergeEngine engine = options.get(CoreOptions.MERGE_ENGINE); + switch (engine) { + case DEDUPLICATE: + return; + case PARTIAL_UPDATE: + if (options.get(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE) + || options.getOptional( Review Comment: [P1] Do not treat the sequence-group option as sufficient for SQL DELETE. In Paimon 1.4.2, `PartialUpdateMergeFunction.retractWithSequenceGroup` skips a group when all of its sequence fields are NULL; for an existing row `meetInsert` remains true and `currentDeleteRow` remains false, so the merged result stays INSERT. This full-row DELETE can therefore report success while leaving the row, and MERGE DELETE shares the same path. `partial-update.remove-record-on-sequence-group` is conditional per group, not an unconditional SQL-delete contract. Require `partial-update.remove-record-on-delete=true` for Doris DELETE/MERGE DELETE, or implement a representation/check that guarantees whole-row removal, and add a NULL sequence-group regression. ########## 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); + 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()); + ExpressionAnalyzer analyzer = expressionAnalyzer(child, cascadesContext); Review Comment: [P1] Preserve unmasked target values when building these changelog rows. At this point `child` has already passed through `LogicalCheckPolicy`; when a data-mask applies, `CheckPolicy` replaces the target output with mask expressions while deliberately retaining the original column names and qualifiers. Resolving against that scope therefore writes masked values back for every unassigned UPDATE column and for DELETE row images, while MERGE also matches and projects keys from the masked aliases. A user with LOAD plus masked SELECT access can silently corrupt the Paimon table (and a masked PK can redirect which key is changed). Please keep the row-policy filter, but form physical row images/keys from the raw target slots or reject row-level DML under data masks, and add UPDATE/DELETE/MERGE mask 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]
