Merge pull request #585 from pingcap/siddontang/update-reorg

update reorg

Author: qiuyesuifeng

ddl/column.go

@@ -335,6 +335,10 @@ func (d *ddl) backfillColumnData(t table.Table, columnInfo *model.ColumnInfo, ha
 		log.Info("backfill column...", handle)
 
 		err := kv.RunInNewTxn(d.store, true, func(txn kv.Transaction) error {
+			if err := d.isOwnerInReorg(txn); err != nil {
+				return errors.Trace(err)
+			}
+
 			// First check if row exists.
 			exist, err := checkRowExist(txn, t, handle)
 			if err != nil {
@@ -401,6 +405,10 @@ func (d *ddl) dropTableColumn(t table.Table, colInfo *model.ColumnInfo, reorgInf
 		seekHandle = handles[len(handles)-1] + 1
 
 		err = kv.RunInNewTxn(d.store, true, func(txn kv.Transaction) error {
+			if err := d.isOwnerInReorg(txn); err != nil {
+				return errors.Trace(err)
+			}
+
 			var h int64
 			for _, h = range handles {
 				key := t.RecordKey(h, col)

ddl/index.go

@@ -421,6 +421,10 @@ func (d *ddl) backfillTableIndex(t table.Table, indexInfo *model.IndexInfo, hand
 		log.Debug("building index...", handle)
 
 		err := kv.RunInNewTxn(d.store, true, func(txn kv.Transaction) error {
+			if err := d.isOwnerInReorg(txn); err != nil {
+				return errors.Trace(err)
+			}
+
 			// first check row exists
 			exist, err := checkRowExist(txn, t, handle)
 			if err != nil {

ddl/reorg.go

@@ -108,28 +108,54 @@ func (d *ddl) runReorgJob(f func() error) error {
 		}()
 	}
 
-	waitTimeout := chooseLeaseTime(d.lease, waitReorgTimeout)
+	waitTimeout := waitReorgTimeout
+	// if d.lease is 0, we are using a local storage,
+	// and we can wait the reorganization to be done here.
+	// if d.lease > 0, we don't need to wait here because
+	// we will wait 2 * lease outer and try checking again,
+	// so we use a very little timeout here.
+	if d.lease > 0 {
+		waitTimeout = 1 * time.Millisecond
+	}
 
 	// wait reorganization job done or timeout
 	select {
 	case err := <-d.reorgDoneCh:
 		d.reorgDoneCh = nil
 		return errors.Trace(err)
-	case <-time.After(waitTimeout):
-		// if timeout, we will return, check the owner and retry wait job done again.
-		return errWaitReorgTimeout
 	case <-d.quitCh:
 		// we return errWaitReorgTimeout here too, so that outer loop will break.
 		return errWaitReorgTimeout
+	case <-time.After(waitTimeout):
+		// if timeout, we will return, check the owner and retry to wait job done again.
+		return errWaitReorgTimeout
 	}
 }
 
+func (d *ddl) isOwnerInReorg(txn kv.Transaction) error {
+	t := meta.NewMeta(txn)
+	owner, err := t.GetDDLOwner()
+	if err != nil {
+		return errors.Trace(err)
+	} 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 nil
+}
+
 func (d *ddl) delKeysWithPrefix(prefix string) error {
 	keys := make([]string, maxBatchSize)
 
 	for {
 		keys := keys[0:0]
 		err := kv.RunInNewTxn(d.store, true, func(txn kv.Transaction) error {
+			if err := d.isOwnerInReorg(txn); err != nil {
+				return errors.Trace(err)
+			}
+
 			iter, err := txn.Seek([]byte(prefix))
 			if err != nil {
 				return errors.Trace(err)

ddl/reorg_test.go

@@ -112,3 +112,57 @@ func (s *testDDLSuite) TestReorg(c *C) {
 	err = info.RemoveHandle()
 	c.Assert(err, IsNil)
 }
+
+func (s *testDDLSuite) TestReorgOwner(c *C) {
+	store := testCreateStore(c, "test_reorg_owner")
+	defer store.Close()
+
+	lease := 50 * time.Millisecond
+
+	d1 := newDDL(store, nil, nil, lease)
+	defer d1.close()
+
+	ctx := testNewContext(c, d1)
+
+	testCheckOwner(c, d1, true)
+
+	d2 := newDDL(store, nil, nil, lease)
+	defer d2.close()
+
+	dbInfo := testSchemaInfo(c, d1, "test")
+	testCreateSchema(c, ctx, d1, dbInfo)
+
+	tblInfo := testTableInfo(c, d1, "t", 3)
+	testCreateTable(c, ctx, d1, dbInfo, tblInfo)
+
+	t := testGetTable(c, d1, dbInfo.ID, tblInfo.ID)
+
+	num := 10
+	for i := 0; i < num; i++ {
+		_, err := t.AddRecord(ctx, []interface{}{i, i, i})
+		c.Assert(err, IsNil)
+	}
+
+	err := ctx.FinishTxn(false)
+	c.Assert(err, IsNil)
+
+	tc := &testDDLCallback{}
+	tc.onJobRunBefore = func(job *model.Job) {
+		if job.SchemaState == model.StateDeleteReorganization {
+			d1.close()
+		}
+	}
+
+	d1.hook = tc
+
+	testDropSchema(c, ctx, d1, dbInfo)
+
+	err = kv.RunInNewTxn(d1.store, false, func(txn kv.Transaction) error {
+		t := meta.NewMeta(txn)
+		db, err1 := t.GetDatabase(dbInfo.ID)
+		c.Assert(err1, IsNil)
+		c.Assert(db, IsNil)
+		return nil
+	})
+	c.Assert(err, IsNil)
+}

tidb.go

@@ -74,7 +74,7 @@ func (dm *domainMap) Get(store kv.Storage) (d *domain.Domain, err error) {
 	// if not local storage, first we will use a little lease time to
 	// bootstrap quickly, after bootstrapped, we will reset the lease time.
 	if !localstore.IsLocalStore(store) {
-		lease = 100 * time.Second
+		lease = 100 * time.Millisecond
 	}
 	d, err = domain.NewDomain(store, lease)
 	if err != nil {