diff --git a/executor/errors.go b/executor/errors.go
index 0e6ad100c1d4d..b7f8ce2ebf19f 100644
--- a/executor/errors.go
+++ b/executor/errors.go
@@ -34,7 +34,7 @@ var (
 	ErrGetStartTS      = terror.ClassExecutor.New(codeGetStartTS, "Can not get start ts")
 	ErrUnknownPlan     = terror.ClassExecutor.New(codeUnknownPlan, "Unknown plan")
 	ErrPrepareMulti    = terror.ClassExecutor.New(codePrepareMulti, "Can not prepare multiple statements")
-	ErrPrepareDDL      = terror.ClassExecutor.New(codePrepareDDL, "Can not prepare DDL statements")
+	ErrPrepareDDL      = terror.ClassExecutor.New(codePrepareDDL, "Can not prepare DDL statements with parameters")
 	ErrResultIsEmpty   = terror.ClassExecutor.New(codeResultIsEmpty, "result is empty")
 	ErrBuildExecutor   = terror.ClassExecutor.New(codeErrBuildExec, "Failed to build executor")
 	ErrBatchInsertFail = terror.ClassExecutor.New(codeBatchInsertFail, "Batch insert failed, please clean the table and try again.")
diff --git a/executor/prepared.go b/executor/prepared.go
index ecbcaede4753e..bfdb1cc6611f3 100644
--- a/executor/prepared.go
+++ b/executor/prepared.go
@@ -133,9 +133,6 @@ func (e *PrepareExec) Next(ctx context.Context, req *chunk.RecordBatch) error {
 		return ErrPrepareMulti
 	}
 	stmt := stmts[0]
-	if _, ok := stmt.(ast.DDLNode); ok {
-		return ErrPrepareDDL
-	}
 	err = ResetContextOfStmt(e.ctx, stmt)
 	if err != nil {
 		return err
@@ -143,6 +140,11 @@ func (e *PrepareExec) Next(ctx context.Context, req *chunk.RecordBatch) error {
 	var extractor paramMarkerExtractor
 	stmt.Accept(&extractor)
 
+	// DDL Statements can not accept parameters
+	if _, ok := stmt.(ast.DDLNode); ok && len(extractor.markers) > 0 {
+		return ErrPrepareDDL
+	}
+
 	// Prepare parameters should NOT over 2 bytes(MaxUint16)
 	// https://dev.mysql.com/doc/internals/en/com-stmt-prepare-response.html#packet-COM_STMT_PREPARE_OK.
 	if len(extractor.markers) > math.MaxUint16 {
diff --git a/executor/prepared_test.go b/executor/prepared_test.go
index 1eb64770f412f..82286e5c60e9a 100644
--- a/executor/prepared_test.go
+++ b/executor/prepared_test.go
@@ -30,3 +30,11 @@ func (s *testSuite1) TestPreparedNameResolver(c *C) {
 	_, err = tk.Exec("prepare stmt from '(select * FROM t) union all (select * FROM t) order by a limit ?'")
 	c.Assert(err.Error(), Equals, "[planner:1054]Unknown column 'a' in 'order clause'")
 }
+
+// a 'create table' DDL statement should be accepted if it has no parameters.
+func (s *testSuite1) TestPreparedDDL(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("prepare stmt from 'create table t (id int, KEY id (id))'")
+}