fix drop database without lock table (#19698)

fix drop database without lock table

Approved by: @qingxinhome, @zhangxu19830126, @m-schen, @ouyuanning, @daviszhen, @sukki37, @aunjgr

Author: iamlinjunhong

pkg/defines/type.go

@@ -227,6 +227,8 @@ type IsMoLogger struct{}
 
 type SourceScanResKey struct{}
 
+type IgnoreForeignKey struct{}
+
 // Determine if now is a bg sql.
 type BgKey struct{}
 

pkg/sql/compile/compile.go

@@ -4801,6 +4801,12 @@ func (c *Compile) runSqlWithResult(sql string, accountId int32) (executor.Result
 		WithTimeZone(c.proc.GetSessionInfo().TimeZone).
 		WithLowerCaseTableNames(&lower)
 
+	if qry, ok := c.pn.Plan.(*plan.Plan_Ddl); ok {
+		if qry.Ddl.DdlType == plan.DataDefinition_DROP_DATABASE {
+			opts = opts.WithStatementOption(executor.StatementOption{}.WithIgnoreForeignKey())
+		}
+	}
+
 	ctx := c.proc.Ctx
 	if accountId >= 0 {
 		ctx = defines.AttachAccountId(c.proc.Ctx, uint32(accountId))

pkg/sql/compile/ddl.go

@@ -130,19 +130,84 @@ func (s *Scope) DropDatabase(c *Compile) error {
 		}
 	}
 
+	// whether foreign_key_checks = 0 or 1
+	err = s.removeFkeysRelationships(c, dbName)
+	if err != nil {
+		return err
+	}
+
+	database, err := c.e.Database(c.proc.Ctx, dbName, c.proc.GetTxnOperator())
+	if err != nil {
+		return err
+	}
+	relations, err := database.Relations(c.proc.Ctx)
+	if err != nil {
+		return err
+	}
+	var ignoreTables []string
+	for _, r := range relations {
+		t, err := database.Relation(c.proc.Ctx, r, nil)
+		if err != nil {
+			return err
+		}
+		defs, err := t.TableDefs(c.proc.Ctx)
+		if err != nil {
+			return err
+		}
+
+		constrain := GetConstraintDefFromTableDefs(defs)
+		for _, ct := range constrain.Cts {
+			if ds, ok := ct.(*engine.IndexDef); ok {
+				for _, d := range ds.Indexes {
+					ignoreTables = append(ignoreTables, d.IndexTableName)
+				}
+			}
+		}
+
+		for _, def := range defs {
+			if partitionDef, ok := def.(*engine.PartitionDef); ok {
+				if partitionDef.Partitioned > 0 {
+					p := &plan.PartitionByDef{}
+					err = p.UnMarshalPartitionInfo(([]byte)(partitionDef.Partition))
+					if err != nil {
+						return err
+					}
+					ignoreTables = append(ignoreTables, p.PartitionTableNames...)
+				}
+				break
+			}
+		}
+	}
+
+	deleteTables := make([]string, 0, len(relations)-len(ignoreTables))
+	for _, r := range relations {
+		isIndexTable := false
+		for _, d := range ignoreTables {
+			if d == r {
+				isIndexTable = true
+				break
+			}
+		}
+		if !isIndexTable {
+			deleteTables = append(deleteTables, r)
+		}
+	}
+
+	for _, t := range deleteTables {
+		dropSql := fmt.Sprintf(dropTableBeforeDropDatabase, dbName, t)
+		err = c.runSql(dropSql)
+		if err != nil {
+			return err
+		}
+	}
+
 	sql := s.Plan.GetDdl().GetDropDatabase().GetCheckFKSql()
 	if len(sql) != 0 {
 		if err = runDetectFkReferToDBSql(c, sql); err != nil {
 			return err
 		}
 	}
 
-	// whether foreign_key_checks = 0 or 1
-	err = s.removeFkeysRelationships(c, dbName)
-	if err != nil {
-		return err
-	}
-
 	err = c.e.Delete(c.proc.Ctx, dbName, c.proc.GetTxnOperator())
 	if err != nil {
 		return err

pkg/sql/compile/sql_executor.go

@@ -262,6 +262,11 @@ func (exec *txnExecutor) Exec(
 		defer recoverAccount(exec, originAccountID)
 	}
 	//-----------------------------------------------------------------------------------------
+	if statementOption.IgnoreForeignKey() {
+		exec.ctx = context.WithValue(exec.ctx,
+			defines.IgnoreForeignKey{},
+			true)
+	}
 
 	receiveAt := time.Now()
 	lower := exec.opts.LowerCaseTableNames()

pkg/sql/compile/util.go

@@ -95,6 +95,10 @@ var (
 	updateMoIndexesAlgoParams                           = `update mo_catalog.mo_indexes set algo_params = '%s' where table_id = %v and name = '%s';`
 )
 
+var (
+	dropTableBeforeDropDatabase = "drop table if exists `%v`.`%v`;"
+)
+
 var (
 	deleteMoTablePartitionsWithDatabaseIdFormat = `delete from mo_catalog.mo_table_partitions where database_id = %v;`
 	deleteMoTablePartitionsWithTableIdFormat    = `delete from mo_catalog.mo_table_partitions where table_id = %v;`

pkg/sql/plan/build_ddl.go

@@ -2656,8 +2656,14 @@ func buildDropTable(stmt *tree.DropTable, ctx CompilerContext) (*Plan, error) {
 			return nil, moerr.NewInternalError(ctx.GetContext(), "only the sys account can drop the cluster table")
 		}
 
+		ignore := false
+		val := ctx.GetContext().Value(defines.IgnoreForeignKey{})
+		if val != nil {
+			ignore = val.(bool)
+		}
+
 		dropTable.TableId = tableDef.TblId
-		if tableDef.Fkeys != nil {
+		if tableDef.Fkeys != nil && !ignore {
 			for _, fk := range tableDef.Fkeys {
 				if fk.ForeignTbl == 0 {
 					continue
@@ -2668,7 +2674,7 @@ func buildDropTable(stmt *tree.DropTable, ctx CompilerContext) (*Plan, error) {
 
 		// collect child tables that needs remove fk relationships
 		// with the table
-		if tableDef.RefChildTbls != nil {
+		if tableDef.RefChildTbls != nil && !ignore {
 			for _, childTbl := range tableDef.RefChildTbls {
 				if childTbl == 0 {
 					continue

pkg/sql/plan/build_dml_util.go

@@ -3552,7 +3552,13 @@ func IsFkBannedDatabase(db string) bool {
 
 // IsForeignKeyChecksEnabled returns the system variable foreign_key_checks is true or false
 func IsForeignKeyChecksEnabled(ctx CompilerContext) (bool, error) {
-	if disable := ctx.GetContext().Value(defines.DisableFkCheck{}); disable != nil {
+	ignore := ctx.GetContext().Value(defines.IgnoreForeignKey{})
+	if ignore != nil {
+		return !ignore.(bool), nil
+	}
+
+	disable := ctx.GetContext().Value(defines.DisableFkCheck{})
+	if disable != nil {
 		return !disable.(bool), nil
 	}
 

pkg/util/executor/options.go

@@ -28,6 +28,12 @@ func (opts Options) WithDisableIncrStatement() Options {
 	return opts
 }
 
+// WithDisableIncrStatement disable incr statement
+func (opts StatementOption) WithIgnoreForeignKey() StatementOption {
+	opts.ignoreForeignKey = true
+	return opts
+}
+
 // WithTxn exec sql in a exists txn
 func (opts Options) WithTxn(txnOp client.TxnOperator) Options {
 	opts.txnOp = txnOp
@@ -172,6 +178,10 @@ func (opts StatementOption) DisableLog() bool {
 	return opts.disableLog
 }
 
+func (opts StatementOption) IgnoreForeignKey() bool {
+	return opts.ignoreForeignKey
+}
+
 func (opts Options) WithDisableTrace() Options {
 	opts.txnOpts = append(opts.txnOpts, client.WithDisableTrace(true))
 	return opts

pkg/util/executor/types.go

@@ -66,9 +66,10 @@ type Options struct {
 
 // StatementOption statement execute option.
 type StatementOption struct {
-	waitPolicy lock.WaitPolicy
-	accountId  uint32
-	disableLog bool
+	waitPolicy       lock.WaitPolicy
+	accountId        uint32
+	disableLog       bool
+	ignoreForeignKey bool
 }
 
 // Result exec sql result