Merge pull request #626 from andyli029/20200426_fix_insert_deadlock

planner: InsertPlan sorts subquery by subTable in increasing order to…

src/planner/insert_plan.go

@@ -163,8 +163,16 @@ func (p *InsertPlan) Build() error {
 		val.vals = append(val.vals, row)
 	}
 
+	// sorts SQL by rewrittenTable in increasing order to avoid deadlock #605.
+	ks := []string{}
+	for k, _ := range vals {
+		ks = append(ks, k)
+	}
+	sort.Strings(ks)
+
 	// Rebuild querys with router info.
-	for rewritten, v := range vals {
+	for _, rewritten := range ks {
+		v := vals[rewritten]
 		buf := sqlparser.NewTrackedBuffer(nil)
 		buf.Myprintf("%s %v%sinto %s.%s%v %v%v", node.Action, node.Comments, node.Ignore, database, rewritten, node.Columns, v.vals, node.OnDup)
 		tuple := xcontext.QueryTuple{
@@ -190,8 +198,6 @@ func (p *InsertPlan) JSON() string {
 	}
 
 	var parts []xcontext.QueryTuple
-	// Sort.
-	sort.Sort(xcontext.QueryTuples(p.Querys))
 	parts = append(parts, p.Querys...)
 	exp := &explain{
 		RawQuery:   p.RawQuery,

src/planner/insert_plan_test.go

@@ -96,6 +96,71 @@ func TestInsertPlan(t *testing.T) {
 	}
 }
 
+func TestInsertPlanSort(t *testing.T) {
+	results := []string{
+		`{
+	"RawQuery": "insert into sbtest.A(id, b, c) values(1,2,3), (23,4,5), (65536,3,4)",
+	"Partitions": [
+		{
+			"Query": "insert into sbtest.A5(id, b, c) values (65536, 3, 4)",
+			"Backend": "backend6",
+			"Range": "[0-512)"
+		},
+		{
+			"Query": "insert into sbtest.A6(id, b, c) values (1, 2, 3), (23, 4, 5)",
+			"Backend": "backend6",
+			"Range": "[512-4096)"
+		}
+	]
+}`,
+		`{
+	"RawQuery": "insert into sbtest.A(id, b, c) values(65536,3,4), (23,4,5), (1,2,3)",
+	"Partitions": [
+		{
+			"Query": "insert into sbtest.A5(id, b, c) values (65536, 3, 4)",
+			"Backend": "backend6",
+			"Range": "[0-512)"
+		},
+		{
+			"Query": "insert into sbtest.A6(id, b, c) values (23, 4, 5), (1, 2, 3)",
+			"Backend": "backend6",
+			"Range": "[512-4096)"
+		}
+	]
+}`,
+	}
+	querys := []string{
+		"insert into sbtest.A(id, b, c) values(1,2,3), (23,4,5), (65536,3,4)",
+		"insert into sbtest.A(id, b, c) values(65536,3,4), (23,4,5), (1,2,3)",
+	}
+
+	log := xlog.NewStdLog(xlog.Level(xlog.PANIC))
+	database := "sbtest"
+
+	route, cleanup := router.MockNewRouter(log)
+	defer cleanup()
+
+	err := route.AddForTest(database, router.MockTableDeadLockConfig())
+	assert.Nil(t, err)
+	for i, query := range querys {
+		node, err := sqlparser.Parse(query)
+		assert.Nil(t, err)
+		plan := NewInsertPlan(log, database, query, node.(*sqlparser.Insert), route)
+
+		// plan build
+		{
+			err := plan.Build()
+			assert.Nil(t, err)
+			got := plan.JSON()
+			log.Info(got)
+			want := results[i]
+			assert.Equal(t, want, got)
+			plan.Type()
+			plan.Size()
+		}
+	}
+}
+
 func TestInsertUnsupportedPlan(t *testing.T) {
 	querys := []string{
 		"insert into sbtest.A(b, c, id) values(1,2)",

src/router/mock.go

@@ -105,6 +105,31 @@ func MockTableMConfig() *config.TableConfig {
 	return mock
 }
 
+// MockTableDeadLockConfig config.
+func MockTableDeadLockConfig() *config.TableConfig {
+	mock := &config.TableConfig{
+		Name:       "A",
+		ShardType:  "HASH",
+		ShardKey:   "id",
+		Partitions: make([]*config.PartitionConfig, 0, 16),
+	}
+
+	S256512 := &config.PartitionConfig{
+		Table:   "A5",
+		Segment: "0-512",
+		Backend: "backend6",
+	}
+
+	S5121024 := &config.PartitionConfig{
+		Table:   "A6",
+		Segment: "512-4096",
+		Backend: "backend6",
+	}
+
+	mock.Partitions = append(mock.Partitions,S256512, S5121024)
+	return mock
+}
+
 // MockTableBConfig config.
 func MockTableBConfig() *config.TableConfig {
 	mock := &config.TableConfig{

src/router/router_test.go

@@ -94,6 +94,9 @@ func TestRouterAdd(t *testing.T) {
 		want := "router.add.db[sbtest].table[A].exists"
 		got := err.Error()
 		assert.Equal(t, want, got)
+
+		err = router.addTable("sbtest", MockTableDeadLockConfig())
+		assert.NotNil(t, err)
 	}
 
 	// unsupport shardtype