diff --git a/pkg/ddl/job_table.go b/pkg/ddl/job_table.go
index fa8234e14916b..80ce5bd74d464 100644
--- a/pkg/ddl/job_table.go
+++ b/pkg/ddl/job_table.go
@@ -421,6 +421,7 @@ func (s *jobScheduler) startDispatch() error {
 		if err := s.checkAndUpdateClusterState(false); err != nil {
 			continue
 		}
+		failpoint.InjectCall("beforeAllLoadDDLJobAndRun")
 		s.loadDDLJobAndRun(se, s.generalDDLWorkerPool, s.getGeneralJob)
 		s.loadDDLJobAndRun(se, s.reorgWorkerPool, s.getReorgJob)
 	}
diff --git a/pkg/ddl/job_table_test.go b/pkg/ddl/job_table_test.go
index 1ebf442cff256..e09e0c08c9288 100644
--- a/pkg/ddl/job_table_test.go
+++ b/pkg/ddl/job_table_test.go
@@ -228,3 +228,38 @@ func TestUpgradingRelatedJobState(t *testing.T) {
 		dom.DDL().StateSyncer().UpdateGlobalState(context.Background(), &syncer.StateInfo{State: syncer.StateNormalRunning})
 	}
 }
+
+func TestGeneralDDLWithQuery(t *testing.T) {
+	store, _ := testkit.CreateMockStoreAndDomain(t)
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("CREATE TABLE t (id INT NOT NULL);")
+
+	var beforeRunCh = make(chan struct{})
+	testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/ddl/beforeAllLoadDDLJobAndRun", func() {
+		<-beforeRunCh
+	})
+	var ch = make(chan struct{})
+	testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/ddl/waitJobSubmitted", func() {
+		<-ch
+	})
+	// 2 general DDLs shouldn't be blocked by each other for MDL, i.e. the "create view xx from select xxx"
+	// should not fill the MDL related tables.
+	var wg util.WaitGroupWrapper
+	wg.Run(func() {
+		tk := testkit.NewTestKit(t, store)
+		tk.MustExec("use test")
+		tk.MustExec("alter table t add column b int")
+	})
+	ch <- struct{}{}
+	wg.Run(func() {
+		tk := testkit.NewTestKit(t, store)
+		tk.MustExec("use test")
+		tk.MustExec("create view v as select * from t")
+	})
+	ch <- struct{}{}
+	tk.MustQuery("select count(1) from mysql.tidb_ddl_job").Check(testkit.Rows("2"))
+	close(beforeRunCh)
+	wg.Wait()
+}
diff --git a/pkg/executor/ddl.go b/pkg/executor/ddl.go
index 6090efc170641..7883dba3fac30 100644
--- a/pkg/executor/ddl.go
+++ b/pkg/executor/ddl.go
@@ -309,6 +309,7 @@ func (e *DDLExec) executeCreateView(ctx context.Context, s *ast.CreateViewStmt)
 		return exeerrors.ErrViewInvalid.GenWithStackByArgs(s.ViewName.Schema.L, s.ViewName.Name.L)
 	}
 
+	e.Ctx().GetSessionVars().ClearRelatedTableForMDL()
 	return domain.GetDomain(e.Ctx()).DDL().CreateView(e.Ctx(), s)
 }
 
diff --git a/pkg/planner/core/preprocess.go b/pkg/planner/core/preprocess.go
index e98999eb6d938..1b1747dcb50b7 100644
--- a/pkg/planner/core/preprocess.go
+++ b/pkg/planner/core/preprocess.go
@@ -148,7 +148,9 @@ const (
 	inPrepare preprocessorFlag = 1 << iota
 	// inTxnRetry is set when visiting in transaction retry.
 	inTxnRetry
-	// inCreateOrDropTable is set when visiting create/drop table statement.
+	// inCreateOrDropTable is set when visiting create/drop table/view/sequence,
+	// rename table, alter table add foreign key, and BR restore.
+	// TODO need a better name to clarify it's meaning
 	inCreateOrDropTable
 	// parentIsJoin is set when visiting node's parent is join.
 	parentIsJoin
@@ -927,6 +929,8 @@ func (p *preprocessor) checkCreateTableGrammar(stmt *ast.CreateTableStmt) {
 	}
 	if stmt.Select != nil {
 		// FIXME: a temp error noticing 'not implemented' (issue 4754)
+		// Note: if we implement it later, please clear it's MDL related tables for
+		// it like what CREATE VIEW does.
 		p.err = errors.New("'CREATE TABLE ... SELECT' is not implemented yet")
 		return
 	} else if len(stmt.Cols) == 0 && stmt.ReferTable == nil {
diff --git a/pkg/sessionctx/variable/session.go b/pkg/sessionctx/variable/session.go
index bfada56dc2c83..530f9b54cc08b 100644
--- a/pkg/sessionctx/variable/session.go
+++ b/pkg/sessionctx/variable/session.go
@@ -3636,6 +3636,16 @@ func (s *SessionVars) GetRelatedTableForMDL() *sync.Map {
 	return s.TxnCtx.relatedTableForMDL
 }
 
+// ClearRelatedTableForMDL clears the related table for MDL.
+// related tables for MDL is filled during build logical plan or Preprocess for all DataSources,
+// even for queries inside DDLs like `create view as select xxx` and `create table as select xxx`.
+// it should be cleared before we execute the DDL statement.
+func (s *SessionVars) ClearRelatedTableForMDL() {
+	s.TxnCtx.tdmLock.Lock()
+	defer s.TxnCtx.tdmLock.Unlock()
+	s.TxnCtx.relatedTableForMDL = nil
+}
+
 // EnableForceInlineCTE returns the session variable enableForceInlineCTE
 func (s *SessionVars) EnableForceInlineCTE() bool {
 	return s.enableForceInlineCTE