From 9a16c8d7d6b32669b63be8b57efe2e1ea7d70bd9 Mon Sep 17 00:00:00 2001
From: tiancaiamao <tiancaiamao@gmail.com>
Date: Thu, 10 Jan 2019 12:17:49 +0800
Subject: [PATCH] planner/core: fix a bug that check update privilege use wrong
 `AsName` and DBName

* Fix issue 8911
* Check privilege should use table OrigName, instead of AsName
* Some col.DBName is not set,  it should be CurrentDB name
---
 planner/core/logical_plan_builder.go | 23 +++++++++++++++--------
 planner/core/logical_plan_test.go    |  7 +++++++
 session/session_test.go              | 11 +++++++++++
 3 files changed, 33 insertions(+), 8 deletions(-)

diff --git a/planner/core/logical_plan_builder.go b/planner/core/logical_plan_builder.go
index a65812350e96b..6e4eb5fbccacb 100644
--- a/planner/core/logical_plan_builder.go
+++ b/planner/core/logical_plan_builder.go
@@ -1333,7 +1333,7 @@ func (g *gbyResolver) Leave(inNode ast.Node) (ast.Node, bool) {
 
 func tblInfoFromCol(from ast.ResultSetNode, col *expression.Column) *model.TableInfo {
 	var tableList []*ast.TableName
-	tableList = extractTableList(from, tableList)
+	tableList = extractTableList(from, tableList, true)
 	for _, field := range tableList {
 		if field.Name.L == col.TblName.L {
 			return field.TableInfo
@@ -2304,7 +2304,7 @@ func (b *PlanBuilder) buildUpdate(update *ast.UpdateStmt) (Plan, error) {
 	}
 
 	var tableList []*ast.TableName
-	tableList = extractTableList(sel.From.TableRefs, tableList)
+	tableList = extractTableList(sel.From.TableRefs, tableList, false)
 	for _, t := range tableList {
 		dbName := t.Schema.L
 		if dbName == "" {
@@ -2424,7 +2424,12 @@ func (b *PlanBuilder) buildUpdateLists(tableList []*ast.TableName, list []*ast.A
 	}
 	for _, assign := range newList {
 		col := assign.Col
-		b.visitInfo = appendVisitInfo(b.visitInfo, mysql.UpdatePriv, col.DBName.L, col.TblName.L, "", nil)
+
+		dbName := col.DBName.L
+		if dbName == "" {
+			dbName = b.ctx.GetSessionVars().CurrentDB
+		}
+		b.visitInfo = appendVisitInfo(b.visitInfo, mysql.UpdatePriv, dbName, col.OrigTblName.L, "", nil)
 	}
 	return newList, p, nil
 }
@@ -2538,7 +2543,7 @@ func (b *PlanBuilder) buildDelete(delete *ast.DeleteStmt) (Plan, error) {
 	del.SetSchema(expression.NewSchema())
 
 	var tableList []*ast.TableName
-	tableList = extractTableList(delete.TableRefs.TableRefs, tableList)
+	tableList = extractTableList(delete.TableRefs.TableRefs, tableList, true)
 
 	// Collect visitInfo.
 	if delete.Tables != nil {
@@ -2757,14 +2762,16 @@ func buildWindowSpecs(specs []ast.WindowSpec) (map[string]ast.WindowSpec, error)
 }
 
 // extractTableList extracts all the TableNames from node.
-func extractTableList(node ast.ResultSetNode, input []*ast.TableName) []*ast.TableName {
+// If asName is true, extract AsName prior to OrigName.
+// Privilege check should use OrigName, while expression may use AsName.
+func extractTableList(node ast.ResultSetNode, input []*ast.TableName, asName bool) []*ast.TableName {
 	switch x := node.(type) {
 	case *ast.Join:
-		input = extractTableList(x.Left, input)
-		input = extractTableList(x.Right, input)
+		input = extractTableList(x.Left, input, asName)
+		input = extractTableList(x.Right, input, asName)
 	case *ast.TableSource:
 		if s, ok := x.Source.(*ast.TableName); ok {
-			if x.AsName.L != "" {
+			if x.AsName.L != "" && asName {
 				newTableName := *s
 				newTableName.Name = x.AsName
 				s.Name = x.AsName
diff --git a/planner/core/logical_plan_test.go b/planner/core/logical_plan_test.go
index 677b4ad2a74ce..b45f738681abc 100644
--- a/planner/core/logical_plan_test.go
+++ b/planner/core/logical_plan_test.go
@@ -1385,6 +1385,13 @@ func (s *testPlanSuite) TestVisitInfo(c *C) {
 				{mysql.SelectPriv, "test", "t", "", nil},
 			},
 		},
+		{
+			sql: "update t a1 set a1.a = a1.a + 1",
+			ans: []visitInfo{
+				{mysql.UpdatePriv, "test", "t", "", nil},
+				{mysql.SelectPriv, "test", "t", "", nil},
+			},
+		},
 		{
 			sql: "select a, sum(e) from t group by a",
 			ans: []visitInfo{
diff --git a/session/session_test.go b/session/session_test.go
index df77ed3c401fc..b7185580bdc28 100644
--- a/session/session_test.go
+++ b/session/session_test.go
@@ -2457,6 +2457,17 @@ func (s *testSessionSuite) TestUpdatePrivilege(c *C) {
 	// In fact, the privlege check for t1 should be update, and for t2 should be select.
 	_, err = tk1.Exec("update t1,t2 set t1.id = t2.id;")
 	c.Assert(err, IsNil)
+
+	// Fix issue 8911
+	tk.MustExec("create database weperk")
+	tk.MustExec("use weperk")
+	tk.MustExec("create table tb_wehub_server (id int, active_count int, used_count int)")
+	tk.MustExec("create user 'weperk'")
+	tk.MustExec("grant all privileges on weperk.* to 'weperk'@'%'")
+	c.Assert(tk1.Se.Auth(&auth.UserIdentity{Username: "weperk", Hostname: "%"},
+		[]byte(""), []byte("")), IsTrue)
+	tk1.MustExec("use weperk")
+	tk1.MustExec("update tb_wehub_server a set a.active_count=a.active_count+1,a.used_count=a.used_count+1 where id=1")
 }
 
 func (s *testSessionSuite) TestTxnGoString(c *C) {