cherry-pick to 1.1-dev: add foreign key self refer (#14887)

docs： 见 https://github.com/matrixorigin/docs/pull/270/files

### 外键逻辑的修改

例子：
```sql
create table t1(
a int primary key,
b int,
constraint `c1` foreign key `fk1` (b) references t1(a)
)
```

1. create table 增加外键自引用

- 名称的处理

mysql 会为b建立次级索引KEY `fk1` (b)。将fk1作为次级索引的名称。
约束名称为c1。

mo不会在b上建立次级索引。因此fk1不被使用。

- 新增constraint name

原先没指定constraint name时，默认是空串。没有constraint name，alter table无法删除外键。

当输入的constraint name为空串时，生成一个uuid。
当输入的constraint name为空白符串时，报错。
其它，用输入的constarint name。

在增加constraint时，会检查重复。
在删除constraint时，会检查是否存在。

- getForeignKeyData拆出来checkFkColsAreValid。

checkFkColsAreValid检查外键的列是否合法

对于非自引用外键，getForeignKeyData的逻辑不变。
对于自引用外键，getForeignKeyData仅处理外键的定义，生产fkdata。
等tableDef的pk，uk都准备好后。再由checkFkColsAreValid检查外键的列是否合法。
因为pk，uk在语法上，可能在外键定义之后。



2. alter table 增加/删除 外键自引用


- 新增alter table ... drop constraint `c1`

alter table ... drop foreign key `c1` 与 alter table ... drop constraint `c1` 相同都是删除外键。
drop foreign key 的名称也要填`c1`，而不是`fk1`

- alter table ... add/drop constraint/foreign key 会检查约束的存在与否。

- alter table ... add constraint/foreign key

会检查表中的数据满足外键条件。如果不满足，报错。且add失败。

3. drop/truncate table

表被外键引用，都是外键自引用。表是可以删除的。

4. 外键自引用时，父表id 和 自表id都设置为0.

plan.ForeignKeyDef.ForeignTbl = 0
RefChildTableDef.Tables 中自引用的tableId也为0

因此碰到tableId为0的情况，说明与外键自引用有关，需要特殊处理。

5. 特殊的case

- insert into t2 values (1,2),(2,1);

- mo不报错
- mysql 报错

- insert into t2 values (2,2);

都不报错。

### 用sql检查外键自引用

mo的外键自引用检查在插入数据之后，进行的。
与mysql的做法不完全相同，mysql没有深入调研。

1. 生成外键检查sql

需要检查的场景：

alter table ... add foreign key
alter table ... add constraint
insert
update
load

在上述语句构建plan时，会生成一个外键检查的sql。`genSqlForCheckFKConstraints`和`genSqlsForCheckFKSelfRefer`
构造这样的sql。

构建sql的方式，检查外键约束是否满足。

单个字段情形

父表:
T(a)
子表:
S(b)
foreign key (b) references T(a)


生成的sql :
select count(*) == 0 from (
select distinct S.b from S where S.b is not null
except
select distinct T.a from T
)
如果结果是true，则表中的数据满足约束条件。

多个字段情形

父表:
T(a,b)
子表:
S(c,d)
foreign key (c,d) references T(a,b)


生成的sql :
select count(*) == 0 from (
select distinct S.c,S.d from S where S.c is not null and S.d is not null
except
select distinct T.a,T.b from T
)
如果结果是true，则表中的数据满足约束条件。

如果有多个外键，每个外键都会生成一个sql。
对于外键自引用，S和T是同一个表。

2. 在语句执行完成后，会再执行外键检查sql。

如果检查失败，会报错。
在compile.run中，defer函数会执行`detectFkSelfRefer`来检查外键约束。

### 已知的问题

乐观事务下，下面的case 与 悲观事务的行为不一致。

```
[SCRIPT   FILE]: foreign_key/fk_self_refer2.sql
[ROW    NUMBER]: 117
[SQL STATEMENT]: insert into t1 values (1,2,3);
[EXPECT RESULT]:
Duplicate entry '1' for key '__mo_index_idx_col'
[ACTUAL RESULT]:
Cannot add or update a child row: a foreign key constraint fails

[SCRIPT   FILE]: foreign_key/fk_self_refer3.sql
[ROW    NUMBER]: 43
[SQL STATEMENT]: update t1 set b = 3 where b = 4;
[EXPECT RESULT]:
Duplicate entry '(1,3)' for key '__mo_cpkey_col'
[ACTUAL RESULT]:
Duplicate entry '3a15013a1503' for key '__mo_cpkey_col'

[SCRIPT   FILE]: foreign_key/fk_self_refer3.sql
[ROW    NUMBER]: 58
[SQL STATEMENT]: update t1 set c = 2 where b = 5;
[EXPECT RESULT]:
Duplicate entry '(1,2)' for key '__mo_index_idx_col'
[ACTUAL RESULT]:
Duplicate entry '3a15013a1502' for key '__mo_index_idx_col'

[SCRIPT   FILE]: foreign_key/fk_self_refer3.sql
[ROW    NUMBER]: 160
[SQL STATEMENT]: update t1 set c = 4 where b = 3;
[EXPECT RESULT]:
Duplicate entry '(1,4)' for key '__mo_index_idx_col'
[ACTUAL RESULT]:
Duplicate entry '3a15013a1504' for key '__mo_index_idx_col'

```

### 进一步

foreign_key_checks
在次级索引上加外键

Approved by: @nnsgmsone, @heni02, @ouyuanning, @aunjgr, @zhangxu19830126, @iamlinjunhong, @sukki37

.gitignore

@@ -31,6 +31,7 @@ pkg/udf/pythonservice/pyserver/udf/
 pkg/udf/pythonservice/pyserver/__pycache__/
 path_to_file
 tester-log/
+pkg/util/trace/impl/motrace/pprof/
 
 # bvt generate files
 test/distributed/resources/into_outfile/outfile_*.csv

pkg/common/moerr/error.go

@@ -154,7 +154,10 @@ const (
 	ErrDependentByPartitionFunction             uint16 = 20465
 	ErrAlterOperationNotSupportedReasonFkRename uint16 = 20466
 	ErrPrimaryCantHaveNull                      uint16 = 20467
-
+	ErrPartitionMgmtOnNonpartitioned            uint16 = 20468
+	ErrFKRowIsReferenced                        uint16 = 20469
+	ErrDuplicateKeyName                         uint16 = 20470
+	ErrFKNoReferencedRow2                       uint16 = 20471
 	// Group 5: rpc timeout
 	// ErrRPCTimeout rpc timeout
 	ErrRPCTimeout uint16 = 20500
@@ -376,7 +379,10 @@ var errorMsgRefer = map[uint16]moErrorMsgItem{
 	ErrDependentByPartitionFunction:             {ER_DEPENDENT_BY_PARTITION_FUNC, []string{MySQLDefaultSqlState}, "Column '%s' has a partitioning function dependency and cannot be dropped or renamed"},
 	ErrAlterOperationNotSupportedReasonFkRename: {ER_ALTER_OPERATION_NOT_SUPPORTED_REASON_FK_RENAME, []string{MySQLDefaultSqlState}, "Columns participating in a foreign key are renamed"},
 	ErrPrimaryCantHaveNull:                      {ER_PRIMARY_CANT_HAVE_NULL, []string{MySQLDefaultSqlState}, "All parts of a PRIMARY KEY must be NOT NULL; if you need NULL in a key, use UNIQUE instead"},
-
+	ErrPartitionMgmtOnNonpartitioned:            {ER_PARTITION_MGMT_ON_NONPARTITIONED, []string{MySQLDefaultSqlState}, "Partition management on a not partitioned table is not possible"},
+	ErrFKRowIsReferenced:                        {ER_ROW_IS_REFERENCED, []string{MySQLDefaultSqlState}, "Cannot delete or update a parent row: a foreign key constraint fails"},
+	ErrDuplicateKeyName:                         {ER_DUP_KEYNAME, []string{MySQLDefaultSqlState}, "Duplicate foreign key constraint name '%-.192s'"},
+	ErrFKNoReferencedRow2:                       {ER_NO_REFERENCED_ROW_2, []string{"23000"}, "Cannot add or update a child row: a foreign key constraint fails"},
 	// Group 5: rpc timeout
 	ErrRPCTimeout:   {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "rpc timeout"},
 	ErrClientClosed: {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "client closed"},
@@ -1305,6 +1311,18 @@ func NewErrTooManyParameter(ctx context.Context) *Error {
 	return newError(ctx, ErrTooManyParameter)
 }
 
+func NewErrFKRowIsReferenced(ctx context.Context) *Error {
+	return newError(ctx, ErrFKRowIsReferenced)
+}
+
+func NewErrDuplicateKeyName(ctx context.Context, fkName any) *Error {
+	return newError(ctx, ErrDuplicateKeyName, fkName)
+}
+
+func NewErrFKNoReferencedRow2(ctx context.Context) *Error {
+	return newError(ctx, ErrFKNoReferencedRow2)
+}
+
 var contextFunc atomic.Value
 
 func SetContextFunc(f func() context.Context) {

pkg/sql/compile/alter.go

@@ -143,7 +143,7 @@ func (s *Scope) AlterTableCopy(c *Compile) error {
 			return err
 		}
 
-		// update foreign key child table references
+		// update foreign key child table references to the current table
 		for _, tblId := range qry.CopyTableDef.RefChildTbls {
 			if err = updateTableForeignKeyColId(c, qry.ChangeTblColIdMap, tblId, originRel.GetTableID(c.ctx), newRel.GetTableID(c.ctx)); err != nil {
 				return err
@@ -172,9 +172,16 @@ func (s *Scope) AlterTable(c *Compile) error {
 
 // updateTableForeignKeyColId update foreign key colid of child table references
 func updateTableForeignKeyColId(c *Compile, changColDefMap map[uint64]*plan.ColDef, childTblId uint64, oldParentTblId uint64, newParentTblId uint64) error {
-	_, _, childRelation, err := c.e.GetRelationById(c.ctx, c.proc.TxnOperator, childTblId)
-	if err != nil {
-		return err
+	var childRelation engine.Relation
+	var err error
+	if childTblId == 0 {
+		//fk self refer does not update
+		return nil
+	} else {
+		_, _, childRelation, err = c.e.GetRelationById(c.ctx, c.proc.TxnOperator, childTblId)
+		if err != nil {
+			return err
+		}
 	}
 	childTableDef, err := childRelation.TableDefs(c.ctx)
 	if err != nil {

pkg/sql/compile/compile.go

@@ -431,6 +431,21 @@ func (c *Compile) Run(_ uint64) (result *util2.RunResult, err error) {
 			if runC.fuzzy != nil && runC.fuzzy.cnt > 0 && err == nil {
 				err = runC.fuzzy.backgroundSQLCheck(runC)
 			}
+
+			//detect fk self refer
+			//update, insert
+			query := c.pn.GetQuery()
+			if err == nil && query != nil && (query.StmtType == plan.Query_INSERT ||
+				query.StmtType == plan.Query_UPDATE) && len(query.GetDetectSqls()) != 0 {
+				err = detectFkSelfRefer(runC, query.DetectSqls)
+			}
+			//alter table ... add/drop foreign key
+			if err == nil && c.pn.GetDdl() != nil {
+				alterTable := c.pn.GetDdl().GetAlterTable()
+				if alterTable != nil && len(alterTable.GetDetectSqls()) != 0 {
+					err = detectFkSelfRefer(runC, alterTable.GetDetectSqls())
+				}
+			}
 		}
 
 		putCompile(c)
@@ -4237,3 +4252,39 @@ func (c *Compile) fatalLog(retry int, err error) {
 func (c *Compile) SetBuildPlanFunc(buildPlanFunc func() (*plan2.Plan, error)) {
 	c.buildPlanFunc = buildPlanFunc
 }
+
+// detectFkSelfRefer checks if foreign key self refer confirmed
+func detectFkSelfRefer(c *Compile, detectSqls []string) error {
+	if len(detectSqls) == 0 {
+		return nil
+	}
+	for _, sql := range detectSqls {
+		err := runDetectSql(c, sql)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// runDetectSql runs the fk detecting sql
+func runDetectSql(c *Compile, sql string) error {
+	res, err := c.runSqlWithResult(sql)
+	if err != nil {
+		logutil.Errorf("The sql that caused the fk self refer check failed is %s, and generated background sql is %s", c.sql, sql)
+		return err
+	}
+	defer res.Close()
+
+	if res.Batches != nil {
+		vs := res.Batches[0].Vecs
+		if vs != nil && vs[0].Length() > 0 {
+			yes := vector.GetFixedAt[bool](vs[0], 0)
+			if !yes {
+				return moerr.NewErrFKNoReferencedRow2(c.ctx)
+			}
+		}
+	}
+	return nil
+}