From 15be7e5cac47d93ae597671058ab34a81b2d4f64 Mon Sep 17 00:00:00 2001
From: Zhang Jian <zjsariel@gmail.com>
Date: Wed, 8 Aug 2018 17:01:21 +0800
Subject: [PATCH] plan: make `USE INDEX(`PRIMARY`)` works on the integer
 primary key (#7298) (#7316)

---
 plan/planbuilder.go      | 22 +++++++++++++++++-----
 plan/planbuilder_test.go | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+), 5 deletions(-)

diff --git a/plan/planbuilder.go b/plan/planbuilder.go
index 4fc9b10a0f3a0..d5302f93596ec 100644
--- a/plan/planbuilder.go
+++ b/plan/planbuilder.go
@@ -270,15 +270,27 @@ func (b *planBuilder) detectSelectAgg(sel *ast.SelectStmt) bool {
 	return false
 }
 
-func getPathByIndexName(paths []*accessPath, idxName model.CIStr) *accessPath {
+func getPathByIndexName(paths []*accessPath, idxName model.CIStr, tblInfo *model.TableInfo) *accessPath {
+	var tablePath *accessPath
 	for _, path := range paths {
+		if path.isTablePath {
+			tablePath = path
+			continue
+		}
 		if path.index.Name.L == idxName.L {
 			return path
 		}
 	}
+	if isPrimaryIndexHint(idxName) && tblInfo.PKIsHandle {
+		return tablePath
+	}
 	return nil
 }
 
+func isPrimaryIndexHint(indexName model.CIStr) bool {
+	return indexName.L == "primary"
+}
+
 func getPossibleAccessPaths(indexHints []*ast.IndexHint, tblInfo *model.TableInfo) ([]*accessPath, error) {
 	publicPaths := make([]*accessPath, 0, len(tblInfo.Indices)+1)
 	publicPaths = append(publicPaths, &accessPath{isTablePath: true})
@@ -298,7 +310,7 @@ func getPossibleAccessPaths(indexHints []*ast.IndexHint, tblInfo *model.TableInf
 
 		hasScanHint = true
 		for _, idxName := range hint.IndexNames {
-			path := getPathByIndexName(publicPaths[1:], idxName)
+			path := getPathByIndexName(publicPaths, idxName, tblInfo)
 			if path == nil {
 				return nil, ErrKeyDoesNotExist.GenByArgs(idxName, tblInfo.Name)
 			}
@@ -319,7 +331,7 @@ func getPossibleAccessPaths(indexHints []*ast.IndexHint, tblInfo *model.TableInf
 		available = publicPaths
 	}
 
-	available = removeIgnoredPaths(available, ignored)
+	available = removeIgnoredPaths(available, ignored, tblInfo)
 
 	// If we have got "FORCE" or "USE" index hint but got no available index,
 	// we have to use table scan.
@@ -329,13 +341,13 @@ func getPossibleAccessPaths(indexHints []*ast.IndexHint, tblInfo *model.TableInf
 	return available, nil
 }
 
-func removeIgnoredPaths(paths, ignoredPaths []*accessPath) []*accessPath {
+func removeIgnoredPaths(paths, ignoredPaths []*accessPath, tblInfo *model.TableInfo) []*accessPath {
 	if len(ignoredPaths) == 0 {
 		return paths
 	}
 	remainedPaths := make([]*accessPath, 0, len(paths))
 	for _, path := range paths {
-		if path.isTablePath || getPathByIndexName(ignoredPaths, path.index.Name) == nil {
+		if path.isTablePath || getPathByIndexName(ignoredPaths, path.index.Name, tblInfo) == nil {
 			remainedPaths = append(remainedPaths, path)
 		}
 	}
diff --git a/plan/planbuilder_test.go b/plan/planbuilder_test.go
index e5cfae3d2aac6..4e26a9b701f8f 100644
--- a/plan/planbuilder_test.go
+++ b/plan/planbuilder_test.go
@@ -16,6 +16,7 @@ package plan
 import (
 	. "github.com/pingcap/check"
 	"github.com/pingcap/tidb/ast"
+	"github.com/pingcap/tidb/model"
 )
 
 var _ = Suite(&testPlanBuilderSuite{})
@@ -56,3 +57,34 @@ func (s *testPlanBuilderSuite) TestShow(c *C) {
 		}
 	}
 }
+
+func (s *testPlanBuilderSuite) TestGetPathByIndexName(c *C) {
+	tblInfo := &model.TableInfo{
+		Indices:    make([]*model.IndexInfo, 0),
+		PKIsHandle: true,
+	}
+
+	accessPath := []*accessPath{
+		{isTablePath: true},
+		{index: &model.IndexInfo{Name: model.NewCIStr("idx")}},
+	}
+
+	path := getPathByIndexName(accessPath, model.NewCIStr("idx"), tblInfo)
+	c.Assert(path, NotNil)
+	c.Assert(path, Equals, accessPath[1])
+
+	path = getPathByIndexName(accessPath, model.NewCIStr("primary"), tblInfo)
+	c.Assert(path, NotNil)
+	c.Assert(path, Equals, accessPath[0])
+
+	path = getPathByIndexName(accessPath, model.NewCIStr("not exists"), tblInfo)
+	c.Assert(path, IsNil)
+
+	tblInfo = &model.TableInfo{
+		Indices:    make([]*model.IndexInfo, 0),
+		PKIsHandle: false,
+	}
+
+	path = getPathByIndexName(accessPath, model.NewCIStr("primary"), tblInfo)
+	c.Assert(path, IsNil)
+}