[BugFix] Add GlobalConstraintManager to manage foreign key constraint…

…s parent and children relation (#50737)

Signed-off-by: shuming.li <[email protected]>

fe/fe-core/src/main/java/com/starrocks/alter/AlterMVJobExecutor.java

@@ -19,11 +19,11 @@
 import com.starrocks.analysis.IntLiteral;
 import com.starrocks.analysis.StringLiteral;
 import com.starrocks.analysis.TableName;
-import com.starrocks.catalog.ForeignKeyConstraint;
 import com.starrocks.catalog.MaterializedView;
 import com.starrocks.catalog.Table;
 import com.starrocks.catalog.TableProperty;
-import com.starrocks.catalog.UniqueConstraint;
+import com.starrocks.catalog.constraint.ForeignKeyConstraint;
+import com.starrocks.catalog.constraint.UniqueConstraint;
 import com.starrocks.common.DdlException;
 import com.starrocks.common.MetaNotFoundException;
 import com.starrocks.common.Pair;

fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeHandler.java

@@ -53,7 +53,6 @@
 import com.starrocks.catalog.Column;
 import com.starrocks.catalog.ColumnId;
 import com.starrocks.catalog.Database;
-import com.starrocks.catalog.ForeignKeyConstraint;
 import com.starrocks.catalog.Index;
 import com.starrocks.catalog.KeysType;
 import com.starrocks.catalog.LocalTablet;
@@ -72,7 +71,8 @@
 import com.starrocks.catalog.TableProperty;
 import com.starrocks.catalog.Tablet;
 import com.starrocks.catalog.Type;
-import com.starrocks.catalog.UniqueConstraint;
+import com.starrocks.catalog.constraint.ForeignKeyConstraint;
+import com.starrocks.catalog.constraint.UniqueConstraint;
 import com.starrocks.common.AnalysisException;
 import com.starrocks.common.Config;
 import com.starrocks.common.DdlException;

fe/fe-core/src/main/java/com/starrocks/catalog/BaseTableInfo.java

@@ -110,6 +110,24 @@ public long getTableId() {
         return this.tableId;
     }
 
+    /**
+     * Called when a table is renamed.
+     * @param newTable the new table with the new table name
+     */
+    public void onTableRename(Table newTable, String oldTableName) {
+        if (newTable == null) {
+            return;
+        }
+
+        // only changes the table name if the old table name is the same as the current table name
+        if (this.tableName != null && this.tableName.equals(oldTableName)) {
+            if (newTable instanceof OlapTable) {
+                this.tableId = newTable.getId();
+            }
+            this.tableName = newTable.getName();
+        }
+    }
+
     public String toString() {
         if (isInternalCatalog()) {
             return Joiner.on(".").join(InternalCatalog.DEFAULT_INTERNAL_CATALOG_NAME, dbId, tableId);

fe/fe-core/src/main/java/com/starrocks/catalog/HiveTable.java

@@ -45,6 +45,8 @@
 import com.starrocks.analysis.DescriptorTable.ReferencedPartitionInfo;
 import com.starrocks.analysis.Expr;
 import com.starrocks.analysis.LiteralExpr;
+import com.starrocks.catalog.constraint.ForeignKeyConstraint;
+import com.starrocks.catalog.constraint.UniqueConstraint;
 import com.starrocks.common.Config;
 import com.starrocks.common.util.TimeUtils;
 import com.starrocks.common.util.concurrent.lock.LockType;

fe/fe-core/src/main/java/com/starrocks/catalog/MaterializedView.java

@@ -37,6 +37,7 @@
 import com.starrocks.backup.mv.MvBackupInfo;
 import com.starrocks.backup.mv.MvBaseTableBackupInfo;
 import com.starrocks.backup.mv.MvRestoreContext;
+import com.starrocks.catalog.constraint.ForeignKeyConstraint;
 import com.starrocks.common.DdlException;
 import com.starrocks.common.FeConstants;
 import com.starrocks.common.MaterializedViewExceptions;

fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java

@@ -66,6 +66,9 @@
 import com.starrocks.catalog.MaterializedIndex.IndexState;
 import com.starrocks.catalog.Partition.PartitionState;
 import com.starrocks.catalog.Replica.ReplicaState;
+import com.starrocks.catalog.constraint.ForeignKeyConstraint;
+import com.starrocks.catalog.constraint.GlobalConstraintManager;
+import com.starrocks.catalog.constraint.UniqueConstraint;
 import com.starrocks.clone.TabletChecker;
 import com.starrocks.clone.TabletSchedCtx;
 import com.starrocks.clone.TabletScheduler;
@@ -570,6 +573,7 @@ public void setName(String newName) {
         }
 
         // change name
+        String oldTableName = this.name;
         this.name = newName;
 
         // change single partition name
@@ -590,6 +594,25 @@ public void setName(String newName) {
             Preconditions.checkState(expressionRangePartitionInfo.getPartitionExprsSize() == 1);
             expressionRangePartitionInfo.renameTableName("", newName);
         }
+
+        if (tableProperty != null) {
+            // renew unique constraints
+            List<UniqueConstraint> tpUniqueConstraints = tableProperty.getUniqueConstraints();
+            if (!CollectionUtils.isEmpty(tpUniqueConstraints)) {
+                for (UniqueConstraint constraint : tpUniqueConstraints) {
+                    constraint.onTableRename(this, oldTableName);
+                }
+                setUniqueConstraints(tpUniqueConstraints);
+            }
+            // renew foreign constraints
+            List<ForeignKeyConstraint> tpForeignConstraints = tableProperty.getForeignKeyConstraints();
+            if (!CollectionUtils.isEmpty(tpForeignConstraints)) {
+                for (ForeignKeyConstraint constraint : tpForeignConstraints) {
+                    constraint.onTableRename(this, oldTableName);
+                }
+                setForeignKeyConstraints(tpForeignConstraints);
+            }
+        }
     }
 
     public boolean hasMaterializedIndex(String indexName) {
@@ -3015,6 +3038,10 @@ public void onReload() {
         analyzePartitionInfo();
         analyzeRollupIndexMeta();
         tryToAssignIndexId();
+
+        // register constraints from global state manager
+        GlobalConstraintManager globalConstraintManager = GlobalStateMgr.getCurrentState().getGlobalConstraintManager();
+        globalConstraintManager.registerConstraint(this);
     }
 
     @Override
@@ -3146,6 +3173,10 @@ public void onDrop(Database db, boolean force, boolean replay) {
         if (!replay && hasAutoIncrementColumn()) {
             sendDropAutoIncrementMapTask();
         }
+
+        // unregister constraints from global state manager
+        GlobalConstraintManager globalConstraintManager = GlobalStateMgr.getCurrentState().getGlobalConstraintManager();
+        globalConstraintManager.unRegisterConstraint(this);
     }
 
     public void removeTableBinds(boolean isReplay) {

fe/fe-core/src/main/java/com/starrocks/catalog/Table.java

@@ -41,6 +41,8 @@
 import com.google.common.collect.Sets;
 import com.google.gson.annotations.SerializedName;
 import com.starrocks.analysis.DescriptorTable.ReferencedPartitionInfo;
+import com.starrocks.catalog.constraint.ForeignKeyConstraint;
+import com.starrocks.catalog.constraint.UniqueConstraint;
 import com.starrocks.catalog.system.SystemTable;
 import com.starrocks.common.io.Text;
 import com.starrocks.common.io.Writable;

fe/fe-core/src/main/java/com/starrocks/catalog/TableProperty.java

@@ -44,6 +44,8 @@
 import com.google.gson.annotations.SerializedName;
 import com.starrocks.analysis.TableName;
 import com.starrocks.binlog.BinlogConfig;
+import com.starrocks.catalog.constraint.ForeignKeyConstraint;
+import com.starrocks.catalog.constraint.UniqueConstraint;
 import com.starrocks.common.Config;
 import com.starrocks.common.io.Text;
 import com.starrocks.common.io.Writable;

fe/fe-core/src/main/java/com/starrocks/catalog/constraint/Constraint.java

@@ -0,0 +1,55 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 com.starrocks.catalog.constraint;
+
+import com.starrocks.catalog.Table;
+
+/**
+ * Base class for constraints.
+ */
+public abstract class Constraint {
+    // NOTE: A constraint may contain a name, but currently starrocks only supports internal constraints.
+    public static final String TABLE_PROPERTY_CONSTRAINT = "_TABLE_PROPERTIES_";
+
+    /**
+     * Type of constraint, unique/foreign key are supported now.
+     */
+    public enum ConstraintType {
+        UNIQUE,
+        FOREIGN_KEY
+    }
+    private final String name;
+    private final ConstraintType type;
+
+    public Constraint(ConstraintType type, String name) {
+        this.name = name;
+        this.type = type;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public ConstraintType getType() {
+        return type;
+    }
+
+    /**
+     * Called when a table is renamed.
+     * @param newTable
+     * @return
+     */
+    public abstract void onTableRename(Table newTable, String oldTableName);
+}

fe/fe-core/src/main/java/com/starrocks/catalog/ForeignKeyConstraint.java → fe/fe-core/src/main/java/com/starrocks/catalog/constraint/ForeignKeyConstraint.java

@@ -12,34 +12,41 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-
-package com.starrocks.catalog;
+package com.starrocks.catalog.constraint;
 
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Streams;
+import com.starrocks.catalog.BaseTableInfo;
+import com.starrocks.catalog.Column;
+import com.starrocks.catalog.ColumnId;
+import com.starrocks.catalog.Database;
+import com.starrocks.catalog.InternalCatalog;
+import com.starrocks.catalog.OlapTable;
+import com.starrocks.catalog.Table;
 import com.starrocks.common.Pair;
 import com.starrocks.server.GlobalStateMgr;
 import com.starrocks.sql.analyzer.SemanticException;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang.math.NumberUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Objects;
+import java.util.Set;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
-// foreign key constraint is used to guide optimizer rewrite for now,
-// and is not enforced during ingestion.
-// the foreign key property of data should be guaranteed by user.
-//
-// a table may have multi foreign key constraints.
-public class ForeignKeyConstraint {
+// Foreign-key constraint is used to guide optimizer rewrite for now, and is not enforced during ingestion.
+// User should guarantee the foreign key property of data.
+// A table may have multi-foreign-key constraints.
+public class ForeignKeyConstraint extends Constraint {
     private static final Logger LOG = LogManager.getLogger(ForeignKeyConstraint.class);
 
     private static final String FOREIGN_KEY_REGEX = "((\\.?\\w+:?-?)*)\\s*\\(((,?\\s*\\w+\\s*)+)\\)\\s+((?i)REFERENCES)\\s+" +
@@ -57,10 +64,10 @@ public class ForeignKeyConstraint {
     // eg: [column1 -> column1', column2 -> column2']
     private final List<Pair<ColumnId, ColumnId>> columnRefPairs;
 
-    public ForeignKeyConstraint(
-            BaseTableInfo parentTableInfo,
-            BaseTableInfo childTableInfo,
-            List<Pair<ColumnId, ColumnId>> columnRefPairs) {
+    public ForeignKeyConstraint(BaseTableInfo parentTableInfo,
+                                BaseTableInfo childTableInfo,
+                                List<Pair<ColumnId, ColumnId>> columnRefPairs) {
+        super(ConstraintType.FOREIGN_KEY, TABLE_PROPERTY_CONSTRAINT);
         this.parentTableInfo = parentTableInfo;
         this.childTableInfo = childTableInfo;
         this.columnRefPairs = columnRefPairs;
@@ -87,18 +94,12 @@ public List<Pair<String, String>> getColumnNameRefPairs(Table defaultChildTable)
         List<Pair<String, String>> result = new ArrayList<>(columnRefPairs.size());
         for (Pair<ColumnId, ColumnId> pair : columnRefPairs) {
             Column childColumn = childTable.getColumn(pair.first);
-            if (childColumn == null) {
-                LOG.warn("can not find column by column id: {} in table: {}, the column may have been dropped",
-                        pair.first, childTableInfo);
-                continue;
-            }
             Column parentColumn = parentTable.getColumn(pair.second);
-            if (parentColumn == null) {
+            if (childColumn == null || parentColumn == null) {
                 LOG.warn("can not find column by column id: {} in table: {}, the column may have been dropped",
-                        pair.second, parentTableInfo);
+                        pair.first, childTableInfo);
                 continue;
             }
-
             result.add(Pair.create(childColumn.getName(), parentColumn.getName()));
         }
 
@@ -285,4 +286,64 @@ public static String getShowCreateTableConstraintDesc(OlapTable baseTable, List<
 
         return Joiner.on(";").join(constraintStrs);
     }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(columnRefPairs, parentTableInfo, childTableInfo);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || !(obj instanceof ForeignKeyConstraint)) {
+            return false;
+        }
+        ForeignKeyConstraint that = (ForeignKeyConstraint) obj;
+        return Objects.equals(columnRefPairs, that.columnRefPairs) &&
+                Objects.equals(parentTableInfo, that.parentTableInfo) &&
+                Objects.equals(childTableInfo, that.childTableInfo);
+    }
+
+    public void onTableRename(Table table, String oldTableName) {
+        LOG.info("ForeignKeyConstraint onTableRename, table: {}, oldTableName: {}, parentTableInfo: {}, childTableInfo: {}",
+                table.getName(), oldTableName, parentTableInfo, childTableInfo);
+        // parentTableInfo: how to know which other tables that ref for this table?
+        GlobalConstraintManager globalConstraintManager = GlobalStateMgr.getCurrentState().getGlobalConstraintManager();
+        Set<TableWithFKConstraint> refConstraints = globalConstraintManager.getRefConstraints(table);
+        if (!CollectionUtils.isEmpty(refConstraints)) {
+            for (TableWithFKConstraint tableWithFKConstraint : refConstraints) {
+                Table childTable = tableWithFKConstraint.getChildTable();
+                if (childTable == null) {
+                    continue;
+                }
+                List<ForeignKeyConstraint> fks = childTable.getForeignKeyConstraints();
+                if (!CollectionUtils.isEmpty(fks)) {
+                    // refresh child table's foreign key constraints
+                    List<ForeignKeyConstraint> newFKConstraints = Lists.newArrayList();
+                    boolean isChanged = false;
+                    for (ForeignKeyConstraint fk : fks) {
+                        if (fk.parentTableInfo != null) {
+                            fk.parentTableInfo.onTableRename(table, oldTableName);
+                            isChanged = true;
+                        }
+                        newFKConstraints.add(fk);
+                    }
+                    if (isChanged) {
+                        LOG.info("refresh child table's foreign key constraints, parent table, child table: {}, " +
+                                "newFKConstraints: {}", table.getName(), childTable.getName(), newFKConstraints);
+                        childTable.setForeignKeyConstraints(newFKConstraints);
+                    }
+                }
+            }
+        }
+
+        // childTableInfo
+        if (childTableInfo != null) {
+            childTableInfo.onTableRename(table, oldTableName);
+        }
+
+        // columnRefPairs: no needs to change after table rename
+    }
 }