Make index serial scan concurrency configurable and fix golint errors. (

pingcap#2928)

executor/distsql.go

@@ -595,7 +595,7 @@ func (e *XSelectIndexExec) doIndexRequest() (distsql.SelectResult, error) {
 	} else if !e.indexPlan.OutOfOrder {
 		// The cost of index scan double-read is higher than single-read. Usually ordered index scan has a limit
 		// which may not have been pushed down, so we set concurrency to 1 to avoid fetching unnecessary data.
-		e.scanConcurrency = 1
+		e.scanConcurrency = e.ctx.GetSessionVars().IndexSerialScanConcurrency
 	}
 	fieldTypes := make([]*types.FieldType, len(e.indexPlan.Index.Columns))
 	for i, v := range e.indexPlan.Index.Columns {

plan/logical_plan_builder.go

@@ -848,7 +848,7 @@ func (b *planBuilder) unfoldWildStar(p LogicalPlan, selectFields []*ast.SelectFi
 }
 
 func (b *planBuilder) pushTableHints(hints []*ast.TableOptimizerHint) bool {
-	sortMergeTables := make([]model.CIStr, 0)
+	var sortMergeTables []model.CIStr
 	for _, hint := range hints {
 		switch hint.HintName.L {
 		case TiDBMergeJoin:

plan/physical_plan_builder.go

@@ -584,8 +584,8 @@ func compareTypeForOrder(lhs *types.FieldType, rhs *types.FieldType) bool {
 // Generate all possible combinations from join conditions for cost evaluation
 // It will try all keys in join conditions
 func constructPropertyByJoin(join *Join) ([][]*requiredProperty, []int, error) {
-	result := make([][]*requiredProperty, 0)
-	condIndex := make([]int, 0)
+	var result [][]*requiredProperty
+	var condIndex []int
 	if join.EqualConditions == nil {
 		return nil, nil, nil
 	}

session.go

@@ -945,6 +945,7 @@ const loadCommonGlobalVarsSQL = "select * from mysql.global_variables where vari
 	variable.TiDBSkipDDLWait + quoteCommaQuote +
 	variable.TiDBIndexLookupSize + quoteCommaQuote +
 	variable.TiDBIndexLookupConcurrency + quoteCommaQuote +
+	variable.TiDBIndexSerialScanConcurrency + quoteCommaQuote +
 	variable.TiDBDistSQLScanConcurrency + "')"
 
 // LoadCommonGlobalVariableIfNeeded loads and applies commonly used global variables for the session.

sessionctx/variable/session.go

@@ -177,25 +177,29 @@ type SessionVars struct {
 
 	// The number of concurrent dist SQL scan worker.
 	DistSQLScanConcurrency int
+
+	// The number of concurrent index serial scan worker.
+	IndexSerialScanConcurrency int
 }
 
 // NewSessionVars creates a session vars object.
 func NewSessionVars() *SessionVars {
 	return &SessionVars{
-		Users:                    make(map[string]string),
-		Systems:                  make(map[string]string),
-		PreparedStmts:            make(map[uint32]interface{}),
-		PreparedStmtNameToID:     make(map[string]uint32),
-		TxnCtx:                   &TransactionContext{},
-		RetryInfo:                &RetryInfo{},
-		StrictSQLMode:            true,
-		Status:                   mysql.ServerStatusAutocommit,
-		StmtCtx:                  new(StatementContext),
-		AllowAggPushDown:         true,
-		BuildStatsConcurrencyVar: DefBuildStatsConcurrency,
-		IndexLookupSize:          DefIndexLookupSize,
-		IndexLookupConcurrency:   DefIndexLookupConcurrency,
-		DistSQLScanConcurrency:   DefDistSQLScanConcurrency,
+		Users:                      make(map[string]string),
+		Systems:                    make(map[string]string),
+		PreparedStmts:              make(map[uint32]interface{}),
+		PreparedStmtNameToID:       make(map[string]uint32),
+		TxnCtx:                     &TransactionContext{},
+		RetryInfo:                  &RetryInfo{},
+		StrictSQLMode:              true,
+		Status:                     mysql.ServerStatusAutocommit,
+		StmtCtx:                    new(StatementContext),
+		AllowAggPushDown:           true,
+		BuildStatsConcurrencyVar:   DefBuildStatsConcurrency,
+		IndexLookupSize:            DefIndexLookupSize,
+		IndexLookupConcurrency:     DefIndexLookupConcurrency,
+		IndexSerialScanConcurrency: DefIndexSerialScanConcurrency,
+		DistSQLScanConcurrency:     DefDistSQLScanConcurrency,
 	}
 }
 

sessionctx/variable/sysvar.go

@@ -599,6 +599,7 @@ var defaultSysVars = []*SysVar{
 	{ScopeGlobal | ScopeSession, TiDBDistSQLScanConcurrency, strconv.Itoa(DefDistSQLScanConcurrency)},
 	{ScopeGlobal | ScopeSession, TiDBIndexLookupSize, strconv.Itoa(DefIndexLookupSize)},
 	{ScopeGlobal | ScopeSession, TiDBIndexLookupConcurrency, strconv.Itoa(DefIndexLookupConcurrency)},
+	{ScopeGlobal | ScopeSession, TiDBIndexSerialScanConcurrency, strconv.Itoa(DefIndexSerialScanConcurrency)},
 	{ScopeGlobal | ScopeSession, TiDBSkipDDLWait, boolToIntStr(DefSkipDDLWait)},
 	{ScopeGlobal | ScopeSession, TiDBSkipUTF8Check, boolToIntStr(DefSkipUTF8Check)},
 }

sessionctx/variable/tidb_vars.go

@@ -70,6 +70,10 @@ const (
 	// Set this value higher may reduce the latency but consumes more system resource.
 	TiDBIndexLookupConcurrency = "tidb_index_lookup_concurrency"
 
+	// tidb_index_serial_scan_concurrency is used for controling the concurrency of index scan operation
+	// when we need to keep the data output order the same as the order of index data.
+	TiDBIndexSerialScanConcurrency = "tidb_index_serial_scan_concurrency"
+
 	// tidb_skip_ddl_wait skips the wait tiem of two lease after executing CREATE TABLE statement.
 	// When we have multiple TiDB servers in a cluster, the newly created table may not be available on all TiDB server
 	// until two lease time later, set this value to true will reduce the time to create a table, with the risk that
@@ -83,12 +87,13 @@ const (
 
 // Default TiDB system variable values.
 const (
-	DefIndexLookupConcurrency = 4
-	DefIndexLookupSize        = 20000
-	DefDistSQLScanConcurrency = 10
-	DefBuildStatsConcurrency  = 4
-	DefSkipDDLWait            = false
-	DefSkipUTF8Check          = false
-	DefOptAggPushDown         = true
-	DefOptInSubqUnfolding     = false
+	DefIndexLookupConcurrency     = 4
+	DefIndexSerialScanConcurrency = 1
+	DefIndexLookupSize            = 20000
+	DefDistSQLScanConcurrency     = 10
+	DefBuildStatsConcurrency      = 4
+	DefSkipDDLWait                = false
+	DefSkipUTF8Check              = false
+	DefOptAggPushDown             = true
+	DefOptInSubqUnfolding         = false
 )

sessionctx/varsutil/varsutil.go

@@ -130,6 +130,8 @@ func SetSessionSystemVar(vars *variable.SessionVars, name string, value types.Da
 		vars.IndexLookupSize = tidbOptPositiveInt(sVal, variable.DefIndexLookupSize)
 	case variable.TiDBDistSQLScanConcurrency:
 		vars.DistSQLScanConcurrency = tidbOptPositiveInt(sVal, variable.DefDistSQLScanConcurrency)
+	case variable.TiDBIndexSerialScanConcurrency:
+		vars.IndexSerialScanConcurrency = tidbOptPositiveInt(sVal, variable.DefIndexSerialScanConcurrency)
 	}
 	vars.Systems[name] = sVal
 	return nil

sessionctx/varsutil/varsutil_test.go

@@ -137,6 +137,11 @@ func (s *testVarsutilSuite) TestVarsutil(c *C) {
 	// Combined sql_mode
 	SetSessionSystemVar(v, "sql_mode", types.NewStringDatum("REAL_AS_FLOAT,ANSI_QUOTES"))
 	c.Assert(v.SQLMode, Equals, mysql.ModeRealAsFloat|mysql.ModeANSIQuotes)
+
+	// Test case for tidb_index_serial_scan_concurrency.
+	c.Assert(v.IndexSerialScanConcurrency, Equals, 1)
+	SetSessionSystemVar(v, variable.TiDBIndexSerialScanConcurrency, types.NewStringDatum("4"))
+	c.Assert(v.IndexSerialScanConcurrency, Equals, 4)
 }
 
 type mockGlobalAccessor struct {