Merge branch 'master' into fix-create-table-like

pingcap · Mar 7, 2019 · 08ace95 · 08ace95
2 parents 121a716 + b3d49c2
commit 08ace95
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 21 deletions.
diff --git a/ddl/foreign_key_test.go b/ddl/foreign_key_test.go
@@ -154,8 +154,9 @@ func (s *testForeighKeySuite) TestForeignKey(c *C) {
 		}
 		checkOK = true
 	}
+	originalHook := d.GetHook()
+	defer d.SetHook(originalHook)
 	d.SetHook(tc)
-
 	d.Stop()
 	d.start(context.Background(), nil)
 
@@ -177,7 +178,9 @@ func (s *testForeighKeySuite) TestForeignKey(c *C) {
 	mu.Lock()
 	checkOK = false
 	mu.Unlock()
-	tc.onJobUpdated = func(job *model.Job) {
+	// fix data race pr/#9491
+	tc2 := &TestDDLCallback{}
+	tc2.onJobUpdated = func(job *model.Job) {
 		if job.State != model.JobStateDone {
 			return
 		}
@@ -196,7 +199,7 @@ func (s *testForeighKeySuite) TestForeignKey(c *C) {
 		}
 		checkOK = true
 	}
-
+	d.SetHook(tc2)
 	d.Stop()
 	d.start(context.Background(), nil)
 
@@ -212,9 +215,6 @@ func (s *testForeighKeySuite) TestForeignKey(c *C) {
 	err = ctx.NewTxn(context.Background())
 	c.Assert(err, IsNil)
 
-	tc.onJobUpdated = func(job *model.Job) {
-	}
-
 	d.Stop()
 	d.start(context.Background(), nil)
 

diff --git a/domain/domain.go b/domain/domain.go
@@ -605,8 +605,9 @@ func (do *Domain) Init(ddlLease time.Duration, sysFactory func(*Domain) (pools.R
 		if addrs := ebd.EtcdAddrs(); addrs != nil {
 			cfg := config.GetGlobalConfig()
 			cli, err := clientv3.New(clientv3.Config{
-				Endpoints:   addrs,
-				DialTimeout: 5 * time.Second,
+				Endpoints:        addrs,
+				AutoSyncInterval: 30 * time.Second,
+				DialTimeout:      5 * time.Second,
 				DialOptions: []grpc.DialOption{
 					grpc.WithUnaryInterceptor(grpc_prometheus.UnaryClientInterceptor),
 					grpc.WithStreamInterceptor(grpc_prometheus.StreamClientInterceptor),

diff --git a/executor/insert_common.go b/executor/insert_common.go
@@ -191,15 +191,13 @@ func (e *InsertValues) insertRows(ctx context.Context, exec func(ctx context.Con
 			return errors.Trace(err)
 		}
 		rows = append(rows, row)
-		if e.rowCount%uint64(batchSize) == 0 {
+		if batchInsert && e.rowCount%uint64(batchSize) == 0 {
 			if err = exec(ctx, rows); err != nil {
-				return err
+				return errors.Trace(err)
 			}
 			rows = rows[:0]
-			if batchInsert {
-				if err = e.doBatchInsert(ctx); err != nil {
-					return err
-				}
+			if err = e.doBatchInsert(ctx); err != nil {
+				return err
 			}
 		}
 	}
@@ -323,15 +321,13 @@ func (e *InsertValues) insertRowsFromSelect(ctx context.Context, exec func(ctx c
 				return errors.Trace(err)
 			}
 			rows = append(rows, row)
-			if e.rowCount%uint64(batchSize) == 0 {
+			if batchInsert && e.rowCount%uint64(batchSize) == 0 {
 				if err = exec(ctx, rows); err != nil {
 					return errors.Trace(err)
 				}
 				rows = rows[:0]
-				if batchInsert {
-					if err = e.doBatchInsert(ctx); err != nil {
-						return err
-					}
+				if err = e.doBatchInsert(ctx); err != nil {
+					return err
 				}
 			}
 		}

diff --git a/store/tikv/kv.go b/store/tikv/kv.go
@@ -51,8 +51,9 @@ type Driver struct {
 
 func createEtcdKV(addrs []string, tlsConfig *tls.Config) (*clientv3.Client, error) {
 	cli, err := clientv3.New(clientv3.Config{
-		Endpoints:   addrs,
-		DialTimeout: 5 * time.Second,
+		Endpoints:        addrs,
+		AutoSyncInterval: 30 * time.Second,
+		DialTimeout:      5 * time.Second,
 		DialOptions: []grpc.DialOption{
 			grpc.WithUnaryInterceptor(grpc_prometheus.UnaryClientInterceptor),
 			grpc.WithStreamInterceptor(grpc_prometheus.StreamClientInterceptor),