[feat](Nereids): drop foreign key after dropping primary key that is …

…referenced by the foreign key (apache#30417)

fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java

@@ -45,9 +45,11 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 public interface TableIf {
     Logger LOG = LogManager.getLogger(TableIf.class);
@@ -341,8 +343,12 @@ default void dropFKReferringPK(TableIf table, PrimaryKeyConstraint constraint) {
         writeLock();
         try {
             Map<String, Constraint> constraintMap = getConstraintsMapUnsafe();
-            constraintMap.entrySet().removeIf(e -> e.getValue() instanceof ForeignKeyConstraint
-                    && ((ForeignKeyConstraint) e.getValue()).isReferringPK(table, constraint));
+            Set<String> fkName = constraintMap.entrySet().stream()
+                    .filter(e -> e.getValue() instanceof ForeignKeyConstraint
+                    && ((ForeignKeyConstraint) e.getValue()).isReferringPK(table, constraint))
+                    .map(Entry::getKey)
+                    .collect(Collectors.toSet());
+            fkName.forEach(constraintMap::remove);
         } finally {
             writeUnlock();
         }

fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ForeignKeyConstraint.java

@@ -58,6 +58,10 @@ public Set<String> getForeignKeyNames() {
         return foreignToReference.keySet();
     }
 
+    public Set<String> getPrimaryKeyNames() {
+        return ImmutableSet.copyOf(foreignToReference.values());
+    }
+
     public Set<String> getReferencedColumnNames() {
         return ImmutableSet.copyOf(foreignToReference.values());
     }
@@ -87,7 +91,7 @@ public TableIf getReferencedTable() {
     }
 
     public Boolean isReferringPK(TableIf table, PrimaryKeyConstraint constraint) {
-        return constraint.getPrimaryKeyNames().equals(getForeignKeyNames())
+        return constraint.getPrimaryKeyNames().equals(getPrimaryKeyNames())
                 && getReferencedTable().equals(table);
     }
 

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ConstraintTest.java

@@ -61,6 +61,15 @@ public void runBeforeAll() throws Exception {
                 + "properties(\n"
                 + "    \"replication_num\"=\"1\"\n"
                 + ")");
+        createTable("create table t3 (\n"
+                + "    k1 int,\n"
+                + "    k2 int\n"
+                + ")\n"
+                + "unique key(k1, k2)\n"
+                + "distributed by hash(k1) buckets 4\n"
+                + "properties(\n"
+                + "    \"replication_num\"=\"1\"\n"
+                + ")");
     }
 
     @Test
@@ -162,4 +171,26 @@ void foreignKeyConstraintTest() throws Exception {
         PlanChecker.from(connectContext).parse("select * from t2").analyze().matches(
                 logicalOlapScan().when(o -> o.getTable().getConstraintsMapUnsafe().isEmpty()));
     }
+
+    @Test
+    void cascadeDropTest() throws Exception {
+        addConstraint("alter table t1 add constraint pk primary key (k1)");
+        addConstraint("alter table t2 add constraint fk foreign key (k1) references t1(k1)");
+        dropConstraint("alter table t1 drop constraint pk");
+
+        PlanChecker.from(connectContext).parse("select * from t2").analyze().matches(
+                logicalOlapScan().when(o -> o.getTable().getConstraintsMapUnsafe().isEmpty()));
+
+        addConstraint("alter table t1 add constraint pk primary key (k1)");
+        addConstraint("alter table t1 add constraint fk foreign key (k1) references t1(k1)");
+        addConstraint("alter table t2 add constraint fk foreign key (k1) references t1(k1)");
+        addConstraint("alter table t3 add constraint fk foreign key (k1) references t1(k1)");
+        dropConstraint("alter table t1 drop constraint pk");
+        PlanChecker.from(connectContext).parse("select * from t1").analyze().matches(
+                logicalOlapScan().when(o -> o.getTable().getConstraintsMapUnsafe().isEmpty()));
+        PlanChecker.from(connectContext).parse("select * from t2").analyze().matches(
+                logicalOlapScan().when(o -> o.getTable().getConstraintsMapUnsafe().isEmpty()));
+        PlanChecker.from(connectContext).parse("select * from t3").analyze().matches(
+                logicalOlapScan().when(o -> o.getTable().getConstraintsMapUnsafe().isEmpty()));
+    }
 }

regression-test/data/nereids_syntax_p0/constraint.out

@@ -1,10 +1,25 @@
 -- This file is automatically generated. You should know what you did if you want to edit this
--- !show_constrait --
+-- !add_primary --
+pk	PRIMARY KEY	PRIMARY KEY (id)
+
+-- !add_unique --
+uk	UNIQUE	UNIQUE (id)
+pk	PRIMARY KEY	PRIMARY KEY (id)
+
+-- !add_foreign --
 uk	UNIQUE	UNIQUE (id)
 pk	PRIMARY KEY	PRIMARY KEY (id)
-fk2	FOREIGN KEY	FOREIGN KEY (id) REFERENCES regression_test_nereids_syntax_p0.t2 (id)
-fk1	FOREIGN KEY	FOREIGN KEY (id) REFERENCES regression_test_nereids_syntax_p0.t1 (id)
+fk1	FOREIGN KEY	FOREIGN KEY (id) REFERENCES regression_test_nereids_syntax_p0.t2 (id)
+
+-- !drop_uk --
+pk	PRIMARY KEY	PRIMARY KEY (id)
+fk1	FOREIGN KEY	FOREIGN KEY (id) REFERENCES regression_test_nereids_syntax_p0.t2 (id)
+
+-- !drop_fk --
+pk	PRIMARY KEY	PRIMARY KEY (id)
+
+-- !drop_pk --
 
--- !show_constrait --
+-- !drop_fk_cascades --
 pk	PRIMARY KEY	PRIMARY KEY (id)
 

regression-test/suites/nereids_syntax_p0/constraint.groovy

@@ -61,21 +61,55 @@ suite("show_constraint") {
     alter table t2 add constraint pk primary key (id)
     """
 
+    qt_add_primary """
+    show constraints from t1;
+    """
+
     sql """
     alter table t1 add constraint uk unique (id)
     """
 
+    qt_add_unique """
+    show constraints from t1;
+    """
+
+    sql """
+    alter table t1 add constraint fk1 foreign key (id) references t2(id)
+    """
+    sql """
+    alter table t2 add constraint fk2 foreign key (id) references t1(id)
+    """
+
+    qt_add_foreign """
+    show constraints from t1;
+    """
+
     sql """
-    alter table t1 add constraint fk1 foreign key (id) references t1(id)
+    alter table t1 drop constraint uk
+    """
+
+    qt_drop_uk """
+    show constraints from t1;
     """
+
     sql """
-    alter table t1 add constraint fk2 foreign key (id) references t2(id)
+    alter table t1 drop constraint fk1
     """
 
-    qt_show_constrait """
+    qt_drop_fk """
     show constraints from t1;
     """
-    qt_show_constrait """
+
+    sql """
+    alter table t1 drop constraint pk
+    """
+
+    qt_drop_pk """
+    show constraints from t1;
+    """
+
+    qt_drop_fk_cascades """
     show constraints from t2;
     """
+
 }