Merge branch 'release-3.1' into automated-cherry-pick-of-pingcap#12173-…

…upstream-release-3.1

cmd/explaintest/r/tpch.result

@@ -1223,7 +1223,7 @@ id	count	task	operator info
 Projection_25	1.00	root	tpch.supplier.s_name, 17_col_0
 └─TopN_28	1.00	root	17_col_0:desc, tpch.supplier.s_name:asc, offset:0, count:100
   └─HashAgg_34	1.00	root	group by:tpch.supplier.s_name, funcs:count(1), firstrow(tpch.supplier.s_name)
-    └─IndexJoin_40	7828961.66	root	anti semi join, inner:IndexLookUp_39, outer key:tpch.l1.l_orderkey, inner key:tpch.l3.l_orderkey, other cond:ne(tpch.l3.l_suppkey, tpch.l1.l_suppkey), ne(tpch.l3.l_suppkey, tpch.supplier.s_suppkey)
+    └─IndexJoin_40	7828961.66	root	anti semi join, inner:IndexLookUp_39, outer key:tpch.l1.l_orderkey, inner key:tpch.l3.l_orderkey, other cond:ne(tpch.l3.l_suppkey, tpch.l1.l_suppkey)
       ├─IndexJoin_56	9786202.08	root	semi join, inner:IndexLookUp_55, outer key:tpch.l1.l_orderkey, inner key:tpch.l2.l_orderkey, other cond:ne(tpch.l2.l_suppkey, tpch.l1.l_suppkey), ne(tpch.l2.l_suppkey, tpch.supplier.s_suppkey)
       │ ├─IndexJoin_62	12232752.60	root	inner join, inner:TableReader_61, outer key:tpch.l1.l_orderkey, inner key:tpch.orders.o_orderkey
       │ │ ├─HashRightJoin_66	12232752.60	root	inner join, inner:HashRightJoin_72, equal:[eq(tpch.supplier.s_suppkey, tpch.l1.l_suppkey)]

executor/show.go

@@ -274,7 +274,7 @@ func (e *ShowExec) fetchShowProcessList() error {
 		if !hasProcessPriv && pi.User != loginUser.Username {
 			continue
 		}
-		row := pi.ToRow(e.Full)
+		row := pi.ToRowForShow(e.Full)
 		e.appendRow(row)
 	}
 	return nil

expression/expression.go

@@ -276,7 +276,7 @@ func EvaluateExprWithNull(ctx sessionctx.Context, schema *Schema, expr Expressio
 		for i, arg := range x.GetArgs() {
 			args[i] = EvaluateExprWithNull(ctx, schema, arg)
 		}
-		return NewFunctionInternal(ctx, x.FuncName.L, types.NewFieldType(mysql.TypeTiny), args...)
+		return NewFunctionInternal(ctx, x.FuncName.L, x.RetType, args...)
 	case *Column:
 		if !schema.Contains(x) {
 			return x

infoschema/tables.go

@@ -543,6 +543,7 @@ var tableProcesslistCols = []columnInfo{
 	{"TIME", mysql.TypeLong, 7, mysql.NotNullFlag, 0, nil},
 	{"STATE", mysql.TypeVarchar, 7, 0, nil, nil},
 	{"INFO", mysql.TypeString, 512, 0, nil, nil},
+	{"MEM", mysql.TypeLonglong, 21, 0, nil, nil},
 }
 
 var tableTiDBIndexesCols = []columnInfo{
@@ -862,7 +863,7 @@ func dataForProcesslist(ctx sessionctx.Context) [][]types.Datum {
 			continue
 		}
 
-		rows := pi.ToRow(true)
+		rows := pi.ToRow()
 		record := types.MakeDatums(rows...)
 		records = append(records, record)
 	}

infoschema/tables_test.go

@@ -131,6 +131,7 @@ func (s *testTableSuite) TestInfoschemaFieldValue(c *C) {
 		User:    "root",
 		Host:    "127.0.0.1",
 		Command: mysql.ComQuery,
+		StmtCtx: tk.Se.GetSessionVars().StmtCtx,
 	}
 	tk.Se.SetSessionManager(sm)
 	tk.MustQuery("SELECT user,host,command FROM information_schema.processlist;").Check(testkit.Rows("root 127.0.0.1 Query"))
@@ -342,19 +343,35 @@ func (s *testTableSuite) TestSomeTables(c *C) {
 		DB:      "information_schema",
 		Command: byte(1),
 		State:   1,
-		Info:    "do something"}
+		Info:    "do something",
+		StmtCtx: tk.Se.GetSessionVars().StmtCtx,
+	}
 	sm.processInfoMap[2] = &util.ProcessInfo{
 		ID:      2,
 		User:    "user-2",
 		Host:    "localhost",
 		DB:      "test",
 		Command: byte(2),
 		State:   2,
-		Info:    "do something"}
+		Info:    strings.Repeat("x", 101),
+		StmtCtx: tk.Se.GetSessionVars().StmtCtx,
+	}
 	tk.Se.SetSessionManager(sm)
-	tk.MustQuery("select * from information_schema.PROCESSLIST order by ID;").Check(
-		testkit.Rows("1 user-1 localhost information_schema Quit 9223372036 1 do something",
-			"2 user-2 localhost test Init DB 9223372036 2 do something"))
+	tk.MustQuery("select * from information_schema.PROCESSLIST order by ID;").Sort().Check(
+		testkit.Rows(
+			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 1 %s 0", "do something"),
+			fmt.Sprintf("2 user-2 localhost test Init DB 9223372036 2 %s 0", strings.Repeat("x", 101)),
+		))
+	tk.MustQuery("SHOW PROCESSLIST;").Sort().Check(
+		testkit.Rows(
+			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 1 %s", "do something"),
+			fmt.Sprintf("2 user-2 localhost test Init DB 9223372036 2 %s", strings.Repeat("x", 100)),
+		))
+	tk.MustQuery("SHOW FULL PROCESSLIST;").Sort().Check(
+		testkit.Rows(
+			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 1 %s", "do something"),
+			fmt.Sprintf("2 user-2 localhost test Init DB 9223372036 2 %s", strings.Repeat("x", 101)),
+		))
 
 	sm = &mockSessionManager{make(map[uint64]*util.ProcessInfo, 2)}
 	sm.processInfoMap[1] = &util.ProcessInfo{
@@ -380,8 +397,8 @@ func (s *testTableSuite) TestSomeTables(c *C) {
 	tk.Se.SetSessionManager(sm)
 	tk.MustQuery("select * from information_schema.PROCESSLIST order by ID;").Check(
 		testkit.Rows(
-			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 1 %s", "<nil>"),
-			fmt.Sprintf("2 user-2 localhost <nil> Init DB 9223372036 2 %s", strings.Repeat("x", 101)),
+			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 1 %s 0", "<nil>"),
+			fmt.Sprintf("2 user-2 localhost <nil> Init DB 9223372036 2 %s 0", strings.Repeat("x", 101)),
 		))
 	tk.MustQuery("SHOW PROCESSLIST;").Sort().Check(
 		testkit.Rows(
@@ -395,11 +412,11 @@ func (s *testTableSuite) TestSomeTables(c *C) {
 		))
 	tk.MustQuery("select * from information_schema.PROCESSLIST where db is null;").Check(
 		testkit.Rows(
-			fmt.Sprintf("2 user-2 localhost <nil> Init DB 9223372036 2 %s", strings.Repeat("x", 101)),
+			fmt.Sprintf("2 user-2 localhost <nil> Init DB 9223372036 2 %s 0", strings.Repeat("x", 101)),
 		))
 	tk.MustQuery("select * from information_schema.PROCESSLIST where Info is null;").Check(
 		testkit.Rows(
-			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 1 %s", "<nil>"),
+			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 1 %s 0", "<nil>"),
 		))
 }
 

planner/core/integration_test.go

@@ -122,3 +122,70 @@ func (s *testIntegrationSuite) TestApplyNotNullFlag(c *C) {
 
 	tk.MustQuery("select IFNULL((select t1.x from t1 where t1.x = t2.x), 'xxx') as col1 from t2").Check(testkit.Rows("xxx"))
 }
+
+func (s *testIntegrationSuite) TestSimplifyOuterJoinWithCast(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int not null, b datetime default null)")
+
+	var input []string
+	var output []struct {
+		SQL  string
+		Plan []string
+	}
+	s.testData.GetTestCases(c, &input, &output)
+	for i, tt := range input {
+		s.testData.OnRecord(func() {
+			output[i].SQL = tt
+			output[i].Plan = s.testData.ConvertRowsToStrings(tk.MustQuery(tt).Rows())
+		})
+		tk.MustQuery(tt).Check(testkit.Rows(output[i].Plan...))
+	}
+}
+
+func (s *testIntegrationSuite) TestAntiJoinConstProp(c *C) {
+	store, dom, err := newStoreWithBootstrap()
+	c.Assert(err, IsNil)
+	tk := testkit.NewTestKit(c, store)
+	defer func() {
+		dom.Close()
+		store.Close()
+	}()
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t1, t2")
+	tk.MustExec("create table t1(a int not null, b int not null)")
+	tk.MustExec("insert into t1 values (1,1)")
+	tk.MustExec("create table t2(a int not null, b int not null)")
+	tk.MustExec("insert into t2 values (2,2)")
+
+	tk.MustQuery("select * from t1 where t1.a not in (select a from t2 where t2.a = t1.a and t2.a > 1)").Check(testkit.Rows(
+		"1 1",
+	))
+	tk.MustQuery("select * from t1 where t1.a not in (select a from t2 where t2.b = t1.b and t2.a > 1)").Check(testkit.Rows(
+		"1 1",
+	))
+	tk.MustQuery("select * from t1 where t1.a not in (select a from t2 where t2.b = t1.b and t2.b > 1)").Check(testkit.Rows(
+		"1 1",
+	))
+	tk.MustQuery("select q.a in (select count(*) from t1 s where not exists (select 1 from t1 p where q.a > 1 and p.a = s.a)) from t1 q").Check(testkit.Rows(
+		"1",
+	))
+	tk.MustQuery("select q.a in (select not exists (select 1 from t1 p where q.a > 1 and p.a = s.a) from t1 s) from t1 q").Check(testkit.Rows(
+		"1",
+	))
+
+	tk.MustExec("drop table t1, t2")
+	tk.MustExec("create table t1(a int not null, b int)")
+	tk.MustExec("insert into t1 values (1,null)")
+	tk.MustExec("create table t2(a int not null, b int)")
+	tk.MustExec("insert into t2 values (2,2)")
+
+	tk.MustQuery("select * from t1 where t1.a not in (select a from t2 where t2.b > t1.b)").Check(testkit.Rows(
+		"1 <nil>",
+	))
+	tk.MustQuery("select * from t1 where t1.a not in (select a from t2 where t1.a = 2)").Check(testkit.Rows(
+		"1 <nil>",
+	))
+}

planner/core/logical_plan_builder.go

@@ -226,9 +226,14 @@ func (p *LogicalJoin) pushDownConstExpr(expr expression.Expression, leftCond []e
 		} else {
 			leftCond = append(leftCond, expr)
 		}
-	case SemiJoin, AntiSemiJoin, InnerJoin:
+	case SemiJoin, InnerJoin:
 		leftCond = append(leftCond, expr)
 		rightCond = append(rightCond, expr)
+	case AntiSemiJoin:
+		if filterCond {
+			leftCond = append(leftCond, expr)
+		}
+		rightCond = append(rightCond, expr)
 	}
 	return leftCond, rightCond
 }
@@ -255,18 +260,13 @@ func (p *LogicalJoin) extractOnCondition(conditions []expression.Expression, der
 					arg0, arg1 = arg1, arg0
 				}
 				if leftCol != nil && rightCol != nil {
-					// Do not derive `is not null` for anti join, since it may cause wrong results.
-					// For example:
-					// `select * from t t1 where t1.a not in (select b from t t2)` does not imply `t2.b is not null`,
-					// `select * from t t1 where t1.a not in (select a from t t2 where t1.b = t2.b` does not imply `t1.b is not null`,
-					// `select * from t t1 where not exists (select * from t t2 where t2.a = t1.a)` does not imply `t1.a is not null`,
-					if deriveLeft && p.JoinType != AntiSemiJoin {
+					if deriveLeft {
 						if isNullRejected(ctx, left.Schema(), expr) && !mysql.HasNotNullFlag(leftCol.RetType.Flag) {
 							notNullExpr := expression.BuildNotNullExpr(ctx, leftCol)
 							leftCond = append(leftCond, notNullExpr)
 						}
 					}
-					if deriveRight && p.JoinType != AntiSemiJoin {
+					if deriveRight {
 						if isNullRejected(ctx, right.Schema(), expr) && !mysql.HasNotNullFlag(rightCol.RetType.Flag) {
 							notNullExpr := expression.BuildNotNullExpr(ctx, rightCol)
 							rightCond = append(rightCond, notNullExpr)

planner/core/logical_plan_test.go

@@ -445,7 +445,7 @@ func (s *testPlanSuite) TestAntiSemiJoinConstFalse(c *C) {
 	}{
 		{
 			sql:      "select a from t t1 where not exists (select a from t t2 where t1.a = t2.a and t2.b = 1 and t2.b = 2)",
-			best:     "Join{DataScan(t1)->DataScan(t2)}->Projection",
+			best:     "Join{DataScan(t1)->DataScan(t2)}(test.t1.a,test.t2.a)->Projection",
 			joinType: "anti semi join",
 		},
 	}

planner/core/logical_plans.go

@@ -327,7 +327,7 @@ type LogicalUnionScan struct {
 	conditions []expression.Expression
 }
 
-// DataSource represents a tablescan without condition push down.
+// DataSource represents a tableScan without condition push down.
 type DataSource struct {
 	logicalSchemaProducer
 

planner/core/rule_predicate_push_down.go

@@ -149,7 +149,7 @@ func (p *LogicalJoin) PredicatePushDown(predicates []expression.Expression) (ret
 		p.LeftConditions = nil
 		ret = append(expression.ScalarFuncs2Exprs(equalCond), otherCond...)
 		ret = append(ret, leftPushCond...)
-	case SemiJoin, AntiSemiJoin, InnerJoin:
+	case SemiJoin, InnerJoin:
 		tempCond := make([]expression.Expression, 0, len(p.LeftConditions)+len(p.RightConditions)+len(p.EqualConditions)+len(p.OtherConditions)+len(predicates))
 		tempCond = append(tempCond, p.LeftConditions...)
 		tempCond = append(tempCond, p.RightConditions...)
@@ -158,13 +158,10 @@ func (p *LogicalJoin) PredicatePushDown(predicates []expression.Expression) (ret
 		tempCond = append(tempCond, predicates...)
 		tempCond = expression.ExtractFiltersFromDNFs(p.ctx, tempCond)
 		tempCond = expression.PropagateConstant(p.ctx, tempCond)
-		// Return table dual when filter is constant false or null. Not applicable to AntiSemiJoin.
-		// TODO: For AntiSemiJoin, we can use outer plan to substitute LogicalJoin actually.
-		if p.JoinType != AntiSemiJoin {
-			dual := conds2TableDual(p, tempCond)
-			if dual != nil {
-				return ret, dual
-			}
+		// Return table dual when filter is constant false or null.
+		dual := conds2TableDual(p, tempCond)
+		if dual != nil {
+			return ret, dual
 		}
 		equalCond, leftPushCond, rightPushCond, otherCond = p.extractOnCondition(tempCond, true, true)
 		p.LeftConditions = nil
@@ -173,6 +170,24 @@ func (p *LogicalJoin) PredicatePushDown(predicates []expression.Expression) (ret
 		p.OtherConditions = otherCond
 		leftCond = leftPushCond
 		rightCond = rightPushCond
+	case AntiSemiJoin:
+		predicates = expression.PropagateConstant(p.ctx, predicates)
+		// Return table dual when filter is constant false or null.
+		dual := conds2TableDual(p, predicates)
+		if dual != nil {
+			return ret, dual
+		}
+		// `predicates` should only contain left conditions or constant filters.
+		_, leftPushCond, rightPushCond, _ = p.extractOnCondition(predicates, true, true)
+		// Do not derive `is not null` for anti join, since it may cause wrong results.
+		// For example:
+		// `select * from t t1 where t1.a not in (select b from t t2)` does not imply `t2.b is not null`,
+		// `select * from t t1 where t1.a not in (select a from t t2 where t1.b = t2.b` does not imply `t1.b is not null`,
+		// `select * from t t1 where not exists (select * from t t2 where t2.a = t1.a)` does not imply `t1.a is not null`,
+		leftCond = leftPushCond
+		rightCond = append(p.RightConditions, rightPushCond...)
+		p.RightConditions = nil
+
 	}
 	leftCond = expression.RemoveDupExprs(p.ctx, leftCond)
 	rightCond = expression.RemoveDupExprs(p.ctx, rightCond)

planner/core/testdata/integration_suite_in.json

@@ -4,12 +4,12 @@
     "cases": [
       // Limit should be pushed down into IndexLookUpReader, row count of IndexLookUpReader and TableScan should be 1.00.
       "explain select * from tbl use index(idx_b_c) where b > 1 limit 2,1",
-	  // Projection atop IndexLookUpReader, Limit should be pushed down into IndexLookUpReader, and Projection should have row count 1.00 as well.
+      // Projection atop IndexLookUpReader, Limit should be pushed down into IndexLookUpReader, and Projection should have row count 1.00 as well.
       "explain select * from tbl use index(idx_b_c) where b > 1 order by b desc limit 2,1",
-	  // Limit should be pushed down into IndexLookUpReader when Selection on top of IndexScan.
-	  "explain select * from tbl use index(idx_b_c) where b > 1 and c > 1 limit 2,1",
-	  // Limit should NOT be pushed down into IndexLookUpReader when Selection on top of TableScan.
-	  "explain select * from tbl use index(idx_b_c) where b > 1 and a > 1 limit 2,1"
+      // Limit should be pushed down into IndexLookUpReader when Selection on top of IndexScan.
+      "explain select * from tbl use index(idx_b_c) where b > 1 and c > 1 limit 2,1",
+      // Limit should NOT be pushed down into IndexLookUpReader when Selection on top of TableScan.
+      "explain select * from tbl use index(idx_b_c) where b > 1 and a > 1 limit 2,1"
     ]
   },
   {
@@ -18,5 +18,12 @@
       // fix #12385
       "explain select * from t t1 left join t t2 on t1.a=t2.a where from_unixtime(t2.b);"
     ]
+  },
+  {
+    "name": "TestSimplifyOuterJoinWithCast",
+    "cases": [
+      // LeftOuterJoin should no be simplified to InnerJoin.
+      "explain select * from t t1 left join t t2 on t1.a = t2.a where cast(t1.b as date) >= '2019-01-01'"
+    ]
   }
 ]

planner/core/testdata/integration_suite_out.json

@@ -20,5 +20,21 @@
         ]
       }
     ]
+  },
+  {
+    "Name": "TestSimplifyOuterJoinWithCast",
+    "Cases": [
+      {
+        "SQL": "explain select * from t t1 left join t t2 on t1.a = t2.a where cast(t1.b as date) >= '2019-01-01'",
+        "Plan": [
+          "HashLeftJoin_8 10000.00 root left outer join, inner:TableReader_13, equal:[eq(test.t1.a, test.t2.a)]",
+          "├─Selection_9 8000.00 root ge(cast(test.t1.b), 2019-01-01 00:00:00.000000)",
+          "│ └─TableReader_11 10000.00 root data:TableScan_10",
+          "│   └─TableScan_10 10000.00 cop table:t1, range:[-inf,+inf], keep order:false, stats:pseudo",
+          "└─TableReader_13 10000.00 root data:TableScan_12",
+          "  └─TableScan_12 10000.00 cop table:t2, range:[-inf,+inf], keep order:false, stats:pseudo"
+        ]
+      }
+    ]
   }
 ]

session/session.go

@@ -98,7 +98,7 @@ var (
 	sessionExecuteParseDurationGeneral    = metrics.SessionExecuteParseDuration.WithLabelValues(metrics.LblGeneral)
 )
 
-// Session context
+// Session context, it is consistent with the lifecycle of a client connection.
 type Session interface {
 	sessionctx.Context
 	Status() uint16                                               // Flag of current status, such as autocommit.