plan: rewrite exact like expression to equal condition

expression/function_traits.go

@@ -25,6 +25,7 @@ var UnCacheableFunctions = map[string]struct{}{
 	ast.ConnectionID: {},
 	ast.LastInsertId: {},
 	ast.Version:      {},
+	ast.Like:         {},
 }
 
 // unFoldableFunctions stores functions which can not be folded duration constant folding stage.

planner/core/expression_rewriter.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/types/parser_driver"
 	"github.com/pingcap/tidb/util/chunk"
+	"github.com/pingcap/tidb/util/stringutil"
 )
 
 // EvalSubquery evaluates incorrelated subqueries once.
@@ -817,7 +818,7 @@ func (er *expressionRewriter) Leave(originInNode ast.Node) (retNode ast.Node, ok
 
 		er.ctxStack[len(er.ctxStack)-1] = expression.BuildCastFunction(er.ctx, arg, v.Tp)
 	case *ast.PatternLikeExpr:
-		er.likeToScalarFunc(v)
+		er.patternLikeToExpression(v)
 	case *ast.PatternRegexpExpr:
 		er.regexpToScalarFunc(v)
 	case *ast.RowExpr:
@@ -1140,16 +1141,44 @@ func (er *expressionRewriter) caseToExpression(v *ast.CaseExpr) {
 	er.ctxStack = append(er.ctxStack, function)
 }
 
-func (er *expressionRewriter) likeToScalarFunc(v *ast.PatternLikeExpr) {
+func (er *expressionRewriter) patternLikeToExpression(v *ast.PatternLikeExpr) {
 	l := len(er.ctxStack)
 	er.err = expression.CheckArgsNotMultiColumnRow(er.ctxStack[l-2:]...)
 	if er.err != nil {
 		return
 	}
-	escapeTp := &types.FieldType{}
-	types.DefaultTypeForValue(int(v.Escape), escapeTp)
-	function := er.notToExpression(v.Not, ast.Like, &v.Type,
-		er.ctxStack[l-2], er.ctxStack[l-1], &expression.Constant{Value: types.NewIntDatum(int64(v.Escape)), RetType: escapeTp})
+
+	var function expression.Expression
+	fieldType := &types.FieldType{}
+	isPatternExactMatch := false
+	// Treat predicate 'like' the same way as predicate '=' when it is an exact match.
+	if patExpression, ok := er.ctxStack[l-1].(*expression.Constant); ok {
+		patString, isNull, err := patExpression.EvalString(nil, chunk.Row{})
+		if err != nil {
+			er.err = errors.Trace(err)
+			return
+		}
+		if !isNull {
+			patValue, patTypes := stringutil.CompilePattern(patString, v.Escape)
+			if stringutil.IsExactMatch(patTypes) {
+				op := ast.EQ
+				if v.Not {
+					op = ast.NE
+				}
+				types.DefaultTypeForValue(string(patValue), fieldType)
+				function, er.err = er.constructBinaryOpFunction(er.ctxStack[l-2],
+					&expression.Constant{Value: types.NewStringDatum(string(patValue)), RetType: fieldType},
+					op)
+				isPatternExactMatch = true
+			}
+		}
+	}
+	if !isPatternExactMatch {
+		types.DefaultTypeForValue(int(v.Escape), fieldType)
+		function = er.notToExpression(v.Not, ast.Like, &v.Type,
+			er.ctxStack[l-2], er.ctxStack[l-1], &expression.Constant{Value: types.NewIntDatum(int64(v.Escape)), RetType: fieldType})
+	}
+
 	er.ctxStack = er.ctxStack[:l-2]
 	er.ctxStack = append(er.ctxStack, function)
 }

planner/core/logical_plan_test.go

@@ -82,7 +82,7 @@ func (s *testPlanSuite) TestPredicatePushDown(c *C) {
 		},
 		{
 			sql:  "select * from t t1, t t2 where t1.a = t2.b and t2.b > 0 and t1.a = t1.c and t1.d like 'abc' and t2.d = t1.d",
-			best: "Join{DataScan(t1)->Sel([like(cast(t1.d), abc, 92)])->DataScan(t2)->Sel([like(cast(t2.d), abc, 92)])}(t1.a,t2.b)(t1.d,t2.d)->Projection",
+			best: "Join{DataScan(t1)->Sel([eq(cast(t1.d), cast(abc))])->DataScan(t2)->Sel([eq(cast(t2.d), cast(abc))])}(t1.a,t2.b)(t1.d,t2.d)->Projection",
 		},
 		{
 			sql:  "select * from t ta join t tb on ta.d = tb.d and ta.d > 1 where tb.a = 0",

planner/core/physical_plan_test.go

@@ -1081,11 +1081,11 @@ func (s *testPlanSuite) TestRefine(c *C) {
 		},
 		{
 			sql:  "select a from t where c_str not like 'abc'",
-			best: "TableReader(Table(t)->Sel([not(like(test.t.c_str, abc, 92))]))->Projection",
+			best: `IndexReader(Index(t.c_d_e_str)[[-inf,"abc") ("abc",+inf]])->Projection`,
 		},
 		{
 			sql:  "select a from t where not (c_str like 'abc' or c_str like 'abd')",
-			best: `TableReader(Table(t)->Sel([and(not(like(test.t.c_str, abc, 92)), not(like(test.t.c_str, abd, 92)))]))->Projection`,
+			best: `IndexReader(Index(t.c_d_e_str)[[-inf,"abc") ("abc","abd") ("abd",+inf]])->Projection`,
 		},
 		{
 			sql:  "select a from t where c_str like '_abc'",
@@ -1132,10 +1132,11 @@ func (s *testPlanSuite) TestRefine(c *C) {
 			sql:  `select a from t where c_str like 123`,
 			best: "IndexReader(Index(t.c_d_e_str)[[\"123\",\"123\"]])->Projection",
 		},
-		// c is type int which will be added cast to specified type when building function signature, no index can be used.
+		// c is type int which will be added cast to specified type when building function signature,
+		// and rewrite predicate like to predicate '='  when exact match , index still can be used.
 		{
 			sql:  `select a from t where c like '1'`,
-			best: "TableReader(Table(t))->Sel([like(cast(test.t.c), 1, 92)])->Projection",
+			best: "IndexReader(Index(t.c_d_e)[[1,1]])->Projection",
 		},
 		{
 			sql:  `select a from t where c = 1.9 and d > 3`,

util/ranger/ranger_test.go

@@ -358,14 +358,14 @@ func (s *testRangerSuite) TestIndexRange(c *C) {
 		{
 			indexPos:    0,
 			exprStr:     "a LIKE 'abc'",
-			accessConds: "[like(test.t.a, abc, 92)]",
+			accessConds: "[eq(test.t.a, abc)]",
 			filterConds: "[]",
 			resultStr:   "[[\"abc\",\"abc\"]]",
 		},
 		{
 			indexPos:    0,
 			exprStr:     `a LIKE "ab\_c"`,
-			accessConds: "[like(test.t.a, ab\\_c, 92)]",
+			accessConds: "[eq(test.t.a, ab_c)]",
 			filterConds: "[]",
 			resultStr:   "[[\"ab_c\",\"ab_c\"]]",
 		},
@@ -379,14 +379,14 @@ func (s *testRangerSuite) TestIndexRange(c *C) {
 		{
 			indexPos:    0,
 			exprStr:     `a LIKE '\%a'`,
-			accessConds: "[like(test.t.a, \\%a, 92)]",
+			accessConds: "[eq(test.t.a, %a)]",
 			filterConds: "[]",
 			resultStr:   `[["%a","%a"]]`,
 		},
 		{
 			indexPos:    0,
 			exprStr:     `a LIKE "\\"`,
-			accessConds: "[like(test.t.a, \\, 92)]",
+			accessConds: "[eq(test.t.a, \\)]",
 			filterConds: "[]",
 			resultStr:   "[[\"\\\",\"\\\"]]",
 		},

util/stringutil/string_util.go

@@ -236,6 +236,16 @@ func DoMatch(str string, patChars, patTypes []byte) bool {
 	return sIdx == len(str)
 }
 
+// IsExactMatch return true if no wildcard character
+func IsExactMatch(patTypes []byte) bool {
+	for _, pt := range patTypes {
+		if pt != patMatch {
+			return false
+		}
+	}
+	return true
+}
+
 // Copy deep copies a string.
 func Copy(src string) string {
 	return string(hack.Slice(src))

util/stringutil/string_util_test.go

@@ -124,3 +124,31 @@ func (s *testStringUtilSuite) TestPatternMatch(c *C) {
 		c.Assert(match, Equals, v.match, Commentf("%v", v))
 	}
 }
+
+func (s *testStringUtilSuite) TestIsExactMatch(c *C) {
+	defer testleak.AfterTest(c)()
+	tbl := []struct {
+		pattern    string
+		escape     byte
+		exactMatch bool
+	}{
+		{``, '\\', true},
+		{`_`, '\\', false},
+		{`%`, '\\', false},
+		{`a`, '\\', true},
+		{`a_`, '\\', false},
+		{`a%`, '\\', false},
+		{`a\_`, '\\', true},
+		{`a\%`, '\\', true},
+		{`a\\`, '\\', true},
+		{`a\\_`, '\\', false},
+		{`a+%`, '+', true},
+		{`a\%`, '+', false},
+		{`a++`, '+', true},
+		{`a++_`, '+', false},
+	}
+	for _, v := range tbl {
+		_, patTypes := CompilePattern(v.pattern, v.escape)
+		c.Assert(IsExactMatch(patTypes), Equals, v.exactMatch, Commentf("%v", v))
+	}
+}