diff --git a/ddl/bg_worker.go b/ddl/bg_worker.go
index 281d467ebf7fb..8c73af6fdd3d1 100644
--- a/ddl/bg_worker.go
+++ b/ddl/bg_worker.go
@@ -34,7 +34,7 @@ func (d *ddl) handleBgJobQueue() error {
 	err := kv.RunInNewTxn(d.store, false, func(txn kv.Transaction) error {
 		t := meta.NewMeta(txn)
 		owner, err := d.checkOwner(t, bgJobFlag)
-		if terror.ErrorEqual(err, ErrNotOwner) {
+		if terror.ErrorEqual(err, errNotOwner) {
 			return nil
 		}
 		if err != nil {
@@ -85,8 +85,7 @@ func (d *ddl) runBgJob(t *meta.Meta, job *model.Job) {
 		err = d.delReorgTable(t, job)
 	default:
 		job.State = model.JobCancelled
-		err = errors.Errorf("invalid background job %v", job)
-
+		err = errInvalidBgJob
 	}
 
 	if err != nil {
diff --git a/ddl/column.go b/ddl/column.go
index c298c130bb929..5c82b0cf8611c 100644
--- a/ddl/column.go
+++ b/ddl/column.go
@@ -18,6 +18,7 @@ import (
 	"github.com/ngaut/log"
 	"github.com/pingcap/tidb/ast"
 	"github.com/pingcap/tidb/column"
+	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/model"
@@ -66,7 +67,7 @@ func (d *ddl) addColumn(tblInfo *model.TableInfo, colInfo *model.ColumnInfo, pos
 	} else if pos.Tp == ast.ColumnPositionAfter {
 		c := findCol(cols, pos.RelativeColumn.Name.L)
 		if c == nil {
-			return nil, 0, errors.Errorf("No such column: %v", pos.RelativeColumn)
+			return nil, 0, infoschema.ErrColumnNotExists.Gen("no such column: %v", pos.RelativeColumn)
 		}
 
 		// Insert position is after the mentioned column.
@@ -109,7 +110,7 @@ func (d *ddl) onAddColumn(t *meta.Meta, job *model.Job) error {
 		if columnInfo.State == model.StatePublic {
 			// we already have a column with same column name
 			job.State = model.JobCancelled
-			return errors.Errorf("ADD COLUMN: column already exist %s", col.Name.L)
+			return infoschema.ErrColumnExists.Gen("ADD COLUMN: column already exist %s", col.Name.L)
 		}
 	} else {
 		columnInfo, offset, err = d.addColumn(tblInfo, col, pos)
@@ -191,7 +192,7 @@ func (d *ddl) onAddColumn(t *meta.Meta, job *model.Job) error {
 		job.State = model.JobDone
 		return nil
 	default:
-		return errors.Errorf("invalid column state %v", columnInfo.State)
+		return ErrInvalidColumnState.Gen("invalid column state %v", columnInfo.State)
 	}
 }
 
@@ -212,12 +213,13 @@ func (d *ddl) onDropColumn(t *meta.Meta, job *model.Job) error {
 	colInfo := findCol(tblInfo.Columns, colName.L)
 	if colInfo == nil {
 		job.State = model.JobCancelled
-		return errors.Errorf("column %s doesn't exist", colName)
+		return infoschema.ErrColumnNotExists.Gen("column %s doesn't exist", colName)
 	}
 
 	if len(tblInfo.Columns) == 1 {
 		job.State = model.JobCancelled
-		return errors.Errorf("can't drop only column %s in table %s", colName, tblInfo.Name)
+		return ErrCantRemoveAllFields.Gen("can't drop only column %s in table %s",
+			colName, tblInfo.Name)
 	}
 
 	// we don't support drop column with index covered now.
@@ -226,7 +228,8 @@ func (d *ddl) onDropColumn(t *meta.Meta, job *model.Job) error {
 		for _, col := range indexInfo.Columns {
 			if col.Name.L == colName.L {
 				job.State = model.JobCancelled
-				return errors.Errorf("can't drop column %s with index %s covered now", colName, indexInfo.Name)
+				return errCantDropColWithIndex.Gen("can't drop column %s with index %s covered now",
+					colName, indexInfo.Name)
 			}
 		}
 	}
@@ -304,7 +307,7 @@ func (d *ddl) onDropColumn(t *meta.Meta, job *model.Job) error {
 		job.State = model.JobDone
 		return nil
 	default:
-		return errors.Errorf("invalid table state %v", tblInfo.State)
+		return ErrInvalidTableState.Gen("invalid table state %v", tblInfo.State)
 	}
 }
 
diff --git a/ddl/ddl.go b/ddl/ddl.go
index 4ee29100290ee..5f00207f1d340 100644
--- a/ddl/ddl.go
+++ b/ddl/ddl.go
@@ -37,6 +37,7 @@ import (
 	"github.com/pingcap/tidb/mysql"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/table"
+	"github.com/pingcap/tidb/terror"
 	"github.com/pingcap/tidb/util/charset"
 	"github.com/pingcap/tidb/util/types"
 	"github.com/twinj/uuid"
@@ -251,7 +252,7 @@ func (d *ddl) CreateSchema(ctx context.Context, schema model.CIStr, charsetInfo
 	is := d.GetInformationSchema()
 	_, ok := is.SchemaByName(schema)
 	if ok {
-		return errors.Trace(infoschema.DatabaseExists)
+		return errors.Trace(infoschema.ErrDatabaseExists)
 	}
 
 	schemaID, err := d.genGlobalID()
@@ -283,7 +284,7 @@ func (d *ddl) DropSchema(ctx context.Context, schema model.CIStr) (err error) {
 	is := d.GetInformationSchema()
 	old, ok := is.SchemaByName(schema)
 	if !ok {
-		return errors.Trace(infoschema.DatabaseNotExists)
+		return errors.Trace(infoschema.ErrDatabaseNotExists)
 	}
 
 	job := &model.Job{
@@ -466,14 +467,14 @@ func columnDefToCol(ctx context.Context, offset int, colDef *ast.ColumnDef) (*co
 			case ast.ColumnOptionDefaultValue:
 				value, err := getDefaultValue(ctx, v, colDef.Tp.Tp, colDef.Tp.Decimal)
 				if err != nil {
-					return nil, nil, errors.Errorf("invalid default value - %s", errors.Trace(err))
+					return nil, nil, ErrColumnBadNull.Gen("invalid default value - %s", err)
 				}
 				col.DefaultValue = value
 				hasDefaultValue = true
 				removeOnUpdateNowFlag(col)
 			case ast.ColumnOptionOnUpdate:
 				if !evaluator.IsCurrentTimeExpr(v.Expr) {
-					return nil, nil, errors.Errorf("invalid ON UPDATE for - %s", col.Name)
+					return nil, nil, ErrInvalidOnUpdate.Gen("invalid ON UPDATE for - %s", col.Name)
 				}
 
 				col.Flag |= mysql.OnUpdateNowFlag
@@ -575,7 +576,7 @@ func checkDefaultValue(c *column.Col, hasDefaultValue bool) error {
 
 	// Set not null but default null is invalid.
 	if mysql.HasNotNullFlag(c.Flag) {
-		return errors.Errorf("invalid default value for %s", c.Name)
+		return ErrColumnBadNull.Gen("invalid default value for %s", c.Name)
 	}
 
 	return nil
@@ -586,7 +587,7 @@ func checkDuplicateColumn(colDefs []*ast.ColumnDef) error {
 	for _, colDef := range colDefs {
 		nameLower := colDef.Name.Name.O
 		if colNames[nameLower] {
-			return errors.Errorf("CREATE TABLE: duplicate column %s", colDef.Name)
+			return infoschema.ErrColumnExists.Gen("duplicate column %s", colDef.Name)
 		}
 		colNames[nameLower] = true
 	}
@@ -605,7 +606,7 @@ func checkConstraintNames(constraints []*ast.Constraint) error {
 		if constr.Name != "" {
 			nameLower := strings.ToLower(constr.Name)
 			if constrNames[nameLower] {
-				return errors.Errorf("CREATE TABLE: duplicate key %s", constr.Name)
+				return infoschema.ErrIndexExists.Gen("CREATE TABLE: duplicate key %s", constr.Name)
 			}
 			constrNames[nameLower] = true
 		}
@@ -646,7 +647,7 @@ func (d *ddl) buildTableInfo(tableName model.CIStr, cols []*column.Col, constrai
 				key := constr.Keys[0]
 				col := column.FindCol(cols, key.Column.Name.O)
 				if col == nil {
-					return nil, errors.Errorf("No such column: %v", key)
+					return nil, infoschema.ErrColumnNotExists.Gen("no such column: %v", key)
 				}
 				switch col.Tp {
 				case mysql.TypeLong, mysql.TypeLonglong:
@@ -663,7 +664,7 @@ func (d *ddl) buildTableInfo(tableName model.CIStr, cols []*column.Col, constrai
 		for _, key := range constr.Keys {
 			col := column.FindCol(cols, key.Column.Name.O)
 			if col == nil {
-				return nil, errors.Errorf("No such column: %v", key)
+				return nil, infoschema.ErrColumnNotExists.Gen("no such column: %v", key)
 			}
 			indexColumns = append(indexColumns, &model.IndexColumn{
 				Name:   key.Column.Name,
@@ -705,10 +706,10 @@ func (d *ddl) CreateTable(ctx context.Context, ident ast.Ident, colDefs []*ast.C
 	is := d.GetInformationSchema()
 	schema, ok := is.SchemaByName(ident.Schema)
 	if !ok {
-		return infoschema.DatabaseNotExists.Gen("database %s not exists", ident.Schema)
+		return infoschema.ErrDatabaseNotExists.Gen("database %s not exists", ident.Schema)
 	}
 	if is.TableExists(ident.Schema, ident.Name) {
-		return errors.Trace(infoschema.TableExists)
+		return errors.Trace(infoschema.ErrTableExists)
 	}
 	if err = checkDuplicateColumn(colDefs); err != nil {
 		return errors.Trace(err)
@@ -768,7 +769,7 @@ func (d *ddl) handleTableOptions(options []*ast.TableOption, tbInfo *model.Table
 func (d *ddl) AlterTable(ctx context.Context, ident ast.Ident, specs []*ast.AlterTableSpec) (err error) {
 	// now we only allow one schema changes at the same time.
 	if len(specs) != 1 {
-		return errors.New("can't run multi schema changes in one DDL")
+		return errRunMultiSchemaChanges
 	}
 
 	for _, spec := range specs {
@@ -805,7 +806,7 @@ func checkColumnConstraint(constraints []*ast.ColumnOption) error {
 	for _, constraint := range constraints {
 		switch constraint.Tp {
 		case ast.ColumnOptionAutoIncrement, ast.ColumnOptionPrimaryKey, ast.ColumnOptionUniq, ast.ColumnOptionUniqKey:
-			return errors.Errorf("unsupported add column constraint - %v", constraint.Tp)
+			return errUnsupportedAddColumn.Gen("unsupported add column constraint - %v", constraint.Tp)
 		}
 	}
 
@@ -823,19 +824,19 @@ func (d *ddl) AddColumn(ctx context.Context, ti ast.Ident, spec *ast.AlterTableS
 	is := d.infoHandle.Get()
 	schema, ok := is.SchemaByName(ti.Schema)
 	if !ok {
-		return errors.Trace(infoschema.DatabaseNotExists)
+		return errors.Trace(infoschema.ErrDatabaseNotExists)
 	}
 
 	t, err := is.TableByName(ti.Schema, ti.Name)
 	if err != nil {
-		return errors.Trace(infoschema.TableNotExists)
+		return errors.Trace(infoschema.ErrTableNotExists)
 	}
 
 	// Check whether added column has existed.
 	colName := spec.Column.Name.Name.O
 	col := column.FindCol(t.Cols(), colName)
 	if col != nil {
-		return errors.Errorf("column %s already exists", colName)
+		return infoschema.ErrColumnExists.Gen("column %s already exists", colName)
 	}
 
 	// ingore table constraints now, maybe return error later
@@ -863,18 +864,18 @@ func (d *ddl) DropColumn(ctx context.Context, ti ast.Ident, colName model.CIStr)
 	is := d.infoHandle.Get()
 	schema, ok := is.SchemaByName(ti.Schema)
 	if !ok {
-		return errors.Trace(infoschema.DatabaseNotExists)
+		return errors.Trace(infoschema.ErrDatabaseNotExists)
 	}
 
 	t, err := is.TableByName(ti.Schema, ti.Name)
 	if err != nil {
-		return errors.Trace(infoschema.TableNotExists)
+		return errors.Trace(infoschema.ErrTableNotExists)
 	}
 
 	// Check whether dropped column has existed.
 	col := column.FindCol(t.Cols(), colName.L)
 	if col == nil {
-		return errors.Errorf("column %s doesn’t exist", colName.L)
+		return infoschema.ErrColumnNotExists.Gen("column %s doesn’t exist", colName.L)
 	}
 
 	job := &model.Job{
@@ -894,12 +895,12 @@ func (d *ddl) DropTable(ctx context.Context, ti ast.Ident) (err error) {
 	is := d.GetInformationSchema()
 	schema, ok := is.SchemaByName(ti.Schema)
 	if !ok {
-		return infoschema.DatabaseNotExists.Gen("database %s not exists", ti.Schema)
+		return infoschema.ErrDatabaseNotExists.Gen("database %s not exists", ti.Schema)
 	}
 
 	tb, err := is.TableByName(ti.Schema, ti.Name)
 	if err != nil {
-		return errors.Trace(infoschema.TableNotExists)
+		return errors.Trace(infoschema.ErrTableNotExists)
 	}
 
 	job := &model.Job{
@@ -917,12 +918,12 @@ func (d *ddl) CreateIndex(ctx context.Context, ti ast.Ident, unique bool, indexN
 	is := d.infoHandle.Get()
 	schema, ok := is.SchemaByName(ti.Schema)
 	if !ok {
-		return infoschema.DatabaseNotExists.Gen("database %s not exists", ti.Schema)
+		return infoschema.ErrDatabaseNotExists.Gen("database %s not exists", ti.Schema)
 	}
 
 	t, err := is.TableByName(ti.Schema, ti.Name)
 	if err != nil {
-		return errors.Trace(infoschema.TableNotExists)
+		return errors.Trace(infoschema.ErrTableNotExists)
 	}
 	indexID, err := d.genGlobalID()
 	if err != nil {
@@ -945,12 +946,12 @@ func (d *ddl) DropIndex(ctx context.Context, ti ast.Ident, indexName model.CIStr
 	is := d.infoHandle.Get()
 	schema, ok := is.SchemaByName(ti.Schema)
 	if !ok {
-		return errors.Trace(infoschema.DatabaseNotExists)
+		return errors.Trace(infoschema.ErrDatabaseNotExists)
 	}
 
 	t, err := is.TableByName(ti.Schema, ti.Name)
 	if err != nil {
-		return errors.Trace(infoschema.TableNotExists)
+		return errors.Trace(infoschema.ErrTableNotExists)
 	}
 
 	job := &model.Job{
@@ -976,3 +977,73 @@ func findCol(cols []*model.ColumnInfo, name string) *model.ColumnInfo {
 
 	return nil
 }
+
+// DDL error codes.
+const (
+	codeInvalidWorker         terror.ErrCode = 1
+	codeNotOwner                             = 2
+	codeInvalidDDLJob                        = 3
+	codeInvalidBgJob                         = 4
+	codeInvalidJobFlag                       = 5
+	codeRunMultiSchemaChanges                = 6
+	codeWaitReorgTimeout                     = 7
+	codeInvalidStoreVer                      = 8
+
+	codeInvalidDBState     = 100
+	codeInvalidTableState  = 101
+	codeInvalidColumnState = 102
+	codeInvalidIndexState  = 103
+
+	codeCantDropColWithIndex = 201
+	codeUnsupportedAddColumn = 202
+
+	codeBadNull             = 1048
+	codeCantRemoveAllFields = 1090
+	codeCantDropFieldOrKey  = 1091
+	codeInvalidOnUpdate     = 1294
+)
+
+var (
+	// errWorkerClosed means we have already closed the DDL worker.
+	errInvalidWorker = terror.ClassDDL.New(codeInvalidWorker, "invalid worker")
+	// errNotOwner means we are not owner and can't handle DDL jobs.
+	errNotOwner              = terror.ClassDDL.New(codeNotOwner, "not Owner")
+	errInvalidDDLJob         = terror.ClassDDL.New(codeInvalidDDLJob, "invalid ddl job")
+	errInvalidBgJob          = terror.ClassDDL.New(codeInvalidBgJob, "invalid background job")
+	errInvalidJobFlag        = terror.ClassDDL.New(codeInvalidJobFlag, "invalid job flag")
+	errRunMultiSchemaChanges = terror.ClassDDL.New(codeRunMultiSchemaChanges, "can't run multi schema change")
+	errWaitReorgTimeout      = terror.ClassDDL.New(codeWaitReorgTimeout, "wait for reorganization timeout")
+	errInvalidStoreVer       = terror.ClassDDL.New(codeInvalidStoreVer, "invalid storage current version")
+
+	// we don't support drop column with index covered now.
+	errCantDropColWithIndex = terror.ClassDDL.New(codeCantDropColWithIndex, "can't drop column with index")
+	errUnsupportedAddColumn = terror.ClassDDL.New(codeUnsupportedAddColumn, "unsupported add column")
+
+	// ErrInvalidDBState returns for invalid database state.
+	ErrInvalidDBState = terror.ClassDDL.New(codeInvalidDBState, "invalid database state")
+	// ErrInvalidTableState returns for invalid Table state.
+	ErrInvalidTableState = terror.ClassDDL.New(codeInvalidTableState, "invalid table state")
+	// ErrInvalidColumnState returns for invalid column state.
+	ErrInvalidColumnState = terror.ClassDDL.New(codeInvalidColumnState, "invalid column state")
+	// ErrInvalidIndexState returns for invalid index state.
+	ErrInvalidIndexState = terror.ClassDDL.New(codeInvalidIndexState, "invalid index state")
+
+	// ErrColumnBadNull returns for a bad null value.
+	ErrColumnBadNull = terror.ClassDDL.New(codeBadNull, "column cann't be null")
+	// ErrCantRemoveAllFields returns for deleting all columns.
+	ErrCantRemoveAllFields = terror.ClassDDL.New(codeCantRemoveAllFields, "can't delete all columns with ALTER TABLE")
+	// ErrCantDropFieldOrKey returns for dropping a non-existent field or key.
+	ErrCantDropFieldOrKey = terror.ClassDDL.New(codeCantDropFieldOrKey, "can't drop field; check that column/key exists")
+	// ErrInvalidOnUpdate returns for invalid ON UPDATE clause.
+	ErrInvalidOnUpdate = terror.ClassDDL.New(codeInvalidOnUpdate, "invalid ON UPDATE clause for the column")
+)
+
+func init() {
+	ddlMySQLERrCodes := map[terror.ErrCode]uint16{
+		codeBadNull:             mysql.ErrBadNull,
+		codeCantRemoveAllFields: mysql.ErrCantRemoveAllFields,
+		codeCantDropFieldOrKey:  mysql.ErrCantDropFieldOrKey,
+		codeInvalidOnUpdate:     mysql.ErrInvalidOnUpdate,
+	}
+	terror.ErrClassToMySQLCodes[terror.ClassDDL] = ddlMySQLERrCodes
+}
diff --git a/ddl/ddl_worker.go b/ddl/ddl_worker.go
index e8bd97024c975..480eb19398b77 100644
--- a/ddl/ddl_worker.go
+++ b/ddl/ddl_worker.go
@@ -150,7 +150,7 @@ func (d *ddl) checkOwner(t *meta.Meta, flag JobType) (*model.Owner, error) {
 
 	if owner.OwnerID != d.uuid {
 		log.Debugf("[ddl] not %s job owner, owner is %s", flag, owner.OwnerID)
-		return nil, errors.Trace(ErrNotOwner)
+		return nil, errors.Trace(errNotOwner)
 	}
 
 	return owner, nil
@@ -185,14 +185,6 @@ func (d *ddl) finishDDLJob(t *meta.Meta, job *model.Job) error {
 	return errors.Trace(err)
 }
 
-// ErrNotOwner means we are not owner and can't handle DDL jobs.
-var ErrNotOwner = errors.New("DDL: not owner")
-
-// ErrWorkerClosed means we have already closed the DDL worker.
-var ErrWorkerClosed = errors.New("DDL: worker is closed")
-
-var errInvalidJobFlag = errors.New("DDL: invalid job flag")
-
 // JobType is job type, including ddl/background.
 type JobType int
 
@@ -224,7 +216,7 @@ func (d *ddl) handleDDLJobQueue() error {
 		err := kv.RunInNewTxn(d.store, false, func(txn kv.Transaction) error {
 			t := meta.NewMeta(txn)
 			owner, err := d.checkOwner(t, ddlJobFlag)
-			if terror.ErrorEqual(err, ErrNotOwner) {
+			if terror.ErrorEqual(err, errNotOwner) {
 				// we are not owner, return and retry checking later.
 				return nil
 			} else if err != nil {
@@ -363,7 +355,7 @@ func (d *ddl) runDDLJob(t *meta.Meta, job *model.Job) {
 	default:
 		// invalid job, cancel it.
 		job.State = model.JobCancelled
-		err = errors.Errorf("invalid ddl job %v", job)
+		err = errInvalidDDLJob.Gen("invalid ddl job %v", job)
 	}
 
 	// saves error in job, so that others can know error happens.
diff --git a/ddl/ddl_worker_test.go b/ddl/ddl_worker_test.go
index c96621abc7bfd..c725f793ab68f 100644
--- a/ddl/ddl_worker_test.go
+++ b/ddl/ddl_worker_test.go
@@ -61,7 +61,7 @@ func testCheckOwner(c *C, d *ddl, isOwner bool, flag JobType) {
 		return
 	}
 
-	c.Assert(terror.ErrorEqual(err, ErrNotOwner), IsTrue)
+	c.Assert(terror.ErrorEqual(err, errNotOwner), IsTrue)
 }
 
 func (s *testDDLSuite) TestCheckOwner(c *C) {
diff --git a/ddl/index.go b/ddl/index.go
index d4f5fc49902f3..3e9c3ec8bf1a4 100644
--- a/ddl/index.go
+++ b/ddl/index.go
@@ -17,6 +17,7 @@ import (
 	"github.com/juju/errors"
 	"github.com/ngaut/log"
 	"github.com/pingcap/tidb/ast"
+	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/model"
@@ -28,13 +29,15 @@ import (
 	"github.com/pingcap/tidb/util/types"
 )
 
-func buildIndexInfo(tblInfo *model.TableInfo, unique bool, indexName model.CIStr, indexID int64, idxColNames []*ast.IndexColName) (*model.IndexInfo, error) {
+func buildIndexInfo(tblInfo *model.TableInfo, unique bool, indexName model.CIStr, indexID int64,
+	idxColNames []*ast.IndexColName) (*model.IndexInfo, error) {
 	// build offsets
 	idxColumns := make([]*model.IndexColumn, 0, len(idxColNames))
 	for _, ic := range idxColNames {
 		col := findCol(tblInfo.Columns, ic.Column.Name.O)
 		if col == nil {
-			return nil, errors.Errorf("CREATE INDEX: column does not exist: %s", ic.Column.Name.O)
+			return nil, infoschema.ErrColumnNotExists.Gen("CREATE INDEX: column does not exist: %s",
+				ic.Column.Name.O)
 		}
 
 		idxColumns = append(idxColumns, &model.IndexColumn{
@@ -113,7 +116,7 @@ func (d *ddl) onCreateIndex(t *meta.Meta, job *model.Job) error {
 			if idx.State == model.StatePublic {
 				// we already have a index with same index name
 				job.State = model.JobCancelled
-				return errors.Errorf("CREATE INDEX: index already exist %s", indexName)
+				return infoschema.ErrIndexExists.Gen("CREATE INDEX: index already exist %s", indexName)
 			}
 
 			indexInfo = idx
@@ -194,7 +197,7 @@ func (d *ddl) onCreateIndex(t *meta.Meta, job *model.Job) error {
 		job.State = model.JobDone
 		return nil
 	default:
-		return errors.Errorf("invalid index state %v", tblInfo.State)
+		return ErrInvalidIndexState.Gen("invalid index state %v", tblInfo.State)
 	}
 }
 
@@ -220,7 +223,7 @@ func (d *ddl) onDropIndex(t *meta.Meta, job *model.Job) error {
 
 	if indexInfo == nil {
 		job.State = model.JobCancelled
-		return errors.Errorf("index %s doesn't exist", indexName)
+		return ErrCantDropFieldOrKey.Gen("index %s doesn't exist", indexName)
 	}
 
 	_, err = t.GenSchemaVersion()
@@ -285,7 +288,7 @@ func (d *ddl) onDropIndex(t *meta.Meta, job *model.Job) error {
 		job.State = model.JobDone
 		return nil
 	default:
-		return errors.Errorf("invalid table state %v", tblInfo.State)
+		return ErrInvalidTableState.Gen("invalid table state %v", tblInfo.State)
 	}
 }
 
diff --git a/ddl/reorg.go b/ddl/reorg.go
index a00c072974555..e8f6828933bb9 100644
--- a/ddl/reorg.go
+++ b/ddl/reorg.go
@@ -97,8 +97,6 @@ func (d *ddl) newReorgContext() context.Context {
 
 const waitReorgTimeout = 10 * time.Second
 
-var errWaitReorgTimeout = errors.New("wait for reorganization timeout")
-
 func (d *ddl) runReorgJob(f func() error) error {
 	if d.reorgDoneCh == nil {
 		// start a reorganization job
@@ -137,7 +135,7 @@ func (d *ddl) runReorgJob(f func() error) error {
 func (d *ddl) isReorgRunnable(txn kv.Transaction) error {
 	if d.isClosed() {
 		// worker is closed, can't run reorganization.
-		return errors.Trace(ErrWorkerClosed)
+		return errors.Trace(errInvalidWorker.Gen("worker is closed"))
 	}
 
 	t := meta.NewMeta(txn)
@@ -147,7 +145,7 @@ func (d *ddl) isReorgRunnable(txn kv.Transaction) error {
 	} else if owner == nil || owner.OwnerID != d.uuid {
 		// if no owner, we will try later, so here just return error.
 		// or another server is owner, return error too.
-		return errors.Trace(ErrNotOwner)
+		return errors.Trace(errNotOwner)
 	}
 
 	return nil
@@ -225,7 +223,7 @@ func (d *ddl) getReorgInfo(t *meta.Meta, job *model.Job) (*reorgInfo, error) {
 		if err != nil {
 			return nil, errors.Trace(err)
 		} else if ver.Ver <= 0 {
-			return nil, errors.Errorf("invalid storage current version %d", ver.Ver)
+			return nil, errInvalidStoreVer.Gen("invalid storage current version %d", ver.Ver)
 		}
 
 		job.SnapshotVer = ver.Ver
diff --git a/ddl/schema.go b/ddl/schema.go
index ad411a8c9cadc..37258b853e7a5 100644
--- a/ddl/schema.go
+++ b/ddl/schema.go
@@ -45,7 +45,7 @@ func (d *ddl) onCreateSchema(t *meta.Meta, job *model.Job) error {
 			if db.ID != schemaID {
 				// database exists, can't create, we should cancel this job now.
 				job.State = model.JobCancelled
-				return errors.Trace(infoschema.DatabaseExists)
+				return errors.Trace(infoschema.ErrDatabaseExists)
 			}
 			dbInfo = db
 		}
@@ -109,7 +109,7 @@ func (d *ddl) onDropSchema(t *meta.Meta, job *model.Job) error {
 	}
 	if dbInfo == nil {
 		job.State = model.JobCancelled
-		return errors.Trace(infoschema.DatabaseNotExists)
+		return errors.Trace(infoschema.ErrDatabaseNotExists)
 	}
 
 	_, err = t.GenSchemaVersion()
diff --git a/ddl/table.go b/ddl/table.go
index ef9708a75bf96..ed4ea49524098 100644
--- a/ddl/table.go
+++ b/ddl/table.go
@@ -37,7 +37,7 @@ func (d *ddl) onCreateTable(t *meta.Meta, job *model.Job) error {
 	tables, err := t.ListTables(schemaID)
 	if terror.ErrorEqual(err, meta.ErrDBNotExists) {
 		job.State = model.JobCancelled
-		return errors.Trace(infoschema.DatabaseNotExists)
+		return errors.Trace(infoschema.ErrDatabaseNotExists)
 	} else if err != nil {
 		return errors.Trace(err)
 	}
@@ -47,7 +47,7 @@ func (d *ddl) onCreateTable(t *meta.Meta, job *model.Job) error {
 			if tbl.ID != tbInfo.ID {
 				// table exists, can't create, we should cancel this job now.
 				job.State = model.JobCancelled
-				return errors.Trace(infoschema.TableExists)
+				return errors.Trace(infoschema.ErrTableExists)
 			}
 
 			tbInfo = tbl
@@ -72,7 +72,7 @@ func (d *ddl) onCreateTable(t *meta.Meta, job *model.Job) error {
 		job.State = model.JobDone
 		return nil
 	default:
-		return errors.Errorf("invalid table state %v", tbInfo.State)
+		return ErrInvalidTableState.Gen("invalid table state %v", tbInfo.State)
 	}
 }
 
@@ -109,14 +109,14 @@ func (d *ddl) onDropTable(t *meta.Meta, job *model.Job) error {
 	tblInfo, err := t.GetTable(schemaID, tableID)
 	if terror.ErrorEqual(err, meta.ErrDBNotExists) {
 		job.State = model.JobCancelled
-		return errors.Trace(infoschema.DatabaseNotExists)
+		return errors.Trace(infoschema.ErrDatabaseNotExists)
 	} else if err != nil {
 		return errors.Trace(err)
 	}
 
 	if tblInfo == nil {
 		job.State = model.JobCancelled
-		return errors.Trace(infoschema.TableNotExists)
+		return errors.Trace(infoschema.ErrTableNotExists)
 	}
 
 	_, err = t.GenSchemaVersion()
@@ -146,7 +146,7 @@ func (d *ddl) onDropTable(t *meta.Meta, job *model.Job) error {
 		job.State = model.JobDone
 		job.SchemaState = model.StateNone
 	default:
-		err = errors.Errorf("invalid table state %v", tblInfo.State)
+		err = ErrInvalidTableState.Gen("invalid table state %v", tblInfo.State)
 	}
 
 	return errors.Trace(err)
@@ -164,17 +164,17 @@ func (d *ddl) getTableInfo(t *meta.Meta, job *model.Job) (*model.TableInfo, erro
 	tblInfo, err := t.GetTable(schemaID, tableID)
 	if terror.ErrorEqual(err, meta.ErrDBNotExists) {
 		job.State = model.JobCancelled
-		return nil, errors.Trace(infoschema.DatabaseNotExists)
+		return nil, errors.Trace(infoschema.ErrDatabaseNotExists)
 	} else if err != nil {
 		return nil, errors.Trace(err)
 	} else if tblInfo == nil {
 		job.State = model.JobCancelled
-		return nil, errors.Trace(infoschema.TableNotExists)
+		return nil, errors.Trace(infoschema.ErrTableNotExists)
 	}
 
 	if tblInfo.State != model.StatePublic {
 		job.State = model.JobCancelled
-		return nil, errors.Errorf("table %s is not in public, but %s", tblInfo.Name.L, tblInfo.State)
+		return nil, ErrInvalidTableState.Gen("table %s is not in public, but %s", tblInfo.Name.L, tblInfo.State)
 	}
 
 	return tblInfo, nil
diff --git a/terror/terror.go b/terror/terror.go
index 46ce78e5f13bf..1596d0d2fce33 100644
--- a/terror/terror.go
+++ b/terror/terror.go
@@ -63,6 +63,7 @@ const (
 	ClassServer
 	ClassVariable
 	ClassExpression
+	ClassDDL
 	// Add more as needed.
 )
 
@@ -85,6 +86,8 @@ func (ec ErrClass) String() string {
 		return "variable"
 	case ClassExpression:
 		return "expression"
+	case ClassDDL:
+		return "ddl"
 	}
 	return strconv.Itoa(int(ec))
 }