diff --git a/ddl/metadatalocktest/BUILD.bazel b/ddl/metadatalocktest/BUILD.bazel
index 25e6058a3a7b4..4f0e67a4e015d 100644
--- a/ddl/metadatalocktest/BUILD.bazel
+++ b/ddl/metadatalocktest/BUILD.bazel
@@ -7,7 +7,7 @@ go_test(
         "mdl_test.go",
     ],
     flaky = True,
-    shard_count = 32,
+    shard_count = 34,
     deps = [
         "//config",
         "//ddl",
diff --git a/ddl/metadatalocktest/mdl_test.go b/ddl/metadatalocktest/mdl_test.go
index 0e9c3e32f06a6..b27ac5e6a9064 100644
--- a/ddl/metadatalocktest/mdl_test.go
+++ b/ddl/metadatalocktest/mdl_test.go
@@ -395,6 +395,90 @@ func TestMDLAutoCommitReadOnly(t *testing.T) {
 	require.Greater(t, ts1, ts2)
 }
 
+func TestMDLAnalyze(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	sv := server.CreateMockServer(t, store)
+
+	sv.SetDomain(dom)
+	dom.InfoSyncer().SetSessionManager(sv)
+	defer sv.Close()
+
+	conn1 := server.CreateMockConn(t, sv)
+	tk := testkit.NewTestKitWithSession(t, store, conn1.Context().Session)
+	conn2 := server.CreateMockConn(t, sv)
+	tkDDL := testkit.NewTestKitWithSession(t, store, conn2.Context().Session)
+	tk.MustExec("use test")
+	tk.MustExec("set global tidb_enable_metadata_lock=1")
+	tk.MustExec("create table t(a int);")
+	tk.MustExec("insert into t values(1);")
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	var ts2 time.Time
+	var ts1 time.Time
+
+	go func() {
+		tk.MustExec("begin")
+		tk.MustExec("analyze table t;")
+		tk.MustQuery("select sleep(2);")
+		tk.MustExec("commit")
+		ts1 = time.Now()
+		wg.Done()
+	}()
+
+	go func() {
+		tkDDL.MustExec("alter table test.t add column b int;")
+		ts2 = time.Now()
+		wg.Done()
+	}()
+
+	wg.Wait()
+	require.Greater(t, ts1, ts2)
+}
+
+func TestMDLAnalyzePartition(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	sv := server.CreateMockServer(t, store)
+
+	sv.SetDomain(dom)
+	dom.InfoSyncer().SetSessionManager(sv)
+	defer sv.Close()
+
+	conn1 := server.CreateMockConn(t, sv)
+	tk := testkit.NewTestKitWithSession(t, store, conn1.Context().Session)
+	conn2 := server.CreateMockConn(t, sv)
+	tkDDL := testkit.NewTestKitWithSession(t, store, conn2.Context().Session)
+	tk.MustExec("use test")
+	tk.MustExec("set @@tidb_partition_prune_mode='dynamic'")
+	tk.MustExec("set global tidb_enable_metadata_lock=1")
+	tk.MustExec("create table t(a int) partition by range(a) ( PARTITION p0 VALUES LESS THAN (0), PARTITION p1 VALUES LESS THAN (100), PARTITION p2 VALUES LESS THAN MAXVALUE );")
+	tk.MustExec("insert into t values(1), (2), (3), (4);")
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	var ts2 time.Time
+	var ts1 time.Time
+
+	go func() {
+		tk.MustExec("begin")
+		tk.MustExec("analyze table t;")
+		tk.MustExec("analyze table t partition p1;")
+		tk.MustQuery("select sleep(2);")
+		tk.MustExec("commit")
+		ts1 = time.Now()
+		wg.Done()
+	}()
+
+	go func() {
+		tkDDL.MustExec("alter table test.t drop partition p2;")
+		ts2 = time.Now()
+		wg.Done()
+	}()
+
+	wg.Wait()
+	require.Greater(t, ts1, ts2)
+}
+
 func TestMDLAutoCommitNonReadOnly(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 	sv := server.CreateMockServer(t, store)
diff --git a/planner/core/preprocess.go b/planner/core/preprocess.go
index 7ea1bf9fa3f8d..eb21166d3194c 100644
--- a/planner/core/preprocess.go
+++ b/planner/core/preprocess.go
@@ -139,7 +139,7 @@ func Preprocess(ctx context.Context, sctx sessionctx.Context, node ast.Node, pre
 	return errors.Trace(v.err)
 }
 
-type preprocessorFlag uint8
+type preprocessorFlag uint64
 
 const (
 	// inPrepare is set when visiting in prepare statement.
@@ -157,6 +157,8 @@ const (
 	inSequenceFunction
 	// initTxnContextProvider is set when we should init txn context in preprocess
 	initTxnContextProvider
+	// inAnalyze is set when visiting an analyze statement.
+	inAnalyze
 )
 
 // Make linter happy.
@@ -393,6 +395,8 @@ func (p *preprocessor) Enter(in ast.Node) (out ast.Node, skipChildren bool) {
 			p.sctx.GetSessionVars().StmtCtx.IsStaleness = true
 			p.IsStaleness = true
 		}
+	case *ast.AnalyzeTableStmt:
+		p.flag |= inAnalyze
 	default:
 		p.flag &= ^parentIsJoin
 	}
@@ -1580,10 +1584,12 @@ func (p *preprocessor) handleTableName(tn *ast.TableName) {
 	if tn.Schema.String() != "" {
 		currentDB = tn.Schema.L
 	}
-	table, err = tryLockMDLAndUpdateSchemaIfNecessary(p.sctx, model.NewCIStr(currentDB), table, p.ensureInfoSchema())
-	if err != nil {
-		p.err = err
-		return
+	if !p.skipLockMDL() {
+		table, err = tryLockMDLAndUpdateSchemaIfNecessary(p.sctx, model.NewCIStr(currentDB), table, p.ensureInfoSchema())
+		if err != nil {
+			p.err = err
+			return
+		}
 	}
 
 	tableInfo := table.Meta()
@@ -1913,3 +1919,11 @@ func tryLockMDLAndUpdateSchemaIfNecessary(sctx sessionctx.Context, dbName model.
 	}
 	return tbl, nil
 }
+
+// skipLockMDL returns true if the preprocessor should skip the lock of MDL.
+func (p *preprocessor) skipLockMDL() bool {
+	// skip lock mdl for IMPORT INTO statement,
+	// because it's a batch process and will do both DML and DDL.
+	// skip lock mdl for ANALYZE statement.
+	return p.flag&inAnalyze > 0
+}