github-actions[bot] commented on code in PR #66307: URL: https://github.com/apache/doris/pull/66307#discussion_r3782943266
########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ConstraintCommandUtils.java: ########## @@ -0,0 +1,355 @@ +// 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.trees.plans.commands; + +import org.apache.doris.catalog.DatabaseIf; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.catalog.info.TableNameInfo; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.util.MetaLockUtils; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.datasource.ExternalCatalog; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.info.TableNameInfoUtils; +import org.apache.doris.qe.ConnectContext; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashSet; +import java.util.IdentityHashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Shared locking helpers for constraint DDL commands. */ +final class ConstraintCommandUtils { + private ConstraintCommandUtils() { + } + + static ExternalCatalogSnapshots snapshotExternalCatalogs(List<TableNameInfo> tableNameInfos) + throws DdlException { + Map<Long, ExternalCatalogSnapshot> snapshots = new LinkedHashMap<>(); + for (TableNameInfo tableNameInfo : tableNameInfos) { + CatalogIf<?> catalog = Env.getCurrentEnv().getCatalogMgr() + .getCatalogOrDdlException(tableNameInfo.getCtl()); + if (catalog instanceof ExternalCatalog) { + ExternalCatalog externalCatalog = (ExternalCatalog) catalog; + snapshots.putIfAbsent(externalCatalog.getId(), + new ExternalCatalogSnapshot(tableNameInfo.getCtl(), externalCatalog, + externalCatalog.snapshotConstraintMetadata())); + } + } + return new ExternalCatalogSnapshots(snapshots); + } + + /** Lock external catalog fences and internal databases referenced by a constraint. */ + static LockedDatabases lockCurrentDatabases(List<TableNameInfo> tableNameInfos, + ExternalCatalogSnapshots externalCatalogSnapshots, List<TableIf> analyzedTables) + throws DdlException { + Map<String, TableIf> analyzedExternalTables = new LinkedHashMap<>(); + for (TableIf table : analyzedTables) { + if (table != null + && table.getDatabase().getCatalog() instanceof ExternalCatalog) { + TableNameInfo tableNameInfo = TableNameInfoUtils.fromCatalogDb( + table.getDatabase().getCatalog(), table.getDatabase(), table); + analyzedExternalTables.put(tableKey(tableNameInfo), table); + } + } + LockedExternalCatalogs lockedExternalCatalogs = externalCatalogSnapshots.lock(); + Map<String, ResolvedDatabase> resolvedByName = new LinkedHashMap<>(); + LockedDatabases lockedDatabases = null; + try { + for (TableNameInfo tableNameInfo : tableNameInfos) { + String databaseKey = databaseKey(tableNameInfo); + if (!resolvedByName.containsKey(databaseKey)) { + CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog = Env.getCurrentEnv() + .getCatalogMgr().getCatalogOrDdlException(tableNameInfo.getCtl()); + if (catalog instanceof ExternalCatalog) { + externalCatalogSnapshots.requireSame(tableNameInfo.getCtl(), catalog); + continue; + } + DatabaseIf<? extends TableIf> database = + catalog.getDbOrDdlException(tableNameInfo.getDb()); + resolvedByName.put(databaseKey, + new ResolvedDatabase(databaseKey, tableNameInfo, catalog, database)); + } + } + Map<String, TableIf> resolvedTables = new LinkedHashMap<>(analyzedExternalTables); + for (TableNameInfo tableNameInfo : tableNameInfos) { + ResolvedDatabase resolvedDatabase = resolvedByName.get(databaseKey(tableNameInfo)); + if (resolvedDatabase != null) { + resolvedTables.put(tableKey(tableNameInfo), + resolvedDatabase.database.getTableNullable(tableNameInfo.getTbl())); + } + } + List<ResolvedDatabase> lockOrder = new ArrayList<>(resolvedByName.values()); + lockOrder.sort(Comparator + .comparingLong((ResolvedDatabase resolved) -> resolved.database.getId()) + .thenComparing(resolved -> resolved.databaseKey)); + for (ResolvedDatabase resolved : lockOrder) { + resolved.database.readLock(); + } + lockedDatabases = new LockedDatabases( + resolvedByName, resolvedTables, lockOrder, lockedExternalCatalogs); + for (ResolvedDatabase resolved : lockOrder) { + if (Env.getCurrentEnv().getCatalogMgr().getCatalog( + resolved.tableNameInfo.getCtl()) != resolved.catalog + || resolved.catalog.getDbNullable(resolved.tableNameInfo.getDb()) + != resolved.database) { + throw new DdlException( + "Database changed while altering constraint on " + + resolved.tableNameInfo); + } + } + return lockedDatabases; + } catch (DdlException | RuntimeException e) { + if (lockedDatabases == null) { + lockedExternalCatalogs.close(); + } else { + lockedDatabases.close(); + } + throw e; + } + } + + /** Lock all currently resolved tables in the same deterministic order used by constraint ADD and DROP. */ + static LockedTables lockCurrentTables( + LockedDatabases lockedDatabases, List<TableNameInfo> tableNameInfos) + throws DdlException { + return lockCurrentTables(lockedDatabases, tableNameInfos, true); + } + + private static LockedTables lockCurrentTables( + LockedDatabases lockedDatabases, List<TableNameInfo> tableNameInfos, + boolean requireAllTables) throws DdlException { + Map<String, TableIf> tablesByName = new LinkedHashMap<>(); + Map<TableIf, Boolean> seenTables = new IdentityHashMap<>(); + List<TableIf> lockOrder = new ArrayList<>(); + for (TableNameInfo tableNameInfo : tableNameInfos) { + TableIf table = lockedDatabases.getCurrentTable(tableNameInfo); + if (table == null && requireAllTables) { + throw new DdlException("Table changed while altering constraint on " + tableNameInfo); + } + tablesByName.put(tableKey(tableNameInfo), table); + if (table != null + && !(table.getDatabase() instanceof ExternalDatabase) + && seenTables.put(table, Boolean.TRUE) == null) { + lockOrder.add(table); + } + } + lockOrder.sort(Comparator Review Comment: [P1] Keep table locks in the global table-ID order This comparator orders by database ID before table ID, but existing multi-table readers such as `MTMVTask` sort all base tables globally by `TableIf.getId()` before calling `MetaLockUtils.readLockTables` specifically to avoid deadlock. With table A(ID 100) in the lower-ID database and B(ID 50) in the higher-ID database, a cross-database FK ADD/DROP takes A then B while an MTMV refresh takes B then A; each can hold one lock while waiting forever for the other. Please use the repository-wide global table-ID order here and add a test whose database-ID order is the reverse of its table-ID order. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java: ########## @@ -893,6 +1118,86 @@ private void validateTableAndColumns(TableNameInfo tableNameInfo, fk.getReferencedColumnNames(), toKey(refTableInfo)); } + } else if (constraint instanceof DistributionMappingConstraint) { + validateDistributionMappingConstraint( + tableNameInfo, table, (DistributionMappingConstraint) constraint); + } + return table; + } + + private void validateResolvedConstraint(TableNameInfo tableNameInfo, TableIf table, Review Comment: [P1] Revalidate ordinary constraint columns under the acquired locks The resolved-table branch bypasses `validateTableAndColumns`, and this method rechecks columns only for a distribution mapping. Analysis happens before `lockCurrentTables`: a concurrent RENAME/DROP COLUMN can complete on the same table object in that gap, so `requireSame` still passes and ADD journals a PK/UK/FK naming a missing column (the referenced FK side has the same race). Nereids later turns those names into uniqueness/FK proofs, which can fail planning or become false after name reuse. Please revalidate the local and referenced columns against the locked objects, with a latch-based race test. This is distinct from the prior mapping thread because mappings are revalidated here while PK/UK/FK newly are not. ########## fe/fe-core/src/main/java/org/apache/doris/backup/RestoreJob.java: ########## @@ -2158,7 +2165,27 @@ protected Status allTabletCommitted(boolean isReplay) { if (db == null) { return new Status(ErrCode.NOT_FOUND, "database " + dbId + " does not exist"); } + com.google.common.collect.Table<Long, Long, SnapshotInfo> savedSnapshotInfos = snapshotInfos; + Status status; + if (!isAtomicRestore) { + status = finishAllTabletsCommitted(db, isReplay); + } else { + if (!db.writeLockIfExist()) { + return Status.OK; + } + try { + status = finishAllTabletsCommitted(db, isReplay); Review Comment: [P2] Await the restore journal after releasing the database lock This new outer write lock encloses all of `finishAllTabletsCommitted()`; on the leader that method synchronously waits in `logRestoreJob(this)`, and `clean_tables` can reach additional journaling drop paths inside the same scope. A slow/full journal now blocks every metadata reader and writer of the restored database, contrary to the FE lock rule, whereas this terminal journal was outside the former per-replacement lock scopes. Please prepare/enqueue the ordered terminal transition while locked, release the database lock before awaiting durability, and add blocked-journal coverage. This is a separate RestoreJob path from the constraint ADD/DROP journal issue. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java: ########## @@ -67,38 +74,103 @@ public DropConstraintCommand(String name, LogicalPlan plan) { @Override public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { TableNameInfo tableNameInfo; - try { - TableIf table = extractTable(ctx, plan); - tableNameInfo = TableNameInfoUtils.fromCatalogDb( - table.getDatabase().getCatalog(), table.getDatabase(), table); - } catch (Exception e) { - // Table may no longer exist (e.g., external table deleted by another system). - // Fall back to extracting the table name from the unresolved plan. - LOG.warn("Table resolution failed for dropping constraint {}, " - + "falling back to name-based lookup: {}", name, e.getMessage()); - tableNameInfo = extractTableNameFromPlan(ctx); + TableNameInfo unresolvedTableName = plan instanceof UnboundRelation + ? extractTableNameFromPlan(ctx) : null; + CatalogIf<?> unresolvedCatalog = unresolvedTableName == null ? null + : Env.getCurrentEnv().getCatalogMgr().getCatalog(unresolvedTableName.getCtl()); + if (unresolvedCatalog instanceof ExternalCatalog) { + // External PK/FK/UK constraints are authoritative in ConstraintManager. Avoid connector + // schema loading so DROP works with cache disabled or session cache bypass. + tableNameInfo = unresolvedTableName; Review Comment: [P2] Canonicalize the external name before manager lookup ADD stores its key from the resolved catalog/database/table objects, but this fast path feeds the raw SQL spelling to `ConstraintManager.toKey`, which is case-sensitive. With an external catalog using case-insensitive, case-preserving names (for example `lower_case_table_names=2`), `ADD` on `table_a` can resolve and store canonical `Table_A`, while the same spelling on DROP looks up `table_a` and reports an unknown constraint. Please obtain the canonical local database/table identity without schema load-through (or normalize the external manager lookup), and add a cache-disabled mixed-case DROP test. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java: ########## @@ -234,10 +305,115 @@ public ImmutableList<UniqueConstraint> getUniqueConstraints( UniqueConstraint.class); } + /** Returns all distribution mapping constraints for the given table. */ + public ImmutableList<DistributionMappingConstraint> getDistributionMappingConstraints( + TableNameInfo tableNameInfo) { + return getConstraintsByType(toKey(tableNameInfo), DistributionMappingConstraint.class); + } + /** - * Remove all constraints for a table and clean up bidirectional references. - * Called when a table is dropped. + * Returns mappings owned by the concrete table object. + * + * <p>The table-local copy follows the table through recycle, backup, restore, and rename lifecycles. + * Optimizer code must use this overload so a stale qualified-name entry can never bind to another table.</p> */ + @SuppressWarnings("deprecation") + public ImmutableList<DistributionMappingConstraint> getDistributionMappingConstraints(TableIf table) { + if (!(table instanceof Table)) { + return ImmutableList.of(); + } + readLock(); + try { + return ((Table) table).getTableAttributes().getConstraintsMap().values().stream() + .filter(DistributionMappingConstraint.class::isInstance) + .map(DistributionMappingConstraint.class::cast) + .collect(ImmutableList.toImmutableList()); + } finally { + readUnlock(); + } + } + + /** Return the table-owned mapping that uses the given column, if any. */ + @SuppressWarnings("deprecation") + public String findDistributionMappingConstraintWithColumn(TableIf table, String columnName) { + if (!(table instanceof Table)) { + return null; + } + readLock(); + try { + return ((Table) table).getTableAttributes().getConstraintsMap().entrySet().stream() + .filter(entry -> entry.getValue() instanceof DistributionMappingConstraint) + .filter(entry -> { + DistributionMappingConstraint mapping = + (DistributionMappingConstraint) entry.getValue(); + return containsIgnoreCase(mapping.getDeterminantColumnNames(), columnName) + || containsIgnoreCase(mapping.getDistributionColumnNames(), columnName); + }) + .map(Entry::getKey) + .findFirst() + .orElse(null); + } finally { + readUnlock(); + } + } + + /** Rebuild the qualified-name index from constraints persisted with a recovered or restored table. */ + @SuppressWarnings("deprecation") + public void restoreTableConstraints(TableNameInfo tableNameInfo, TableIf table) { + if (!(table instanceof Table)) { + return; + } + Map<String, Constraint> tableLocalConstraints = + ((Table) table).getTableAttributes().getConstraintsMap(); + String key = toKey(tableNameInfo); + writeLock(); + try { + Map<String, Constraint> indexedConstraints = constraintsMap.get(key); + if (indexedConstraints != null) { + indexedConstraints.entrySet().removeIf( + entry -> entry.getValue() instanceof DistributionMappingConstraint); + } + for (Entry<String, Constraint> entry : tableLocalConstraints.entrySet()) { + Constraint constraint = entry.getValue(); + if (constraint instanceof DistributionMappingConstraint) { + if (indexedConstraints == null) { + indexedConstraints = new HashMap<>(); + } + indexedConstraints.put(entry.getKey(), constraint); + } + } + if (indexedConstraints == null || indexedConstraints.isEmpty()) { + constraintsMap.remove(key); + } else { + constraintsMap.put(key, indexedConstraints); + } + } finally { + writeUnlock(); + } + } + + /** Populate table-local mapping metadata after loading an image created before table-local ownership. */ + public void syncDistributionMappingsToTables() { + Map<String, Map<String, Constraint>> snapshot; + readLock(); + try { + snapshot = constraintsMap.entrySet().stream() + .collect(Collectors.toMap( + Entry::getKey, + entry -> ImmutableMap.copyOf(entry.getValue()))); + } finally { + readUnlock(); + } + snapshot.forEach((tableKey, constraints) -> { + TableIf table = resolveTableIfPresent(new TableNameInfo(tableKey)); Review Comment: [P1] Skip non-mapping entries before startup table resolution `migrateConstraintsFromTables()` calls this repair whenever the loaded manager is nonempty, but the table is resolved before checking whether the entry contains any `DistributionMappingConstraint`. A persisted external PK/FK/UK therefore makes every FE image load initialize that connector and synchronously list/load remote metadata, potentially stalling startup when the remote system is unavailable, even though mappings are internal `OlapTable` metadata and there is nothing to copy for this entry. Please filter to mapping constraints first (or restrict this pass to the internal catalog), and cover an external PK-only image with a lookup that must not run. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java: ########## @@ -113,13 +139,60 @@ private void checkAlterPriv(ConnectContext ctx, TableNameInfo tableNameInfo) } } - private void addConstraintAndInvalidate( - TableNameInfo tableNameInfo, org.apache.doris.catalog.constraint.Constraint constraint) + private void addConstraintWithLocks(TableNameInfo tableNameInfo, + List<TableNameInfo> affectedTableInfos, + org.apache.doris.catalog.constraint.Constraint constraint, + TableIf analyzedTable, TableIf analyzedReferencedTable, + ConstraintCommandUtils.ExternalCatalogSnapshots externalCatalogSnapshots) throws Exception { - List<MTMV> dependentMtmvs = MTMVUtil.getDependentMtmvsByConstraint(tableNameInfo, constraint); - Env.getCurrentEnv().getConstraintManager().addConstraint(tableNameInfo, name, constraint, false); + List<TableIf> analyzedTables = new ArrayList<>(); + analyzedTables.add(analyzedTable); + if (analyzedReferencedTable != null) { + analyzedTables.add(analyzedReferencedTable); + } + List<MTMV> dependentMtmvs; + try (ConstraintCommandUtils.LockedDatabases lockedDatabases = + ConstraintCommandUtils.lockCurrentDatabases( + affectedTableInfos, externalCatalogSnapshots, analyzedTables); + ConstraintCommandUtils.LockedTables lockedTables = + ConstraintCommandUtils.lockCurrentTables( + lockedDatabases, affectedTableInfos)) { + lockedTables.requireSame(tableNameInfo, analyzedTable); + TableIf currentTable = lockedTables.get(tableNameInfo); + if (constraint instanceof DistributionMappingConstraint) { + Preconditions.checkState(currentTable instanceof OlapTable, + "distribution mapping constraint requires an OLAP table"); + ((OlapTable) currentTable).checkNormalStateForAlter(); + } + TableIf referencedTable = null; + if (constraint instanceof ForeignKeyConstraint) { + TableNameInfo referencedTableInfo = + ((ForeignKeyConstraint) constraint).getReferencedTableName(); + Preconditions.checkNotNull(referencedTableInfo); + referencedTable = lockedTables.get(referencedTableInfo); + lockedTables.requireSame(referencedTableInfo, analyzedReferencedTable); + } + dependentMtmvs = MTMVUtil.getDependentMtmvsByConstraint(tableNameInfo, constraint); + Env.getCurrentEnv().getConstraintManager() + .addConstraintWithResolvedTables( Review Comment: [P2] Keep the journal wait outside metadata locks This call is inside both `LockedDatabases` and `LockedTables`, and the manager synchronously waits in `logAddConstraint` before returning; DROP has the same shape. A slow/full journal therefore holds internal table write locks and database read locks for the whole flush, and a cross-database FK stalls both databases, contrary to the FE metadata-lock rule. Please use a prepared/enqueued transition that preserves journal order but performs the unbounded wait after releasing these locks (rather than simply moving an unordered log call), with blocked-journal coverage. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java: ########## @@ -505,6 +515,50 @@ public Void visitPhysicalHashAggregate(PhysicalHashAggregate<? extends Plan> agg return null; } + private void addColocateMappingRequestForAggregate(PhysicalHashAggregate<? extends Plan> agg) { + DistributionSpec parentDistribution = requestPropertyFromParent.getDistributionSpec(); + if (connectContext == null + || !connectContext.getSessionVariable().isEnableColocateMappingConstraint() + || agg.hasSourceRepeat() + || !(parentDistribution instanceof DistributionSpecHash) + || ((DistributionSpecHash) parentDistribution).getShuffleType() + != ShuffleType.COLOCATE_MAPPING_REQUIRE) { + return; + } + + Map<ExprId, NamedExpression> outputByExprId = agg.getOutputExpressions().stream() + .collect(Collectors.toMap(NamedExpression::getExprId, output -> output, (left, right) -> left)); + Set<ExprId> groupByExprIds = Sets.newHashSet(); + for (Expression groupBy : agg.getGroupByExpressions()) { + if (!(groupBy instanceof SlotReference)) { + return; + } + groupByExprIds.add(((SlotReference) groupBy).getExprId()); + } + List<ExprId> childRequiredExprIds = Lists.newArrayList(); + for (ExprId requiredExprId + : ((DistributionSpecHash) parentDistribution).getOrderedShuffledColumns()) { + NamedExpression output = outputByExprId.get(requiredExprId); + ExprId childExprId; + if (output instanceof Alias && ((Alias) output).child() instanceof SlotReference) { Review Comment: [P2] Remap mapping requests through hash-preserving casts This new request path accepts `Alias(SlotReference)` only, while `computeProjectOutputProperties` explicitly preserves distribution and the hidden mapping proof through `Alias(Cast(Slot))` when `isSameHashValue` is true (for example widening VARCHAR). In an aggregate-to-mapping-join tree whose grouped determinant is exposed through that cast, the join request reaches this `continue`, cannot cover the mapped bucket position, and the ordinary aggregate alternative inserts an avoidable Exchange even though output derivation would retain the proof. Please reuse the same hash-preserving-cast predicate for request remapping and add a widening-VARCHAR determinant regression. -- 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]
