From 83450d774df9128903ec131e12318efacd4b81e4 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 00:38:15 +0800
Subject: [PATCH 01/19] resolve conflicts code

---
 ast/ddl.go                  |   3 +-
 ddl/ddl.go                  |   4 +
 ddl/ddl_api.go              |  39 ++++++++-
 ddl/ddl_db_test.go          | 165 +++++++++++++++++++++++++++---------
 ddl/ddl_worker.go           |   4 +-
 ddl/delete_range.go         |   5 ++
 ddl/partition.go            |  73 +++++++++++++++-
 model/ddl.go                |   2 +
 model/model.go              |   2 +-
 parser/parser.y             |  11 ++-
 parser/parser_test.go       |   3 +
 plan/logical_plan_test.go   |  10 +--
 table/tables/tables_test.go |   2 +-
 13 files changed, 269 insertions(+), 54 deletions(-)

diff --git a/ast/ddl.go b/ast/ddl.go
index 21668716708b9..340027e881c6f 100644
--- a/ast/ddl.go
+++ b/ast/ddl.go
@@ -741,6 +741,7 @@ const (
 	AlterTableRenameIndex
 	AlterTableForce
 	AlterTableAddPartitions
+	AlterTableDropPartition
 
 // TODO: Add more actions
 )
@@ -877,7 +878,7 @@ func (n *TruncateTableStmt) Accept(v Visitor) (Node, bool) {
 
 // PartitionDefinition defines a single partition.
 type PartitionDefinition struct {
-	Name     string
+	Name     model.CIStr
 	LessThan []ExprNode
 	MaxValue bool
 	Comment  string
diff --git a/ddl/ddl.go b/ddl/ddl.go
index 09230bf72771b..8c2fb80c81260 100644
--- a/ddl/ddl.go
+++ b/ddl/ddl.go
@@ -170,6 +170,8 @@ var (
 	ErrPartitionMaxvalue = terror.ClassDDL.New(codePartitionMaxvalue, "MAXVALUE can only be used in last partition definition")
 	// ErrTooManyValues returns cannot have more than one value for this type of partitioning.
 	ErrTooManyValues = terror.ClassDDL.New(codeErrTooManyValues, mysql.MySQLErrName[mysql.ErrTooManyValues])
+	//ErrDropLastPartition returns cannot remove all partitions, use drop table instead.
+	ErrDropLastPartition = terror.ClassDDL.New(codeDropLastPartition, mysql.MySQLErrName[mysql.ErrDropLastPartition])
 )
 
 // DDL is responsible for updating schema in data store and maintaining in-memory InfoSchema cache.
@@ -574,6 +576,7 @@ const (
 	codeRangeNotIncreasing            = terror.ErrCode(mysql.ErrRangeNotIncreasing)
 	codePartitionMaxvalue             = terror.ErrCode(mysql.ErrPartitionMaxvalue)
 	codeErrTooManyValues              = terror.ErrCode(mysql.ErrTooManyValues)
+	codeDropLastPartition             = terror.ErrCode(mysql.ErrDropLastPartition)
 )
 
 func init() {
@@ -613,6 +616,7 @@ func init() {
 		codeRangeNotIncreasing:            mysql.ErrRangeNotIncreasing,
 		codePartitionMaxvalue:             mysql.ErrPartitionMaxvalue,
 		codeErrTooManyValues:              mysql.ErrTooManyValues,
+		codeDropLastPartition:             mysql.ErrDropLastPartition,
 	}
 	terror.ErrClassToMySQLCodes[terror.ClassDDL] = ddlMySQLErrCodes
 }
diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go
index 0e73d664e1d36..0c2f8f91489f1 100644
--- a/ddl/ddl_api.go
+++ b/ddl/ddl_api.go
@@ -984,6 +984,8 @@ func (d *ddl) AlterTable(ctx sessionctx.Context, ident ast.Ident, specs []*ast.A
 			err = d.DropColumn(ctx, ident, spec.OldColumnName.Name)
 		case ast.AlterTableDropIndex:
 			err = d.DropIndex(ctx, ident, model.NewCIStr(spec.Name))
+		case ast.AlterTableDropPartition:
+			err = d.DropTablePartition(ctx, ident, spec)
 		case ast.AlterTableAddConstraint:
 			constr := spec.Constraint
 			switch spec.Constraint.Tp {
@@ -1217,7 +1219,7 @@ func (d *ddl) AddTablePartitions(ctx sessionctx.Context, ident ast.Ident, spec *
 	}
 
 	meta := t.Meta()
-	if meta.GetPartitionInfo() == nil && meta.Partition == nil {
+	if meta.GetPartitionInfo() == nil {
 		return errors.Trace(ErrPartitionMgmtOnNonpartitioned)
 	}
 	partInfo, err := buildPartitionInfo(meta, d, spec)
@@ -1248,6 +1250,41 @@ func (d *ddl) AddTablePartitions(ctx sessionctx.Context, ident ast.Ident, spec *
 	return errors.Trace(err)
 }
 
+func (d *ddl) DropTablePartition(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) error {
+	is := d.infoHandle.Get()
+	schema, ok := is.SchemaByName(ident.Schema)
+	if !ok {
+		return errors.Trace(infoschema.ErrDatabaseNotExists.GenByArgs(schema))
+	}
+	t, err := is.TableByName(ident.Schema, ident.Name)
+	if err != nil {
+		return errors.Trace(infoschema.ErrTableNotExists.GenByArgs(ident.Schema, ident.Name))
+	}
+	meta := t.Meta()
+	if meta.GetPartitionInfo() == nil {
+		return errors.Trace(ErrPartitionMgmtOnNonpartitioned)
+	}
+	err = checkDropTablePartition(meta, spec.Name)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	job := &model.Job{
+		SchemaID:   schema.ID,
+		TableID:    meta.ID,
+		Type:       model.ActionDropTablePartition,
+		BinlogInfo: &model.HistoryInfo{},
+		Args:       []interface{}{spec.Name},
+	}
+
+	err = d.doDDLJob(ctx, job)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	err = d.callHookOnChanged(err)
+	return errors.Trace(err)
+}
+
 // DropColumn will drop a column from the table, now we don't support drop the column with index covered.
 func (d *ddl) DropColumn(ctx sessionctx.Context, ti ast.Ident, colName model.CIStr) error {
 	is := d.GetInformationSchema(ctx)
diff --git a/ddl/ddl_db_test.go b/ddl/ddl_db_test.go
index 59f3166c02060..54154be1f294e 100644
--- a/ddl/ddl_db_test.go
+++ b/ddl/ddl_db_test.go
@@ -1575,11 +1575,11 @@ func (s *testDBSuite) TestCreateTableWithPartition(c *C) {
 	}
 	c.Assert(part.Definitions, HasLen, 3)
 	c.Assert(part.Definitions[0].LessThan[0], Equals, "10")
-	c.Assert(part.Definitions[0].Name, Equals, "p0")
+	c.Assert(part.Definitions[0].Name.L, Equals, "p0")
 	c.Assert(part.Definitions[1].LessThan[0], Equals, "20")
-	c.Assert(part.Definitions[1].Name, Equals, "p1")
+	c.Assert(part.Definitions[1].Name.L, Equals, "p1")
 	c.Assert(part.Definitions[2].LessThan[0], Equals, "MAXVALUE")
-	c.Assert(part.Definitions[2].Name, Equals, "p2")
+	c.Assert(part.Definitions[2].Name.L, Equals, "p2")
 
 	s.tk.MustExec("drop table if exists employees;")
 	sql1 := `create table employees (
@@ -2440,56 +2440,58 @@ func (s *testDBSuite) TestBackwardCompatibility(c *C) {
 }
 
 func (s *testDBSuite) TestAlterTableAddPartition(c *C) {
-	s.tk.MustExec("use test")
-	s.tk.MustExec("drop table if employees")
+	s.tk = testkit.NewTestKit(c, s.store)
+	s.tk.MustExec("use test;")
+	s.tk.MustExec("set @@session.tidb_enable_table_partition=1")
+	s.tk.MustExec("drop table if exists employees;")
 	s.tk.MustExec(`create table employees (
 	id int not null,
-	hired date not null
+	hired int not null
 	)
-	partition by range( year(hired) ) (
+	partition by range( id ) (
 		partition p1 values less than (1991),
 		partition p2 values less than (1996),
 		partition p3 values less than (2001)
 	);`)
 	s.tk.MustExec(`alter table employees add partition (
     partition p4 values less than (2010),
-    partition p5 values less than maxvalue
+    partition p5 values less than MAXVALUE
 	);`)
 
 	ctx := s.tk.Se.(sessionctx.Context)
 	is := domain.GetDomain(ctx).InfoSchema()
-	tbl, err := is.TableByName(model.NewCIStr("test"), model.NewCIStr("tp"))
+	tbl, err := is.TableByName(model.NewCIStr("test"), model.NewCIStr("employees"))
 	c.Assert(err, IsNil)
 	c.Assert(tbl.Meta().Partition, NotNil)
 	part := tbl.Meta().Partition
 	c.Assert(part.Type, Equals, model.PartitionTypeRange)
-	c.Assert(part.Expr, Equals, "`hired`")
+	c.Assert(part.Expr, Equals, "`id`")
 
 	c.Assert(part.Definitions, HasLen, 5)
 	c.Assert(part.Definitions[0].LessThan[0], Equals, "1991")
-	c.Assert(part.Definitions[0].Name, Equals, "p1")
+	c.Assert(part.Definitions[0].Name, Equals, model.NewCIStr("p1"))
 	c.Assert(part.Definitions[1].LessThan[0], Equals, "1996")
-	c.Assert(part.Definitions[1].Name, Equals, "p2")
+	c.Assert(part.Definitions[1].Name, Equals, model.NewCIStr("p2"))
 	c.Assert(part.Definitions[2].LessThan[0], Equals, "2001")
-	c.Assert(part.Definitions[2].Name, Equals, "p3")
+	c.Assert(part.Definitions[2].Name, Equals, model.NewCIStr("p3"))
 	c.Assert(part.Definitions[3].LessThan[0], Equals, "2010")
-	c.Assert(part.Definitions[3].Name, Equals, "p4")
-	c.Assert(part.Definitions[4].LessThan[0], Equals, "maxvalue")
-	c.Assert(part.Definitions[4].Name, Equals, "p5")
+	c.Assert(part.Definitions[3].Name, Equals, model.NewCIStr("p4"))
+	c.Assert(part.Definitions[4].LessThan[0], Equals, "MAXVALUE")
+	c.Assert(part.Definitions[4].Name, Equals, model.NewCIStr("p5"))
 
-	s.tk.MustExec("drop table if t1")
-	s.tk.MustExec("create table t1(a int)")
-	sql1 := `alter table t1 add partition (
+	s.tk.MustExec("drop table if exists table1;")
+	s.tk.MustExec("create table table1(a int)")
+	sql1 := `alter table table1 add partition (
 		partition p1 values less than (2010),
 		partition p2 values less than maxvalue
 	);`
-	s.testErrorCode(c, sql1, mysql.ErrPartitionMgmtOnNonpartitioned)
+	s.testErrorCode(c, sql1, tmysql.ErrPartitionMgmtOnNonpartitioned)
 
-	sql2 := "alter table t1 add partition"
-	s.testErrorCode(c, sql2, mysql.ErrPartitionsMustBeDefined)
+	sql2 := "alter table table1 add partition"
+	s.testErrorCode(c, sql2, tmysql.ErrPartitionsMustBeDefined)
 
-	s.tk.MustExec("drop table if t2")
-	s.tk.MustExec(`create table t2 (
+	s.tk.MustExec("drop table if exists table2;")
+	s.tk.MustExec(`create table table2 (
 	id int not null,
 	hired date not null
 	)
@@ -2498,41 +2500,126 @@ func (s *testDBSuite) TestAlterTableAddPartition(c *C) {
 	partition p2 values less than maxvalue
 	);`)
 
-	sql3 := `alter table t2 add partition (
+	sql3 := `alter table table2 add partition (
 		partition p3 values less than (2010)
 	);`
-	s.testErrorCode(c, sql3, mysql.ErrPartitionMaxvalue)
+	s.testErrorCode(c, sql3, tmysql.ErrPartitionMaxvalue)
 
-	s.tk.MustExec("drop table if t3")
-	s.tk.MustExec(`create table t3 (
+	s.tk.MustExec("drop table if exists table3;")
+	s.tk.MustExec(`create table table3 (
 	id int not null,
 	hired date not null
 	)
 	partition by range( year(hired) ) (
 	partition p1 values less than (1991),
-	partition p3 values less than (2001)
+	partition p2 values less than (2001)
 	);`)
 
-	sql4 := `alter table t3 add partition (
+	sql4 := `alter table table3 add partition (
 		partition p3 values less than (1993)
 	);`
-	s.testErrorCode(c, sql4, mysql.ErrRangeNotIncreasing)
+	s.testErrorCode(c, sql4, tmysql.ErrRangeNotIncreasing)
 
-	sql5 := `alter table t3 add partition (
+	sql5 := `alter table table3 add partition (
 		partition p1 values less than (1993)
 	);`
-	s.testErrorCode(c, sql5, mysql.ErrSameNamePartition)
+	s.testErrorCode(c, sql5, tmysql.ErrSameNamePartition)
 
-	sql6 := `alter table t3 add partition (
+	sql6 := `alter table table3 add partition (
 		partition p1 values less than (1993),
 		partition p1 values less than (1995)
 	);`
-	s.testErrorCode(c, sql6, mysql.ErrSameNamePartition)
+	s.testErrorCode(c, sql6, tmysql.ErrSameNamePartition)
 
-	sql7 := `alter table t3 add partition (
+	sql7 := `alter table table3 add partition (
 		partition p4 values less than (1993),
-		partition p1 values less than (1995)，
-		partition p5 values less than maxvalue,
+		partition p1 values less than (1995),
+		partition p5 values less than maxvalue
 	);`
-	s.testErrorCode(c, sql7, mysql.ErrSameNamePartition)
+	s.testErrorCode(c, sql7, tmysql.ErrSameNamePartition)
+}
+
+func (s *testDBSuite) TestAlterTableDropPartition(c *C) {
+	s.tk = testkit.NewTestKit(c, s.store)
+	s.tk.MustExec("use test")
+	s.tk.MustExec("set @@session.tidb_enable_table_partition=1")
+	s.tk.MustExec("drop table if exists employees")
+	s.tk.MustExec(`create table employees (
+	id int not null,
+	hired int not null
+	)
+	partition by range( hired ) (
+		partition p1 values less than (1991),
+		partition p2 values less than (1996),
+		partition p3 values less than (2001)
+	);`)
+
+	s.tk.MustExec("alter table employees drop partition p3;")
+	ctx := s.tk.Se.(sessionctx.Context)
+	is := domain.GetDomain(ctx).InfoSchema()
+	tbl, err := is.TableByName(model.NewCIStr("test"), model.NewCIStr("employees"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().GetPartitionInfo(), NotNil)
+	part := tbl.Meta().Partition
+	c.Assert(part.Type, Equals, model.PartitionTypeRange)
+	c.Assert(part.Expr, Equals, "`hired`")
+
+	c.Assert(part.Definitions, HasLen, 2)
+	c.Assert(part.Definitions[0].LessThan[0], Equals, "1991")
+	c.Assert(part.Definitions[0].Name, Equals, model.NewCIStr("p1"))
+	c.Assert(part.Definitions[1].LessThan[0], Equals, "1996")
+	c.Assert(part.Definitions[1].Name, Equals, model.NewCIStr("p2"))
+
+	s.tk.MustExec("drop table if exists table1;")
+	s.tk.MustExec("create table table1 (a int);")
+	sql1 := "alter table table1 drop partition p10;"
+	s.testErrorCode(c, sql1, tmysql.ErrPartitionMgmtOnNonpartitioned)
+
+	s.tk.MustExec("drop table if exists table2;")
+	s.tk.MustExec(`create table table2 (
+	id int not null,
+	hired date not null
+	)
+	partition by range( year(hired) ) (
+		partition p1 values less than (1991),
+		partition p2 values less than (1996),
+		partition p3 values less than (2001)
+	);`)
+	sql2 := "alter table table2 drop partition p10;"
+	s.testErrorCode(c, sql2, tmysql.ErrDropPartitionNonExistent)
+
+	s.tk.MustExec("drop table if exists table3;")
+	s.tk.MustExec(`create table table3 (
+	id int not null
+	)
+	partition by range( id ) (
+		partition p1 values less than (1991)
+	);`)
+	sql3 := "alter table table3 drop partition p1;"
+	s.testErrorCode(c, sql3, tmysql.ErrDropLastPartition)
+
+	s.tk.MustExec("drop table if exists table4;")
+	s.tk.MustExec(`create table table4 (
+	id int not null
+	)
+	partition by range( id ) (
+		partition p1 values less than (10),
+		partition p2 values less than (20),
+		partition p3 values less than MAXVALUE
+	);`)
+
+	s.tk.MustExec("alter table table4 drop partition p2;")
+	is = domain.GetDomain(ctx).InfoSchema()
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("table4"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().GetPartitionInfo(), NotNil)
+	part = tbl.Meta().Partition
+	c.Assert(part.Type, Equals, model.PartitionTypeRange)
+	c.Assert(part.Expr, Equals, "`id`")
+
+	c.Assert(part.Definitions, HasLen, 2)
+	c.Assert(part.Definitions[0].LessThan[0], Equals, "10")
+	c.Assert(part.Definitions[0].Name, Equals, model.NewCIStr("p1"))
+	c.Assert(part.Definitions[1].LessThan[0], Equals, "MAXVALUE")
+	c.Assert(part.Definitions[1].Name, Equals, model.NewCIStr("p3"))
 }
diff --git a/ddl/ddl_worker.go b/ddl/ddl_worker.go
index 13ee77df39711..51fd0537f5dd7 100644
--- a/ddl/ddl_worker.go
+++ b/ddl/ddl_worker.go
@@ -264,7 +264,7 @@ func (w *worker) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
 		}
 		// After rolling back an AddIndex operation, we need to use delete-range to delete the half-done index data.
 		err = w.deleteRange(job)
-	case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex:
+	case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex, model.ActionDropTablePartition:
 		err = w.deleteRange(job)
 	}
 	if err != nil {
@@ -419,6 +419,8 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 		ver, err = onCreateTable(d, t, job)
 	case model.ActionDropTable:
 		ver, err = onDropTable(t, job)
+	case model.ActionDropTablePartition:
+		ver, err = onDropTablePartition(t, job)
 	case model.ActionAddColumn:
 		ver, err = onAddColumn(d, t, job)
 	case model.ActionDropColumn:
diff --git a/ddl/delete_range.go b/ddl/delete_range.go
index b451509875c01..9211545031943 100644
--- a/ddl/delete_range.go
+++ b/ddl/delete_range.go
@@ -241,6 +241,11 @@ func insertJobIntoDeleteRangeTable(ctx sessionctx.Context, job *model.Job) error
 		startKey := tablecodec.EncodeTablePrefix(tableID)
 		endKey := tablecodec.EncodeTablePrefix(tableID + 1)
 		return doInsert(s, job.ID, tableID, startKey, endKey, now)
+	case model.ActionDropTablePartition:
+		paritionID := job.TableID
+		startKey := tablecodec.EncodeTablePrefix(paritionID)
+		endKey := tablecodec.EncodeTablePrefix(paritionID + 1)
+		return doInsert(s, job.ID, paritionID, startKey, endKey, now)
 	// ActionAddIndex needs do it, because it needs to be rolled back when it's canceled.
 	case model.ActionAddIndex:
 		tableID := job.TableID
diff --git a/ddl/partition.go b/ddl/partition.go
index 571a10df3f15b..1393eeb95d597 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/juju/errors"
 	"github.com/pingcap/tidb/ast"
+	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/model"
 	"github.com/pingcap/tidb/mysql"
 	"github.com/pingcap/tidb/sessionctx"
@@ -98,15 +99,15 @@ func checkPartitionNameUnique(tbInfo *model.TableInfo, pi *model.PartitionInfo)
 	if tbInfo.Partition != nil {
 		oldPars := tbInfo.Partition.Definitions
 		for _, oldPar := range oldPars {
-			partNames[strings.ToLower(oldPar.Name)] = struct{}{}
+			partNames[oldPar.Name.L] = struct{}{}
 		}
 	}
 	newPars := pi.Definitions
 	for _, newPar := range newPars {
-		if _, ok := partNames[strings.ToLower(newPar.Name)]; ok {
+		if _, ok := partNames[newPar.Name.L]; ok {
 			return ErrSameNamePartition.GenByArgs(newPar.Name)
 		}
-		partNames[strings.ToLower(newPar.Name)] = struct{}{}
+		partNames[newPar.Name.L] = struct{}{}
 	}
 	return nil
 }
@@ -154,3 +155,69 @@ func validRangePartitionType(col *table.Column) bool {
 		return false
 	}
 }
+
+// checkDropTablePartition checks the partition exists and is not the only partition in the table.
+func checkDropTablePartition(meta *model.TableInfo, partName string) error {
+	oldDefs := meta.Partition.Definitions
+	for _, def := range oldDefs {
+		if strings.EqualFold(def.Name.O, partName) {
+			if len(oldDefs) == 1 {
+				return errors.Trace(ErrDropLastPartition)
+			}
+			return nil
+		}
+	}
+	return errors.Trace(ErrDropPartitionNonExistent.GenByArgs(partName))
+}
+
+func removePartitionInfo(job *model.Job, t *meta.Meta, tblInfo *model.TableInfo, partName string) error {
+	newDefs, oldDefs := []model.PartitionDefinition{}, tblInfo.Partition.Definitions
+	newDefs = make([]model.PartitionDefinition, 0, len(oldDefs)-1)
+	var pid int64
+	for i := 0; i < len(oldDefs); i++ {
+		if !strings.EqualFold(oldDefs[i].Name.O, partName) {
+			continue
+		}
+		pid = oldDefs[i].ID
+		j := i + 1
+		newDefs = append(oldDefs[:i], oldDefs[j:]...)
+		break
+	}
+	job.TableID = pid
+	tblInfo.Partition.Definitions = newDefs
+	return nil
+}
+
+func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
+	var partName string
+	if err := job.DecodeArgs(&partName); err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+	tblInfo, err := getTableInfo(t, job, job.SchemaID)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	err = checkDropTablePartition(tblInfo, partName)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	err = removePartitionInfo(job, t, tblInfo, partName)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
+	ver, err = updateVersionAndTableInfo(t, job, tblInfo, true)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
+	// Finish this job.
+	job.FinishTableJob(model.JobStateDone, model.StateNone, ver, tblInfo)
+
+	return ver, nil
+}
diff --git a/model/ddl.go b/model/ddl.go
index 8e273318af09c..596606fa6b7b9 100644
--- a/model/ddl.go
+++ b/model/ddl.go
@@ -49,6 +49,7 @@ const (
 	ActionModifyTableComment ActionType = 17
 	ActionRenameIndex        ActionType = 18
 	ActionAddTablePartition  ActionType = 19
+	ActionDropTablePartition ActionType = 20
 )
 
 var actionMap = map[ActionType]string{
@@ -71,6 +72,7 @@ var actionMap = map[ActionType]string{
 	ActionModifyTableComment: "modify table comment",
 	ActionRenameIndex:        "rename index",
 	ActionAddTablePartition:  "add partition",
+	ActionDropTablePartition: "drop table partition",
 }
 
 // String return current ddl action in string
diff --git a/model/model.go b/model/model.go
index d7d9c4f2c9f4b..49ea950705d9a 100644
--- a/model/model.go
+++ b/model/model.go
@@ -285,7 +285,7 @@ type PartitionInfo struct {
 // PartitionDefinition defines a single partition.
 type PartitionDefinition struct {
 	ID       int64    `json:"id"`
-	Name     string   `json:"name"`
+	Name     CIStr    `json:"name"`
 	LessThan []string `json:"less_than"`
 	Comment  string   `json:"comment,omitempty"`
 }
diff --git a/parser/parser.y b/parser/parser.y
index f265283cbe9c2..09635031f318f 100644
--- a/parser/parser.y
+++ b/parser/parser.y
@@ -979,6 +979,13 @@ AlterTableSpec:
 	{
 		$$ = &ast.AlterTableSpec{Tp: ast.AlterTableDropPrimaryKey}
 	}
+|	"DROP" "PARTITION" Identifier
+	{
+		$$ = &ast.AlterTableSpec{
+			Tp: ast.AlterTableDropPartition,
+			Name: $3,
+		}
+	}	
 |	"DROP" KeyOrIndex Identifier
 	{
 		$$ = &ast.AlterTableSpec{
@@ -1841,8 +1848,8 @@ PartitionDefinition:
 	"PARTITION" Identifier PartDefValuesOpt PartDefCommentOpt PartDefStorageOpt
 	{
 		partDef := &ast.PartitionDefinition{
-			Name:		$2,
-			Comment:	$4.(string),
+			Name: model.NewCIStr($2),
+			Comment: $4.(string),
 		}
 		switch $3.(type) {
 		case []ast.ExprNode:
diff --git a/parser/parser_test.go b/parser/parser_test.go
index 2d07df6a91dde..06d953c2ec55d 100644
--- a/parser/parser_test.go
+++ b/parser/parser_test.go
@@ -1611,6 +1611,9 @@ func (s *testParserSuite) TestDDL(c *C) {
 				PARTITION P1 VALUES LESS THAN (2010),
 				PARTITION P2 VALUES LESS THAN (2015),
 				PARTITION P3 VALUES LESS THAN MAXVALUE)`, true},
+		// For drop table partition statement.
+		{"alter table t drop partition p1;", true},
+		{"alter table t drop partition p2;", true},
 		{"ALTER TABLE t DISABLE KEYS", true},
 		{"ALTER TABLE t ENABLE KEYS", true},
 		{"ALTER TABLE t MODIFY COLUMN a varchar(255)", true},
diff --git a/plan/logical_plan_test.go b/plan/logical_plan_test.go
index 5213f6bb92e89..6a0f0133f79bc 100644
--- a/plan/logical_plan_test.go
+++ b/plan/logical_plan_test.go
@@ -467,27 +467,27 @@ func (s *testPlanSuite) TestTablePartition(c *C) {
 	definitions := []model.PartitionDefinition{
 		{
 			ID:       41,
-			Name:     "p1",
+			Name:     model.NewCIStr("p1"),
 			LessThan: []string{"16"},
 		},
 		{
 			ID:       42,
-			Name:     "p2",
+			Name:     model.NewCIStr("p2"),
 			LessThan: []string{"32"},
 		},
 		{
 			ID:       43,
-			Name:     "p3",
+			Name:     model.NewCIStr("p3"),
 			LessThan: []string{"64"},
 		},
 		{
 			ID:       44,
-			Name:     "p4",
+			Name:     model.NewCIStr("p4"),
 			LessThan: []string{"128"},
 		},
 		{
 			ID:       45,
-			Name:     "p5",
+			Name:     model.NewCIStr("p5"),
 			LessThan: []string{"maxvalue"},
 		},
 	}
diff --git a/table/tables/tables_test.go b/table/tables/tables_test.go
index 5eb81fd25d1cf..37ff75356e91d 100644
--- a/table/tables/tables_test.go
+++ b/table/tables/tables_test.go
@@ -345,7 +345,7 @@ PARTITION BY RANGE ( id ) (
 	tb, err := ts.dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
 	tbInfo := tb.Meta()
 	p0 := tbInfo.Partition.Definitions[0]
-	c.Assert(p0.Name, Equals, "p0")
+	c.Assert(p0.Name, Equals, model.NewCIStr("p0"))
 
 	c.Assert(ts.se.NewTxn(), IsNil)
 	rid, err := tb.AddRecord(ts.se, types.MakeDatums(1), false)

From 1358251dcc6a671281dec013f8f7575c64f4f69d Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 11:35:43 +0800
Subject: [PATCH 02/19] Address comments

---
 ddl/partition.go | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/ddl/partition.go b/ddl/partition.go
index 1393eeb95d597..325bd83c22037 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -156,7 +156,7 @@ func validRangePartitionType(col *table.Column) bool {
 	}
 }
 
-// checkDropTablePartition checks the partition exists and is not the only partition in the table.
+// checkDropTablePartition checks if the partition exists and does not allow deleting the last existing partition in the table.
 func checkDropTablePartition(meta *model.TableInfo, partName string) error {
 	oldDefs := meta.Partition.Definitions
 	for _, def := range oldDefs {
@@ -171,16 +171,15 @@ func checkDropTablePartition(meta *model.TableInfo, partName string) error {
 }
 
 func removePartitionInfo(job *model.Job, t *meta.Meta, tblInfo *model.TableInfo, partName string) error {
-	newDefs, oldDefs := []model.PartitionDefinition{}, tblInfo.Partition.Definitions
-	newDefs = make([]model.PartitionDefinition, 0, len(oldDefs)-1)
+	oldDefs := tblInfo.Partition.Definitions
+	newDefs := make([]model.PartitionDefinition, 0, len(oldDefs)-1)
 	var pid int64
 	for i := 0; i < len(oldDefs); i++ {
 		if !strings.EqualFold(oldDefs[i].Name.O, partName) {
 			continue
 		}
 		pid = oldDefs[i].ID
-		j := i + 1
-		newDefs = append(oldDefs[:i], oldDefs[j:]...)
+		newDefs = append(oldDefs[:i], oldDefs[i+1:]...)
 		break
 	}
 	job.TableID = pid

From 41a420f44d655e0703df6f77775d26a49496d212 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 13:29:57 +0800
Subject: [PATCH 03/19] add alterTableDropPartition tests

---
 ddl/ddl_db_test.go | 40 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 40 insertions(+)

diff --git a/ddl/ddl_db_test.go b/ddl/ddl_db_test.go
index 54154be1f294e..d4dbf30844dae 100644
--- a/ddl/ddl_db_test.go
+++ b/ddl/ddl_db_test.go
@@ -2622,4 +2622,44 @@ func (s *testDBSuite) TestAlterTableDropPartition(c *C) {
 	c.Assert(part.Definitions[0].Name, Equals, model.NewCIStr("p1"))
 	c.Assert(part.Definitions[1].LessThan[0], Equals, "MAXVALUE")
 	c.Assert(part.Definitions[1].Name, Equals, model.NewCIStr("p3"))
+
+	s.tk.MustExec("drop table if exists table4;")
+	s.tk.MustExec(` create table tr(
+		id int, name varchar(50), 
+		purchased date
+	)
+	partition by range( year(purchased) ) (
+    	partition p0 values less than (1990),
+    	partition p1 values less than (1995),
+    	partition p2 values less than (2000),
+    	partition p3 values less than (2005),
+    	partition p4 values less than (2010),
+    	partition p5 values less than (2015)
+   	);`)
+	s.tk.MustExec(` INSERT INTO tr VALUES
+   		(1, 'desk organiser', '2003-10-15'),
+   		(2, 'alarm clock', '1997-11-05'),
+   		(3, 'chair', '2009-03-10'),
+   		(4, 'bookcase', '1989-01-10'),
+    	(5, 'exercise bike', '2014-05-09'),
+    	(6, 'sofa', '1987-06-05'),
+    	(7, 'espresso maker', '2011-11-22'),
+    	(8, 'aquarium', '1992-08-04'),
+    	(9, 'study desk', '2006-09-16'),
+    	(10, 'lava lamp', '1998-12-25');`)
+	result := s.tk.MustQuery("select * from tr where purchased between '1995-01-01' and '1999-12-31';")
+	result.Check(testkit.Rows(`2 alarm clock 1997-11-05`, `10 lava lamp 1998-12-25`))
+	s.tk.MustExec("alter table tr drop partition p2;")
+	result = s.tk.MustQuery("select * from tr where purchased between '1995-01-01' and '1999-12-31';")
+	result.Check(testkit.Rows())
+
+	result = s.tk.MustQuery("select * from tr where purchased between '2010-01-01' and '2014-12-31';")
+	result.Check(testkit.Rows(`5 exercise bike 2014-05-09`, `7 espresso maker 2011-11-22`))
+	s.tk.MustExec("alter table tr drop partition p5;")
+	result = s.tk.MustQuery("select * from tr where purchased between '2010-01-01' and '2014-12-31';")
+	result.Check(testkit.Rows())
+
+	s.tk.MustExec("alter table tr drop partition p4;")
+	result = s.tk.MustQuery("select * from tr where purchased between '2005-01-01' and '2009-12-31';")
+	result.Check(testkit.Rows())
 }

From a1cec0e06161927211dd42015f252984a21a169e Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 14:17:30 +0800
Subject: [PATCH 04/19] resolving code conflicts

---
 ddl/{ddl_db_test.go => db_test.go} | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)
 rename ddl/{ddl_db_test.go => db_test.go} (99%)

diff --git a/ddl/ddl_db_test.go b/ddl/db_test.go
similarity index 99%
rename from ddl/ddl_db_test.go
rename to ddl/db_test.go
index d4dbf30844dae..7e21bd35c1848 100644
--- a/ddl/ddl_db_test.go
+++ b/ddl/db_test.go
@@ -2623,7 +2623,7 @@ func (s *testDBSuite) TestAlterTableDropPartition(c *C) {
 	c.Assert(part.Definitions[1].LessThan[0], Equals, "MAXVALUE")
 	c.Assert(part.Definitions[1].Name, Equals, model.NewCIStr("p3"))
 
-	s.tk.MustExec("drop table if exists table4;")
+	s.tk.MustExec("drop table if exists tr;")
 	s.tk.MustExec(` create table tr(
 		id int, name varchar(50), 
 		purchased date

From d175ea590af472770c2a725b2c83d8bc73967b31 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 15:14:07 +0800
Subject: [PATCH 05/19] Update db_test.go

---
 ddl/db_test.go | 2 --
 1 file changed, 2 deletions(-)

diff --git a/ddl/db_test.go b/ddl/db_test.go
index 14e643ff9d7e6..fe3ca1a28b39c 100644
--- a/ddl/db_test.go
+++ b/ddl/db_test.go
@@ -90,7 +90,6 @@ func (s *testDBSuite) SetUpSuite(c *C) {
 	s.cluster = mocktikv.NewCluster()
 	mocktikv.BootstrapWithSingleStore(s.cluster)
 	s.mvccStore = mocktikv.MustNewMVCCStore()
-
 	s.store, err = mockstore.NewMockTikvStore(
 		mockstore.WithCluster(s.cluster),
 		mockstore.WithMVCCStore(s.mvccStore),
@@ -927,7 +926,6 @@ func (s *testDBSuite) TestAddColumnTooMany(c *C) {
 	s.tk.MustExec("use test")
 	count := ddl.TableColumnCountLimit - 1
 	var cols []string
-
 	for i := 0; i < count; i++ {
 		cols = append(cols, fmt.Sprintf("a%d int", i))
 	}

From 7edcb1b3a8c6702fcc06a1bf3a0df6fdc6b85cf9 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 15:17:21 +0800
Subject: [PATCH 06/19] Update db_test.go

---
 ddl/db_test.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/ddl/db_test.go b/ddl/db_test.go
index fe3ca1a28b39c..ab7e6da3cf146 100644
--- a/ddl/db_test.go
+++ b/ddl/db_test.go
@@ -2392,7 +2392,6 @@ func (s *testDBSuite) TestBackwardCompatibility(c *C) {
 }
 
 func (s *testDBSuite) TestAlterTableAddPartition(c *C) {
-
 	s.tk = testkit.NewTestKit(c, s.store)
 	s.tk.MustExec("use test;")
 	s.tk.MustExec("set @@session.tidb_enable_table_partition=1")

From f571a4982ceb78073ca6ed669ae6b86f7a7ed0b0 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 15:54:42 +0800
Subject: [PATCH 07/19] Address comments

---
 ddl/db_test.go | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/ddl/db_test.go b/ddl/db_test.go
index ab7e6da3cf146..0d942fd52a0a8 100644
--- a/ddl/db_test.go
+++ b/ddl/db_test.go
@@ -2398,9 +2398,9 @@ func (s *testDBSuite) TestAlterTableAddPartition(c *C) {
 	s.tk.MustExec("drop table if exists employees;")
 	s.tk.MustExec(`create table employees (
 	id int not null,
-	hired int not null
+	hired date not null
 	)
-	partition by range( id ) (
+	partition by range( year(hired) ) (
 		partition p1 values less than (1991),
 		partition p2 values less than (1996),
 		partition p3 values less than (2001)
@@ -2418,7 +2418,7 @@ func (s *testDBSuite) TestAlterTableAddPartition(c *C) {
 	part := tbl.Meta().Partition
 	c.Assert(part.Type, Equals, model.PartitionTypeRange)
 
-	c.Assert(part.Expr, Equals, "`id`")
+	c.Assert(part.Expr, Equals, "year(`hired`)")
 	c.Assert(part.Definitions, HasLen, 5)
 	c.Assert(part.Definitions[0].LessThan[0], Equals, "1991")
 	c.Assert(part.Definitions[0].Name, Equals, model.NewCIStr("p1"))

From 1b0f57ca37a678fb668e1040da049370c66f0b26 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 16:09:48 +0800
Subject: [PATCH 08/19] formatting code

---
 ddl/db_test.go | 22 +++++++++++-----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/ddl/db_test.go b/ddl/db_test.go
index 0d942fd52a0a8..b988086120df9 100644
--- a/ddl/db_test.go
+++ b/ddl/db_test.go
@@ -2589,17 +2589,17 @@ func (s *testDBSuite) TestAlterTableDropPartition(c *C) {
     	partition p4 values less than (2010),
     	partition p5 values less than (2015)
    	);`)
-	s.tk.MustExec(` INSERT INTO tr VALUES
-   		(1, 'desk organiser', '2003-10-15'),
-   		(2, 'alarm clock', '1997-11-05'),
-   		(3, 'chair', '2009-03-10'),
-   		(4, 'bookcase', '1989-01-10'),
-    	(5, 'exercise bike', '2014-05-09'),
-    	(6, 'sofa', '1987-06-05'),
-    	(7, 'espresso maker', '2011-11-22'),
-    	(8, 'aquarium', '1992-08-04'),
-    	(9, 'study desk', '2006-09-16'),
-    	(10, 'lava lamp', '1998-12-25');`)
+	s.tk.MustExec(`INSERT INTO tr VALUES
+	(1, 'desk organiser', '2003-10-15'),
+	(2, 'alarm clock', '1997-11-05'),
+	(3, 'chair', '2009-03-10'),
+	(4, 'bookcase', '1989-01-10'),
+	(5, 'exercise bike', '2014-05-09'),
+	(6, 'sofa', '1987-06-05'),
+	(7, 'espresso maker', '2011-11-22'),
+	(8, 'aquarium', '1992-08-04'),
+	(9, 'study desk', '2006-09-16'),
+	(10, 'lava lamp', '1998-12-25');`)
 	result := s.tk.MustQuery("select * from tr where purchased between '1995-01-01' and '1999-12-31';")
 	result.Check(testkit.Rows(`2 alarm clock 1997-11-05`, `10 lava lamp 1998-12-25`))
 	s.tk.MustExec("alter table tr drop partition p2;")

From 553faa3778ba658a05e69e823932e2850e4c5113 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 16:26:55 +0800
Subject: [PATCH 09/19] add comments

---
 ddl/delete_range.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/ddl/delete_range.go b/ddl/delete_range.go
index 9211545031943..13c1ef1a1c97b 100644
--- a/ddl/delete_range.go
+++ b/ddl/delete_range.go
@@ -242,6 +242,7 @@ func insertJobIntoDeleteRangeTable(ctx sessionctx.Context, job *model.Job) error
 		endKey := tablecodec.EncodeTablePrefix(tableID + 1)
 		return doInsert(s, job.ID, tableID, startKey, endKey, now)
 	case model.ActionDropTablePartition:
+		// the job.tableID here stores paritionID.
 		paritionID := job.TableID
 		startKey := tablecodec.EncodeTablePrefix(paritionID)
 		endKey := tablecodec.EncodeTablePrefix(paritionID + 1)

From b6f117de01b3045e88b9a808316c345046643498 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 16:38:18 +0800
Subject: [PATCH 10/19] Address comments

---
 ddl/delete_range.go | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/ddl/delete_range.go b/ddl/delete_range.go
index 13c1ef1a1c97b..fc4c20e842867 100644
--- a/ddl/delete_range.go
+++ b/ddl/delete_range.go
@@ -242,11 +242,11 @@ func insertJobIntoDeleteRangeTable(ctx sessionctx.Context, job *model.Job) error
 		endKey := tablecodec.EncodeTablePrefix(tableID + 1)
 		return doInsert(s, job.ID, tableID, startKey, endKey, now)
 	case model.ActionDropTablePartition:
-		// the job.tableID here stores paritionID.
-		paritionID := job.TableID
-		startKey := tablecodec.EncodeTablePrefix(paritionID)
-		endKey := tablecodec.EncodeTablePrefix(paritionID + 1)
-		return doInsert(s, job.ID, paritionID, startKey, endKey, now)
+		// the `job.TableID` here stores partitionID.
+		partitionID := job.TableID
+		startKey := tablecodec.EncodeTablePrefix(partitionID)
+		endKey := tablecodec.EncodeTablePrefix(partitionID + 1)
+		return doInsert(s, job.ID, partitionID, startKey, endKey, now)
 	// ActionAddIndex needs do it, because it needs to be rolled back when it's canceled.
 	case model.ActionAddIndex:
 		tableID := job.TableID

From e49308385163dfad2921ab2884adc11ac00ff195 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 17:40:03 +0800
Subject: [PATCH 11/19] Address comments

---
 ddl/delete_range.go |  2 +-
 ddl/partition.go    | 13 +++++--------
 2 files changed, 6 insertions(+), 9 deletions(-)

diff --git a/ddl/delete_range.go b/ddl/delete_range.go
index fc4c20e842867..4643d17d4688c 100644
--- a/ddl/delete_range.go
+++ b/ddl/delete_range.go
@@ -242,7 +242,7 @@ func insertJobIntoDeleteRangeTable(ctx sessionctx.Context, job *model.Job) error
 		endKey := tablecodec.EncodeTablePrefix(tableID + 1)
 		return doInsert(s, job.ID, tableID, startKey, endKey, now)
 	case model.ActionDropTablePartition:
-		// the `job.TableID` here stores partitionID.
+		// The `job.TableID` here stores partitionID.
 		partitionID := job.TableID
 		startKey := tablecodec.EncodeTablePrefix(partitionID)
 		endKey := tablecodec.EncodeTablePrefix(partitionID + 1)
diff --git a/ddl/partition.go b/ddl/partition.go
index 325bd83c22037..3714345f25b75 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -170,7 +170,7 @@ func checkDropTablePartition(meta *model.TableInfo, partName string) error {
 	return errors.Trace(ErrDropPartitionNonExistent.GenByArgs(partName))
 }
 
-func removePartitionInfo(job *model.Job, t *meta.Meta, tblInfo *model.TableInfo, partName string) error {
+func removePartitionInfo(job *model.Job, tblInfo *model.TableInfo, partName string) {
 	oldDefs := tblInfo.Partition.Definitions
 	newDefs := make([]model.PartitionDefinition, 0, len(oldDefs)-1)
 	var pid int64
@@ -184,9 +184,10 @@ func removePartitionInfo(job *model.Job, t *meta.Meta, tblInfo *model.TableInfo,
 	}
 	job.TableID = pid
 	tblInfo.Partition.Definitions = newDefs
-	return nil
 }
 
+// onDropTablePartition delete old table partition meta.
+// A background job will be created to delete old partition data.
 func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	var partName string
 	if err := job.DecodeArgs(&partName); err != nil {
@@ -198,17 +199,13 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
-
+	// If an error occurs , it returns that it cannot delete all partitions or that the partition doesn't exist.
 	err = checkDropTablePartition(tblInfo, partName)
 	if err != nil {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
-
-	err = removePartitionInfo(job, t, tblInfo, partName)
-	if err != nil {
-		return ver, errors.Trace(err)
-	}
+	removePartitionInfo(job, tblInfo, partName)
 
 	ver, err = updateVersionAndTableInfo(t, job, tblInfo, true)
 	if err != nil {

From 0708ecdc3d0d9600bb363cd313bd0a8f570a0c05 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 18:23:55 +0800
Subject: [PATCH 12/19] change function comments

---
 ddl/partition.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/ddl/partition.go b/ddl/partition.go
index 3714345f25b75..87e0290d0021e 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -186,7 +186,7 @@ func removePartitionInfo(job *model.Job, tblInfo *model.TableInfo, partName stri
 	tblInfo.Partition.Definitions = newDefs
 }
 
-// onDropTablePartition delete old table partition meta.
+// onDropTablePartition delete old partition meta.
 // A background job will be created to delete old partition data.
 func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	var partName string
@@ -199,7 +199,7 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
-	// If an error occurs , it returns that it cannot delete all partitions or that the partition doesn't exist.
+	// If an error occurs, it returns that it cannot delete all partitions or that the partition doesn't exist.
 	err = checkDropTablePartition(tblInfo, partName)
 	if err != nil {
 		job.State = model.JobStateCancelled

From 6e524b0bfc1015d66f805a68d1e3ae1d5e1231c6 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Tue, 10 Jul 2018 20:30:49 +0800
Subject: [PATCH 13/19] Address comments

---
 ddl/db_test.go   | 2 --
 ddl/partition.go | 3 +--
 2 files changed, 1 insertion(+), 4 deletions(-)

diff --git a/ddl/db_test.go b/ddl/db_test.go
index b988086120df9..590f2343335af 100644
--- a/ddl/db_test.go
+++ b/ddl/db_test.go
@@ -2516,7 +2516,6 @@ func (s *testDBSuite) TestAlterTableDropPartition(c *C) {
 	part := tbl.Meta().Partition
 	c.Assert(part.Type, Equals, model.PartitionTypeRange)
 	c.Assert(part.Expr, Equals, "`hired`")
-
 	c.Assert(part.Definitions, HasLen, 2)
 	c.Assert(part.Definitions[0].LessThan[0], Equals, "1991")
 	c.Assert(part.Definitions[0].Name, Equals, model.NewCIStr("p1"))
@@ -2569,7 +2568,6 @@ func (s *testDBSuite) TestAlterTableDropPartition(c *C) {
 	part = tbl.Meta().Partition
 	c.Assert(part.Type, Equals, model.PartitionTypeRange)
 	c.Assert(part.Expr, Equals, "`id`")
-
 	c.Assert(part.Definitions, HasLen, 2)
 	c.Assert(part.Definitions[0].LessThan[0], Equals, "10")
 	c.Assert(part.Definitions[0].Name, Equals, model.NewCIStr("p1"))
diff --git a/ddl/partition.go b/ddl/partition.go
index 87e0290d0021e..382569d0e79dc 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -186,7 +186,7 @@ func removePartitionInfo(job *model.Job, tblInfo *model.TableInfo, partName stri
 	tblInfo.Partition.Definitions = newDefs
 }
 
-// onDropTablePartition delete old partition meta.
+// onDropTablePartition deletes old partition meta.
 // A background job will be created to delete old partition data.
 func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	var partName string
@@ -196,7 +196,6 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	}
 	tblInfo, err := getTableInfo(t, job, job.SchemaID)
 	if err != nil {
-		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
 	// If an error occurs, it returns that it cannot delete all partitions or that the partition doesn't exist.

From 0c86c043f590ea9fe0925989270ea313facc60c3 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Wed, 11 Jul 2018 10:30:33 +0800
Subject: [PATCH 14/19] fix the updateTable problem

---
 ddl/partition.go | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/ddl/partition.go b/ddl/partition.go
index 382569d0e79dc..4c472942cf42d 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -170,7 +170,7 @@ func checkDropTablePartition(meta *model.TableInfo, partName string) error {
 	return errors.Trace(ErrDropPartitionNonExistent.GenByArgs(partName))
 }
 
-func removePartitionInfo(job *model.Job, tblInfo *model.TableInfo, partName string) {
+func removePartitionInfo(job *model.Job, tblInfo *model.TableInfo, partName string) int64 {
 	oldDefs := tblInfo.Partition.Definitions
 	newDefs := make([]model.PartitionDefinition, 0, len(oldDefs)-1)
 	var pid int64
@@ -182,12 +182,11 @@ func removePartitionInfo(job *model.Job, tblInfo *model.TableInfo, partName stri
 		newDefs = append(oldDefs[:i], oldDefs[i+1:]...)
 		break
 	}
-	job.TableID = pid
 	tblInfo.Partition.Definitions = newDefs
+	return pid
 }
 
 // onDropTablePartition deletes old partition meta.
-// A background job will be created to delete old partition data.
 func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	var partName string
 	if err := job.DecodeArgs(&partName); err != nil {
@@ -204,8 +203,7 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
-	removePartitionInfo(job, tblInfo, partName)
-
+	pid := removePartitionInfo(job, tblInfo, partName)
 	ver, err = updateVersionAndTableInfo(t, job, tblInfo, true)
 	if err != nil {
 		return ver, errors.Trace(err)
@@ -213,6 +211,7 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 
 	// Finish this job.
 	job.FinishTableJob(model.JobStateDone, model.StateNone, ver, tblInfo)
-
+	// A background job will be created to delete old partition data.
+	job.TableID = pid
 	return ver, nil
 }

From 63fc3156b021261ad5d51dd882b6dcef2820c604 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Wed, 11 Jul 2018 10:51:55 +0800
Subject: [PATCH 15/19] add removePartitionInfo comments

---
 ddl/partition.go | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/ddl/partition.go b/ddl/partition.go
index 4c472942cf42d..3690233dc27e6 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -170,7 +170,8 @@ func checkDropTablePartition(meta *model.TableInfo, partName string) error {
 	return errors.Trace(ErrDropPartitionNonExistent.GenByArgs(partName))
 }
 
-func removePartitionInfo(job *model.Job, tblInfo *model.TableInfo, partName string) int64 {
+//  removePartitionInfo each ddl job deletes a partition.
+func removePartitionInfo(tblInfo *model.TableInfo, partName string) int64 {
 	oldDefs := tblInfo.Partition.Definitions
 	newDefs := make([]model.PartitionDefinition, 0, len(oldDefs)-1)
 	var pid int64
@@ -203,7 +204,7 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
-	pid := removePartitionInfo(job, tblInfo, partName)
+	pid := removePartitionInfo(tblInfo, partName)
 	ver, err = updateVersionAndTableInfo(t, job, tblInfo, true)
 	if err != nil {
 		return ver, errors.Trace(err)

From 3e856abe1d4bb4ba1fa34358e150aeba14e5632f Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Wed, 11 Jul 2018 10:54:50 +0800
Subject: [PATCH 16/19] add function comments

---
 ddl/partition.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/ddl/partition.go b/ddl/partition.go
index 3690233dc27e6..08fb8c3a34666 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -170,7 +170,7 @@ func checkDropTablePartition(meta *model.TableInfo, partName string) error {
 	return errors.Trace(ErrDropPartitionNonExistent.GenByArgs(partName))
 }
 
-//  removePartitionInfo each ddl job deletes a partition.
+// removePartitionInfo each ddl job deletes a partition.
 func removePartitionInfo(tblInfo *model.TableInfo, partName string) int64 {
 	oldDefs := tblInfo.Partition.Definitions
 	newDefs := make([]model.PartitionDefinition, 0, len(oldDefs)-1)

From ef9063cd977a40e47c962f1eefb8641efe5be5d1 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Wed, 11 Jul 2018 16:18:26 +0800
Subject: [PATCH 17/19] Address comments

---
 ddl/delete_range.go | 7 +++++--
 ddl/partition.go    | 6 +++---
 2 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/ddl/delete_range.go b/ddl/delete_range.go
index 4643d17d4688c..4b50e8d8f654e 100644
--- a/ddl/delete_range.go
+++ b/ddl/delete_range.go
@@ -242,8 +242,11 @@ func insertJobIntoDeleteRangeTable(ctx sessionctx.Context, job *model.Job) error
 		endKey := tablecodec.EncodeTablePrefix(tableID + 1)
 		return doInsert(s, job.ID, tableID, startKey, endKey, now)
 	case model.ActionDropTablePartition:
-		// The `job.TableID` here stores partitionID.
-		partitionID := job.TableID
+		var partName interface{}
+		var partitionID int64
+		if err := job.DecodeArgs(&partName, &partitionID); err != nil {
+			return errors.Trace(err)
+		}
 		startKey := tablecodec.EncodeTablePrefix(partitionID)
 		endKey := tablecodec.EncodeTablePrefix(partitionID + 1)
 		return doInsert(s, job.ID, partitionID, startKey, endKey, now)
diff --git a/ddl/partition.go b/ddl/partition.go
index 08fb8c3a34666..e3c7fc4ae2289 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -160,7 +160,7 @@ func validRangePartitionType(col *table.Column) bool {
 func checkDropTablePartition(meta *model.TableInfo, partName string) error {
 	oldDefs := meta.Partition.Definitions
 	for _, def := range oldDefs {
-		if strings.EqualFold(def.Name.O, partName) {
+		if strings.EqualFold(def.Name.L, strings.ToLower(partName)) {
 			if len(oldDefs) == 1 {
 				return errors.Trace(ErrDropLastPartition)
 			}
@@ -204,7 +204,7 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
-	pid := removePartitionInfo(tblInfo, partName)
+	partitionID := removePartitionInfo(tblInfo, partName)
 	ver, err = updateVersionAndTableInfo(t, job, tblInfo, true)
 	if err != nil {
 		return ver, errors.Trace(err)
@@ -213,6 +213,6 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	// Finish this job.
 	job.FinishTableJob(model.JobStateDone, model.StateNone, ver, tblInfo)
 	// A background job will be created to delete old partition data.
-	job.TableID = pid
+	job.Args = append(job.Args, partitionID)
 	return ver, nil
 }

From cf97e7140eacc440dee9355f2009af75bfb7a296 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Wed, 11 Jul 2018 16:48:13 +0800
Subject: [PATCH 18/19] Address comments

---
 ddl/delete_range.go | 3 +--
 ddl/partition.go    | 2 +-
 2 files changed, 2 insertions(+), 3 deletions(-)

diff --git a/ddl/delete_range.go b/ddl/delete_range.go
index 4b50e8d8f654e..bd97635bf0530 100644
--- a/ddl/delete_range.go
+++ b/ddl/delete_range.go
@@ -242,9 +242,8 @@ func insertJobIntoDeleteRangeTable(ctx sessionctx.Context, job *model.Job) error
 		endKey := tablecodec.EncodeTablePrefix(tableID + 1)
 		return doInsert(s, job.ID, tableID, startKey, endKey, now)
 	case model.ActionDropTablePartition:
-		var partName interface{}
 		var partitionID int64
-		if err := job.DecodeArgs(&partName, &partitionID); err != nil {
+		if err := job.DecodeArgs(&partitionID); err != nil {
 			return errors.Trace(err)
 		}
 		startKey := tablecodec.EncodeTablePrefix(partitionID)
diff --git a/ddl/partition.go b/ddl/partition.go
index e3c7fc4ae2289..3a31bc7ba71c6 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -213,6 +213,6 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	// Finish this job.
 	job.FinishTableJob(model.JobStateDone, model.StateNone, ver, tblInfo)
 	// A background job will be created to delete old partition data.
-	job.Args = append(job.Args, partitionID)
+	job.Args = []interface{}{partitionID}
 	return ver, nil
 }

From 8ce934c25bb3051327a09ef53a6e6ed61c04b986 Mon Sep 17 00:00:00 2001
From: ciscoxll <xiaoliangliang@pingcap.com>
Date: Wed, 11 Jul 2018 17:11:45 +0800
Subject: [PATCH 19/19] add ErrDropPartitionNonExistent test

---
 ddl/db_test.go   | 15 +++++++++++++++
 ddl/partition.go |  2 +-
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/ddl/db_test.go b/ddl/db_test.go
index 590f2343335af..610f61ffdc5f0 100644
--- a/ddl/db_test.go
+++ b/ddl/db_test.go
@@ -2613,4 +2613,19 @@ func (s *testDBSuite) TestAlterTableDropPartition(c *C) {
 	s.tk.MustExec("alter table tr drop partition p4;")
 	result = s.tk.MustQuery("select * from tr where purchased between '2005-01-01' and '2009-12-31';")
 	result.Check(testkit.Rows())
+
+	s.tk.MustExec("drop table if exists table4;")
+	s.tk.MustExec(`create table table4 (
+		id int not null
+	)
+	partition by range( id ) (
+		partition Par1 values less than (1991),
+		partition pAR2 values less than (1992),
+		partition Par3 values less than (1995),
+		partition PaR5 values less than (1996)
+	);`)
+	s.tk.MustExec("alter table table4 drop partition Par2;")
+	s.tk.MustExec("alter table table4 drop partition PAR5;")
+	sql4 := "alter table table4 drop partition PAR0;"
+	s.testErrorCode(c, sql4, tmysql.ErrDropPartitionNonExistent)
 }
diff --git a/ddl/partition.go b/ddl/partition.go
index 3a31bc7ba71c6..5ca3ed31855a4 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -176,7 +176,7 @@ func removePartitionInfo(tblInfo *model.TableInfo, partName string) int64 {
 	newDefs := make([]model.PartitionDefinition, 0, len(oldDefs)-1)
 	var pid int64
 	for i := 0; i < len(oldDefs); i++ {
-		if !strings.EqualFold(oldDefs[i].Name.O, partName) {
+		if !strings.EqualFold(oldDefs[i].Name.L, strings.ToLower(partName)) {
 			continue
 		}
 		pid = oldDefs[i].ID