github-actions[bot] commented on code in PR #66307: URL: https://github.com/apache/doris/pull/66307#discussion_r3773247407
########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ConstraintCommandUtils.java: ########## @@ -0,0 +1,200 @@ +// 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 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() { + } + + /** Lock all databases referenced by a constraint in a deterministic order. */ + static LockedDatabases lockCurrentDatabases(List<TableNameInfo> tableNameInfos) + throws DdlException { + Map<String, ResolvedDatabase> resolvedByName = new LinkedHashMap<>(); + for (TableNameInfo tableNameInfo : tableNameInfos) { + String databaseKey = databaseKey(tableNameInfo); + if (!resolvedByName.containsKey(databaseKey)) { + CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog = Env.getCurrentEnv() + .getCatalogMgr().getCatalogOrDdlException(tableNameInfo.getCtl()); + DatabaseIf<? extends TableIf> database = + catalog.getDbOrDdlException(tableNameInfo.getDb()); + resolvedByName.put(databaseKey, + new ResolvedDatabase(databaseKey, tableNameInfo, catalog, database)); + } + } + 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 lockedDatabases = new LockedDatabases(resolvedByName, lockOrder); + try { + 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) { + 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.get(tableNameInfo) Review Comment: [P1] Keep connector loading outside the database lock set At this point `lockCurrentDatabases()` has already acquired every affected database read lock. For an `ExternalDatabase`, this polymorphic lookup can run `makeSureInitialized()` and load table names/objects; the subsequent constraint validation can also load schema through connector caches. A cross-database FK can therefore hold both database locks for unbounded remote work, violating the FE metadata-lock rule and blocking writers on every affected database when a connector is slow or unavailable. Please resolve/load external metadata before acquiring this lock set, then revalidate through a cache-generation/identity protocol under lightweight locks (or restrict this topology to the internal catalog), with cold-cache and concurrent-reset coverage. ########## fe/fe-core/src/test/java/org/apache/doris/qe/SqlCacheTest.java: ########## @@ -75,4 +94,183 @@ public void testSqlCache() throws Exception { executeNereidsSql("admin set frontend config ('sql_cache_manage_num'='1')"); Assertions.assertEquals(1, sqlCacheManager.getSqlCaches().asMap().size()); } + + @Test + public void testInvalidateSqlCacheByPersistedTableName() throws Exception { + TableIf table = Env.getCurrentInternalCatalog() + .getDbOrDdlException("sql_cache_constraint_test").getTableOrDdlException("t"); + SqlCacheContext cacheContext = new SqlCacheContext(new UserIdentity("admin", "127.0.0.1")); + cacheContext.addUsedTable(table); + NereidsSqlCacheManager sqlCacheManager = Env.getCurrentEnv().getSqlCacheManager(); + sqlCacheManager.getSqlCaches().put("mapping_constraint_cache", cacheContext); + long initialSequence = sqlCacheManager.getTableInvalidationSequence( + TableNameInfoUtils.fromTableOrNull(table)); + + sqlCacheManager.invalidateAboutTableAndFencePublication(TableNameInfoUtils.fromTableOrNull(table)); + + Assertions.assertNull(sqlCacheManager.getSqlCaches().getIfPresent("mapping_constraint_cache")); + Assertions.assertTrue(sqlCacheManager.getTableInvalidationSequence( + TableNameInfoUtils.fromTableOrNull(table)) > initialSequence); + } + + @Test + public void testReplayInvalidationRevokesLookupValue() throws Exception { + connectContext.getSessionVariable().setEnableSqlCache(true); + NereidsSqlCacheManager sqlCacheManager = Env.getCurrentEnv().getSqlCacheManager(); + sqlCacheManager.invalidateAll(); + String sql = "select 300"; + prepareFeCacheContext(sql); + sqlCacheManager.tryAddFeSqlCache(connectContext, sql); + Assertions.assertEquals(1, sqlCacheManager.getSqlCaches().asMap().size()); + + try (MockedStatic<StmtExecutor> stmtExecutor = Mockito.mockStatic(StmtExecutor.class)) { + stmtExecutor.when(() -> StmtExecutor.syncJournalIfNeeded(connectContext)) + .thenAnswer(invocation -> { + sqlCacheManager.invalidateAll(); + return null; + }); + Assertions.assertFalse(sqlCacheManager.tryParseSql(connectContext, sql).isPresent()); + } + } + + @Test + public void testTableInvalidationDoesNotRejectUnrelatedCachePublication() throws Exception { + connectContext.getSessionVariable().setEnableSqlCache(true); + NereidsSqlCacheManager sqlCacheManager = Env.getCurrentEnv().getSqlCacheManager(); + sqlCacheManager.invalidateAll(); + String sql = "select 350"; + prepareFeCacheContext(sql); + long initialSequence = sqlCacheManager.getPublicationSequence(); + TableIf table = Env.getCurrentInternalCatalog() + .getDbOrDdlException("sql_cache_constraint_test").getTableOrDdlException("t"); + + sqlCacheManager.invalidateAboutTable(table); + sqlCacheManager.tryAddFeSqlCache(connectContext, sql); + + Assertions.assertEquals(initialSequence, sqlCacheManager.getPublicationSequence()); + Assertions.assertEquals(1, sqlCacheManager.getSqlCaches().asMap().size()); + } + + @Test + public void testTableInvalidationRejectsLatePublicationForUsedTable() throws Exception { + connectContext.getSessionVariable().setEnableSqlCache(true); + NereidsSqlCacheManager sqlCacheManager = Env.getCurrentEnv().getSqlCacheManager(); + sqlCacheManager.invalidateAll(); + String sql = "select 400"; + TableIf table = Env.getCurrentInternalCatalog() + .getDbOrDdlException("sql_cache_constraint_test").getTableOrDdlException("t"); + prepareFeCacheContext(sql, table); + + CountDownLatch publicationReady = new CountDownLatch(1); + CountDownLatch invalidationFinished = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future<?> publication = executor.submit(() -> { + publicationReady.countDown(); + Assertions.assertTrue(invalidationFinished.await(30, TimeUnit.SECONDS)); + sqlCacheManager.tryAddFeSqlCache(connectContext, sql); + return null; + }); + Future<?> invalidation = executor.submit(() -> { + Assertions.assertTrue(publicationReady.await(30, TimeUnit.SECONDS)); + try { + sqlCacheManager.invalidateAboutTableAndFencePublication(table); + } finally { + invalidationFinished.countDown(); + } + return null; + }); + publication.get(30, TimeUnit.SECONDS); + invalidation.get(30, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + } + Assertions.assertTrue(sqlCacheManager.getSqlCaches().asMap().isEmpty()); + + prepareFeCacheContext(sql, table); + sqlCacheManager.tryAddFeSqlCache(connectContext, sql); + Assertions.assertEquals(1, sqlCacheManager.getSqlCaches().asMap().size()); Review Comment: [P2] Isolate the singleton cache between test methods This added test deliberately exits with one entry in the Env-owned SQL-cache singleton, and two other added publication tests can do the same. `TestWithFeService` has no per-method cache cleanup, while `testSqlCache()` assumes an empty global cache, inserts two entries, and asserts the exact size is two; for example, running this producer immediately before `testSqlCache()` makes that assertion observe three. That existing test also changes `sql_cache_manage_num` to one without restoring it. Please clear the cache in per-method setup/cleanup and restore the mutable capacity so selected, IDE, and reordered execution are independent. ########## fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java: ########## @@ -893,6 +1100,74 @@ private void validateTableAndColumns(TableNameInfo tableNameInfo, fk.getReferencedColumnNames(), toKey(refTableInfo)); } + } else if (constraint instanceof DistributionMappingConstraint) { + validateDistributionMappingConstraint( + tableNameInfo, table, (DistributionMappingConstraint) constraint); + } + return table; + } + + private TableIf resolveTableIfPresent(TableNameInfo tableNameInfo) { + try { + return resolveTableForValidation(tableNameInfo); + } catch (AnalysisException e) { + LOG.debug("Table {} is unavailable while synchronizing table-local constraints", + tableNameInfo, e); + return null; + } + } + + @SuppressWarnings("deprecation") + private void putTableLocalConstraint(TableIf table, String constraintName, Constraint constraint) { + if (table instanceof Table) { + ((Table) table).getTableAttributes().getConstraintsMap().put(constraintName, constraint); + } + } + + @SuppressWarnings("deprecation") + private void removeTableLocalConstraint(TableIf table, String constraintName) { + if (table instanceof Table) { + ((Table) table).getTableAttributes().getConstraintsMap().remove(constraintName); + } + } + + private void validateDistributionMappingConstraint(TableNameInfo tableNameInfo, TableIf table, + DistributionMappingConstraint constraint) { + if (!(table instanceof OlapTable)) { + throw new AnalysisException("Distribution mapping constraint only supports OLAP tables"); + } + validateColumnsExist(table, constraint.getDeterminantColumnNames(), toKey(tableNameInfo)); + validateColumnsExist(table, constraint.getDistributionColumnNames(), toKey(tableNameInfo)); + TreeSet<String> determinantColumns = new TreeSet<>(String.CASE_INSENSITIVE_ORDER); + determinantColumns.addAll(constraint.getDeterminantColumnNames()); + if (determinantColumns.size() != constraint.getDeterminantColumnNames().size()) { + throw new AnalysisException("Determinant columns in distribution mapping constraint must be unique"); + } + TreeSet<String> distributionColumns = new TreeSet<>(String.CASE_INSENSITIVE_ORDER); + distributionColumns.addAll(constraint.getDistributionColumnNames()); + if (distributionColumns.size() != constraint.getDistributionColumnNames().size()) { + throw new AnalysisException("Distribution columns in distribution mapping constraint must be unique"); + } + + OlapTable olapTable = (OlapTable) table; + if (!(olapTable.getDefaultDistributionInfo() instanceof HashDistributionInfo)) { Review Comment: [P2] Fence HASH-to-RANDOM conversion against mappings This validation establishes that a mapping is valid only while the table has hash distribution and its targets are an ordered subset of the hash key. The parallel `distribution_type=random` ALTER path still calls `Env.convertDistributionType()`, and leader plus replay replace the default and partition distributions without rejecting or removing either mapping copy. The table then retains persisted/SHOW/recover metadata that could no longer be added, and mapped-column DROP/RENAME remains blocked even though random scans cannot use the proof. Please reject HASH-to-RANDOM while a mapping exists (or atomically remove and journal it) before conversion, with replay or image-round-trip coverage. -- 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]
