Merge branch 'release-2.1' into automated-cherry-pick-of-pingcap#13062-…

…upstream-release-2.1

cmd/explaintest/r/access_path_selection.result

@@ -19,3 +19,22 @@ IndexLookUp_11	0.00	root
 ├─IndexScan_8	1.00	cop	table:outdated_statistics, index:a, b, range:[1 1,1 1], keep order:false
 └─Selection_10	0.00	cop	eq(test.outdated_statistics.c, 1)
   └─TableScan_9	1.00	cop	table:outdated_statistics, keep order:false
+CREATE TABLE `access_path_selection` (
+`a` int,
+`b` int,
+KEY `IDX_a` (`a`),
+KEY `IDX_b` (`b`),
+KEY `IDX_ab` (`a`, `b`)
+);
+explain select a, b from access_path_selection order by _tidb_rowid;
+id	count	task	operator info
+Projection_5	10000.00	root	test.access_path_selection.a, test.access_path_selection.b
+└─TableReader_11	10000.00	root	data:TableScan_10
+  └─TableScan_10	10000.00	cop	table:access_path_selection, range:[-inf,+inf], keep order:true, stats:pseudo
+explain select max(_tidb_rowid) from access_path_selection;
+id	count	task	operator info
+StreamAgg_13	1.00	root	funcs:max(test.access_path_selection._tidb_rowid)
+└─Limit_17	1.00	root	offset:0, count:1
+  └─TableReader_27	1.00	root	data:Limit_26
+    └─Limit_26	1.00	cop	offset:0, count:1
+      └─TableScan_25	1.25	cop	table:access_path_selection, range:[-inf,+inf], keep order:true, desc, stats:pseudo

cmd/explaintest/t/access_path_selection.test

@@ -19,3 +19,14 @@ insert into outdated_statistics values (1, 3, 3);
 # This wrong case is solved by Skyline Pruning.
 analyze table outdated_statistics index idx_ab;
 explain select * from outdated_statistics where a=1 and b=1 and c=1;
+
+CREATE TABLE `access_path_selection` (
+`a` int,
+`b` int,
+KEY `IDX_a` (`a`),
+KEY `IDX_b` (`b`),
+KEY `IDX_ab` (`a`, `b`)
+);
+# _tidb_rowid should also be considered as PK.
+explain select a, b from access_path_selection order by _tidb_rowid;
+explain select max(_tidb_rowid) from access_path_selection;

ddl/index.go

@@ -139,6 +139,10 @@ func buildIndexColumns(columns []*model.ColumnInfo, idxColNames []*ast.IndexColN
 }
 
 func buildIndexInfo(tblInfo *model.TableInfo, indexName model.CIStr, idxColNames []*ast.IndexColName, state model.SchemaState) (*model.IndexInfo, error) {
+	if err := checkTooLongIndex(indexName); err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	idxColumns, err := buildIndexColumns(tblInfo.Columns, idxColNames)
 	if err != nil {
 		return nil, errors.Trace(err)

domain/info.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/ddl/util"
 	"github.com/pingcap/tidb/owner"
+	"github.com/pingcap/tidb/sessionctx/binloginfo"
 	"github.com/pingcap/tidb/util/hack"
 	"github.com/pingcap/tidb/util/logutil"
 	"github.com/pingcap/tidb/util/printer"
@@ -56,11 +57,12 @@ type InfoSyncer struct {
 // It will not be updated when tidb-server running. So please only put static information in ServerInfo struct.
 type ServerInfo struct {
 	ServerVersionInfo
-	ID         string `json:"ddl_id"`
-	IP         string `json:"ip"`
-	Port       uint   `json:"listening_port"`
-	StatusPort uint   `json:"status_port"`
-	Lease      string `json:"lease"`
+	ID           string `json:"ddl_id"`
+	IP           string `json:"ip"`
+	Port         uint   `json:"listening_port"`
+	StatusPort   uint   `json:"status_port"`
+	Lease        string `json:"lease"`
+	BinlogStatus string `json:"binlog_status"`
 }
 
 // ServerVersionInfo is the server version and git_hash.
@@ -168,7 +170,11 @@ func (is *InfoSyncer) newSessionAndStoreServerInfo(ctx context.Context, retryCnt
 		return errors.Trace(err)
 	}
 	is.session = session
-
+	binloginfo.RegisterStatusListener(func(status binloginfo.BinlogStatus) error {
+		is.info.BinlogStatus = status.String()
+		err := is.storeServerInfo(ctx)
+		return errors.Trace(err)
+	})
 	err = is.storeServerInfo(ctx)
 	return errors.Trace(err)
 }
@@ -194,7 +200,9 @@ func getInfo(ctx context.Context, etcdCli *clientv3.Client, key string, retryCnt
 			continue
 		}
 		for _, kv := range resp.Kvs {
-			info := &ServerInfo{}
+			info := &ServerInfo{
+				BinlogStatus: binloginfo.BinlogStatusUnknown.String(),
+			}
 			err = json.Unmarshal(kv.Value, info)
 			if err != nil {
 				logutil.Logger(context.Background()).Info("get key failed", zap.String("key", string(kv.Key)), zap.ByteString("value", kv.Value),
@@ -212,11 +220,12 @@ func getInfo(ctx context.Context, etcdCli *clientv3.Client, key string, retryCnt
 func getServerInfo(id string) *ServerInfo {
 	cfg := config.GetGlobalConfig()
 	info := &ServerInfo{
-		ID:         id,
-		IP:         cfg.AdvertiseAddress,
-		Port:       cfg.Port,
-		StatusPort: cfg.Status.StatusPort,
-		Lease:      cfg.Lease,
+		ID:           id,
+		IP:           cfg.AdvertiseAddress,
+		Port:         cfg.Port,
+		StatusPort:   cfg.Status.StatusPort,
+		Lease:        cfg.Lease,
+		BinlogStatus: binloginfo.GetStatus().String(),
 	}
 	info.Version = mysql.ServerVersion
 	info.GitHash = printer.TiDBGitHash

executor/ddl_test.go

@@ -449,6 +449,11 @@ func (s *testSuite) TestTooLargeIdentifierLength(c *C) {
 	tk.MustExec(fmt.Sprintf("drop index %s on t", indexName1))
 	_, err = tk.Exec(fmt.Sprintf("create index %s on t(c)", indexName2))
 	c.Assert(err.Error(), Equals, fmt.Sprintf("[ddl:1059]Identifier name '%s' is too long", indexName2))
+
+	// for create table with index.
+	tk.MustExec("drop table t;")
+	_, err = tk.Exec(fmt.Sprintf("create table t(c int, index %s(c));", indexName2))
+	c.Assert(err.Error(), Equals, fmt.Sprintf("[ddl:1059]Identifier name '%s' is too long", indexName2))
 }
 
 func (s *testSuite) TestShardRowIDBits(c *C) {

expression/chunk_executor.go

@@ -54,6 +54,27 @@ func HasGetSetVarFunc(expr Expression) bool {
 	return false
 }
 
+// HasAssignSetVarFunc checks whether an expression contains SetVar function and assign a value
+func HasAssignSetVarFunc(expr Expression) bool {
+	scalaFunc, ok := expr.(*ScalarFunction)
+	if !ok {
+		return false
+	}
+	if scalaFunc.FuncName.L == ast.SetVar {
+		for _, arg := range scalaFunc.GetArgs() {
+			if _, ok := arg.(*ScalarFunction); ok {
+				return true
+			}
+		}
+	}
+	for _, arg := range scalaFunc.GetArgs() {
+		if HasAssignSetVarFunc(arg) {
+			return true
+		}
+	}
+	return false
+}
+
 // VectorizedExecute evaluates a list of expressions column by column and append their results to "output" Chunk.
 func VectorizedExecute(ctx sessionctx.Context, exprs []Expression, iterator *chunk.Iterator4Chunk, output *chunk.Chunk) error {
 	for colID, expr := range exprs {

planner/core/integration_test.go

@@ -172,3 +172,20 @@ func (s *testIntegrationSuite) TestAntiJoinConstProp(c *C) {
 		"1 <nil>",
 	))
 }
+
+func (s *testIntegrationSuite) TestPpdWithSetVar(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 t")
+	tk.MustExec("create table t(c1 int, c2 varchar(255))")
+	tk.MustExec("insert into t values(1,'a'),(2,'d'),(3,'c')")
+
+	tk.MustQuery("select t01.c1,t01.c2,t01.c3 from (select t1.*,@c3:=@c3+1 as c3 from (select t.*,@c3:=0 from t order by t.c1)t1)t01 where t01.c3=1 and t01.c2='d'").Check(testkit.Rows())
+	tk.MustQuery("select t01.c1,t01.c2,t01.c3 from (select t1.*,@c3:=@c3+1 as c3 from (select t.*,@c3:=0 from t order by t.c1)t1)t01 where t01.c3=2 and t01.c2='d'").Check(testkit.Rows("2 d 2"))
+}

planner/core/logical_plans.go

@@ -596,6 +596,12 @@ func isColEqCorColOrConstant(filter expression.Expression, col *expression.Colum
 
 func (ds *DataSource) getPKIsHandleCol() *expression.Column {
 	if !ds.tableInfo.PKIsHandle {
+		// If the PKIsHandle is false, return the ExtraHandleColumn.
+		for i, col := range ds.Columns {
+			if col.ID == model.ExtraHandleID {
+				return ds.schema.Columns[i]
+			}
+		}
 		return nil
 	}
 	for i, col := range ds.Columns {

planner/core/rule_predicate_push_down.go

@@ -335,6 +335,12 @@ func isNullRejected(ctx sessionctx.Context, schema *expression.Schema, expr expr
 func (p *LogicalProjection) PredicatePushDown(predicates []expression.Expression) (ret []expression.Expression, retPlan LogicalPlan) {
 	canBePushed := make([]expression.Expression, 0, len(predicates))
 	canNotBePushed := make([]expression.Expression, 0, len(predicates))
+	for _, expr := range p.Exprs {
+		if expression.HasAssignSetVarFunc(expr) {
+			_, child := p.baseLogicalPlan.PredicatePushDown(nil)
+			return predicates, child
+		}
+	}
 	for _, cond := range predicates {
 		newFilter := expression.ColumnSubstitute(cond, p.Schema(), p.Exprs)
 		if !expression.HasGetSetVarFunc(newFilter) {

server/conn.go

@@ -39,7 +39,6 @@ import (
 	"crypto/tls"
 	"encoding/binary"
 	"fmt"
-	"github.com/pingcap/tidb/plugin"
 	"io"
 	"net"
 	"runtime"
@@ -57,6 +56,7 @@ import (
 	"github.com/pingcap/tidb/executor"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/metrics"
+	"github.com/pingcap/tidb/plugin"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/util/arena"
@@ -947,21 +947,23 @@ func (cc *clientConn) handleLoadStats(ctx context.Context, loadStatsInfo *execut
 // There is a special query `load data` that does not return result, which is handled differently.
 // Query `load stats` does not return result either.
 func (cc *clientConn) handleQuery(ctx context.Context, sql string) (err error) {
-	rs, err := cc.ctx.Execute(ctx, sql)
+	rss, err := cc.ctx.Execute(ctx, sql)
 	if err != nil {
 		metrics.ExecuteErrorCounter.WithLabelValues(metrics.ExecuteErrorToLabel(err)).Inc()
 		return errors.Trace(err)
 	}
 	status := atomic.LoadInt32(&cc.status)
-	if status == connStatusShutdown || status == connStatusWaitShutdown {
-		killConn(cc)
-		return errors.New("killed by another connection")
+	if rss != nil && (status == connStatusShutdown || status == connStatusWaitShutdown) {
+		for _, rs := range rss {
+			terror.Call(rs.Close)
+		}
+		return executor.ErrQueryInterrupted
 	}
-	if rs != nil {
-		if len(rs) == 1 {
-			err = cc.writeResultset(ctx, rs[0], false, 0, 0)
+	if rss != nil {
+		if len(rss) == 1 {
+			err = cc.writeResultset(ctx, rss[0], false, 0, 0)
 		} else {
-			err = cc.writeMultiResultset(ctx, rs, false)
+			err = cc.writeMultiResultset(ctx, rss, false)
 		}
 	} else {
 		loadDataInfo := cc.ctx.Value(executor.LoadDataVarKey)

server/conn_test.go

@@ -22,10 +22,14 @@ import (
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/parser/ast"
 	"github.com/pingcap/parser/mysql"
+	"github.com/pingcap/tidb/executor"
 	"github.com/pingcap/tidb/session"
+	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/mockstore"
 	"github.com/pingcap/tidb/util/arena"
+	"github.com/pingcap/tidb/util/chunk"
 )
 
 type ConnTestSuite struct{}
@@ -235,3 +239,38 @@ func (ts ConnTestSuite) TestConnExecutionTimeout(c *C) {
 
 	c.Assert(failpoint.Disable("github.com/pingcap/tidb/server/FakeClientConn"), IsNil)
 }
+
+type mockTiDBCtx struct {
+	TiDBContext
+	rs  []ResultSet
+	err error
+}
+
+func (c *mockTiDBCtx) Execute(ctx context.Context, sql string) ([]ResultSet, error) {
+	return c.rs, c.err
+}
+
+func (c *mockTiDBCtx) GetSessionVars() *variable.SessionVars {
+	return &variable.SessionVars{}
+}
+
+type mockRecordSet struct{}
+
+func (m mockRecordSet) Fields() []*ast.ResultField                       { return nil }
+func (m mockRecordSet) Next(ctx context.Context, req *chunk.Chunk) error { return nil }
+func (m mockRecordSet) NewChunk() *chunk.Chunk                           { return nil }
+func (m mockRecordSet) Close() error                                     { return nil }
+
+func (ts *ConnTestSuite) TestShutDown(c *C) {
+	cc := &clientConn{}
+
+	rs := &tidbResultSet{recordSet: mockRecordSet{}}
+	// mock delay response
+	cc.ctx = &mockTiDBCtx{rs: []ResultSet{rs}, err: nil}
+	// set killed flag
+	cc.status = connStatusShutdown
+	// assert ErrQueryInterrupted
+	err := cc.handleQuery(context.Background(), "dummy")
+	c.Assert(err, Equals, executor.ErrQueryInterrupted)
+	c.Assert(rs.closed, Equals, int32(1))
+}