ddl: Add the txn entry limit (pingcap#4458)

ddl/ddl_db_test.go

@@ -1080,6 +1080,24 @@ func (s *testDBSuite) TestUpdateMultipleTable(c *C) {
 	tk.MustQuery("select * from t1").Check(testkit.Rows("8 1 9", "8 2 9"))
 }
 
+func (s *testDBSuite) TestCreateTableTooLarge(c *C) {
+	defer testleak.AfterTest(c)
+	s.tk = testkit.NewTestKit(c, s.store)
+	s.tk.MustExec("use test")
+
+	sql := "create table t_too_large ("
+	cnt := 3000
+	for i := 1; i <= cnt; i++ {
+		sql += fmt.Sprintf("a%d double, b%d double, c%d double, d%d double", i, i, i, i)
+		if i != cnt {
+			sql += ","
+		}
+	}
+	sql += ");"
+	_, err := s.tk.Exec(sql)
+	c.Assert(kv.ErrEntryTooLarge.Equal(err), IsTrue, Commentf("sql:%v", sql))
+}
+
 func (s *testDBSuite) TestCreateTableWithLike(c *C) {
 	defer testleak.AfterTest(c)
 	store, err := tidb.NewStore("memory://create_table_like")

ddl/ddl_worker.go

@@ -97,6 +97,23 @@ func (d *ddl) getFirstDDLJob(t *meta.Meta) (*model.Job, error) {
 	return job, errors.Trace(err)
 }
 
+// handleUpdateJobError handles the too large DDL job.
+func (d *ddl) handleUpdateJobError(t *meta.Meta, job *model.Job, err error) error {
+	if err == nil {
+		return nil
+	}
+	if kv.ErrEntryTooLarge.Equal(err) {
+		log.Warnf("[ddl] update DDL job %v failed %v", job, errors.ErrorStack(err))
+		// Reduce this txn entry size.
+		job.BinlogInfo.Clean()
+		job.Error = toTError(err)
+		job.SchemaState = model.StateNone
+		job.State = model.JobCancelled
+		err = d.finishDDLJob(t, job)
+	}
+	return errors.Trace(err)
+}
+
 // updateDDLJob updates the DDL job information.
 // Every time we enter another state except final state, we must call this function.
 func (d *ddl) updateDDLJob(t *meta.Meta, job *model.Job, updateTS uint64) error {
@@ -193,7 +210,7 @@ func (d *ddl) handleDDLJobQueue() error {
 				return errors.Trace(err)
 			}
 			err = d.updateDDLJob(t, job, txn.StartTS())
-			return errors.Trace(err)
+			return errors.Trace(d.handleUpdateJobError(t, job, err))
 		})
 		if err != nil {
 			return errors.Trace(err)

model/ddl.go

@@ -100,6 +100,13 @@ func (h *HistoryInfo) AddTableInfo(schemaVer int64, tblInfo *TableInfo) {
 	h.TableInfo = tblInfo
 }
 
+// Clean cleans history information.
+func (h *HistoryInfo) Clean() {
+	h.SchemaVersion = 0
+	h.DBInfo = nil
+	h.TableInfo = nil
+}
+
 // Job is for a DDL operation.
 type Job struct {
 	ID       int64         `json:"id"`

model/model_test.go

@@ -148,12 +148,13 @@ func (*testModelSuite) TestJobCodec(c *C) {
 	c.Assert(a, DeepEquals, A{Name: ""})
 	c.Assert(len(newJob.String()), Greater, 0)
 
+	job.BinlogInfo.Clean()
 	b1, err := job.Encode(true)
 	c.Assert(err, IsNil)
 	newJob = &Job{}
 	err = newJob.Decode(b1)
 	c.Assert(err, IsNil)
-	c.Assert(newJob.BinlogInfo, DeepEquals, job.BinlogInfo)
+	c.Assert(newJob.BinlogInfo, DeepEquals, &HistoryInfo{})
 	name = CIStr{}
 	a = A{}
 	err = newJob.DecodeArgs(&name, &a)