planner: update cost model2

executor/explain_test.go

@@ -364,7 +364,7 @@ func TestCheckActRowsWithUnistore(t *testing.T) {
 		},
 		{
 			sql:      "select count(*) from t_unistore_act_rows group by b",
-			expected: []string{"2", "2", "2", "4"},
+			expected: []string{"2", "4", "4"},
 		},
 		{
 			sql:      "with cte(a) as (select a from t_unistore_act_rows) select (select 1 from cte limit 1) from cte;",

executor/explainfor_test.go

@@ -462,12 +462,12 @@ func TestPointGetUserVarPlanCache(t *testing.T) {
 	tk.Session().SetSessionManager(&testkit.MockSessionManager{PS: ps})
 	tk.MustQuery(fmt.Sprintf("explain for connection %d", tkProcess.ID)).Check(testkit.Rows( // can use idx_a
 		`Projection_9 1.00 root  test.t1.a, test.t1.b, test.t2.a, test.t2.b`,
-		`└─IndexJoin_17 1.00 root  inner join, inner:TableReader_13, outer key:test.t2.a, inner key:test.t1.a, equal cond:eq(test.t2.a, test.t1.a)`,
-		`  ├─Selection_44(Build) 0.80 root  not(isnull(test.t2.a))`,
-		`  │ └─Point_Get_43 1.00 root table:t2, index:idx_a(a) `,
-		`  └─TableReader_13(Probe) 0.00 root  data:Selection_12`,
-		`    └─Selection_12 0.00 cop[tikv]  eq(test.t1.a, 1)`,
-		`      └─TableRangeScan_11 0.80 cop[tikv] table:t1 range: decided by [eq(test.t1.a, test.t2.a)], keep order:false, stats:pseudo`))
+		`└─MergeJoin_10 1.00 root  inner join, left key:test.t2.a, right key:test.t1.a`,
+		`  ├─Selection_42(Build) 10.00 root  eq(test.t1.a, 1)`,
+		`  │ └─TableReader_41 10.00 root  data:TableRangeScan_40`,
+		`  │   └─TableRangeScan_40 10.00 cop[tikv] table:t1 range:[1,1], keep order:true, stats:pseudo`,
+		`  └─Selection_39(Probe) 0.80 root  not(isnull(test.t2.a))`,
+		`    └─Point_Get_38 1.00 root table:t2, index:idx_a(a) `))
 
 	tk.MustExec("set @a=2")
 	tk.MustQuery("execute stmt using @a").Check(testkit.Rows(
@@ -478,12 +478,12 @@ func TestPointGetUserVarPlanCache(t *testing.T) {
 	tk.Session().SetSessionManager(&testkit.MockSessionManager{PS: ps})
 	tk.MustQuery(fmt.Sprintf("explain for connection %d", tkProcess.ID)).Check(testkit.Rows( // can use idx_a
 		`Projection_9 1.00 root  test.t1.a, test.t1.b, test.t2.a, test.t2.b`,
-		`└─IndexJoin_17 1.00 root  inner join, inner:TableReader_13, outer key:test.t2.a, inner key:test.t1.a, equal cond:eq(test.t2.a, test.t1.a)`,
-		`  ├─Selection_44(Build) 0.80 root  not(isnull(test.t2.a))`,
-		`  │ └─Point_Get_43 1.00 root table:t2, index:idx_a(a) `,
-		`  └─TableReader_13(Probe) 0.00 root  data:Selection_12`,
-		`    └─Selection_12 0.00 cop[tikv]  eq(test.t1.a, 2)`,
-		`      └─TableRangeScan_11 0.80 cop[tikv] table:t1 range: decided by [eq(test.t1.a, test.t2.a)], keep order:false, stats:pseudo`))
+		`└─MergeJoin_10 1.00 root  inner join, left key:test.t2.a, right key:test.t1.a`,
+		`  ├─Selection_42(Build) 10.00 root  eq(test.t1.a, 2)`,
+		`  │ └─TableReader_41 10.00 root  data:TableRangeScan_40`,
+		`  │   └─TableRangeScan_40 10.00 cop[tikv] table:t1 range:[2,2], keep order:true, stats:pseudo`,
+		`  └─Selection_39(Probe) 0.80 root  not(isnull(test.t2.a))`,
+		`    └─Point_Get_38 1.00 root table:t2, index:idx_a(a) `))
 	tk.MustQuery("execute stmt using @a").Check(testkit.Rows(
 		"2 4 2 2",
 	))

executor/join_test.go

@@ -1479,7 +1479,7 @@ func TestIndexNestedLoopHashJoin(t *testing.T) {
 		"      └─TableRowIDScan 27.00 cop[tikv] table:l2 keep order:false"))
 	tk.MustQuery("select * from t l1 where exists ( select * from t l2 where l2.l_orderkey = l1.l_orderkey and l2.l_suppkey <> l1.l_suppkey )order by `l_orderkey`,`l_linenumber`;").Check(testkit.Rows("0 0 0 0", "0 1 0 1", "0 2 0 0", "1 0 1 0", "1 1 1 1", "1 2 1 0", "2 0 0 0", "2 1 0 1", "2 2 0 0"))
 	tk.MustQuery("desc format = 'brief' select count(*) from t l1 where exists ( select * from t l2 where l2.l_orderkey = l1.l_orderkey and l2.l_suppkey <> l1.l_suppkey );").Check(testkit.Rows(
-		"HashAgg 1.00 root  funcs:count(1)->Column#11",
+		"StreamAgg 1.00 root  funcs:count(1)->Column#11",
 		"└─IndexHashJoin 7.20 root  semi join, inner:IndexLookUp, outer key:test.t.l_orderkey, inner key:test.t.l_orderkey, equal cond:eq(test.t.l_orderkey, test.t.l_orderkey), other cond:ne(test.t.l_suppkey, test.t.l_suppkey)",
 		"  ├─TableReader(Build) 9.00 root  data:Selection",
 		"  │ └─Selection 9.00 cop[tikv]  not(isnull(test.t.l_suppkey))",

executor/partition_table_test.go

@@ -3670,11 +3670,11 @@ func TestPartitionTableExplain(t *testing.T) {
 		"  └─IndexFullScan 1.00 cop[tikv] table:t, partition:p1, index:b(b) keep order:false"))
 	tk.MustQuery(`explain format = 'brief' select * from t,t2 where t2.a = 1 and t2.b = t.b and t.a = 1`).Check(testkit.Rows(
 		"HashJoin 1.00 root  inner join, equal:[eq(testpartitiontableexplain.t.b, testpartitiontableexplain.t2.b)]",
-		"├─TableReader(Build) 1.00 root  data:Selection",
-		"│ └─Selection 1.00 cop[tikv]  eq(testpartitiontableexplain.t2.a, 1), not(isnull(testpartitiontableexplain.t2.b))",
-		"│   └─TableFullScan 3.00 cop[tikv] table:t2 keep order:false",
-		"└─Selection(Probe) 1.00 root  not(isnull(testpartitiontableexplain.t.b))",
-		"  └─Point_Get 1.00 root table:t, partition:p1 handle:1"))
+		`├─Selection(Build) 1.00 root  not(isnull(testpartitiontableexplain.t.b))`,
+		`│ └─Point_Get 1.00 root table:t, partition:p1 handle:1`,
+		`└─TableReader(Probe) 1.00 root  data:Selection`,
+		`  └─Selection 1.00 cop[tikv]  eq(testpartitiontableexplain.t2.a, 1), not(isnull(testpartitiontableexplain.t2.b))`,
+		`    └─TableFullScan 3.00 cop[tikv] table:t2 keep order:false`))
 
 	tk.MustExec("set @@tidb_partition_prune_mode = 'dynamic'")
 	tk.MustExec(`analyze table t`)

executor/testdata/executor_suite_out.json

@@ -674,7 +674,7 @@
       {
         "SQL": "select count(*) from t as t1 inner join t as t2 on t1.c1 = t2.c1 where t1.c1 != NULL",
         "Plan": [
-          "HashAgg 1.00 root  funcs:count(1)->Column#7",
+          "StreamAgg 1.00 root  funcs:count(1)->Column#7",
           "└─MergeJoin 0.00 root  inner join, left key:test.t.c1, right key:test.t.c1",
           "  ├─TableDual(Build) 0.00 root  rows:0",
           "  └─TableDual(Probe) 0.00 root  rows:0"
@@ -704,7 +704,7 @@
       {
         "SQL": "select count(*) from t as t1 left join t as t2 on t1.c1 = t2.c1 where t1.c1 != NULL",
         "Plan": [
-          "HashAgg 1.00 root  funcs:count(1)->Column#7",
+          "StreamAgg 1.00 root  funcs:count(1)->Column#7",
           "└─MergeJoin 0.00 root  left outer join, left key:test.t.c1, right key:test.t.c1",
           "  ├─TableDual(Build) 0.00 root  rows:0",
           "  └─TableDual(Probe) 0.00 root  rows:0"

planner/core/integration_test.go

@@ -1231,9 +1231,9 @@ func TestAggPushDownEngine(t *testing.T) {
 	tk.MustExec("set @@session.tidb_isolation_read_engines = 'tiflash'")
 
 	tk.MustQuery("explain format = 'brief' select approx_count_distinct(a) from t").Check(testkit.Rows(
-		"HashAgg 1.00 root  funcs:approx_count_distinct(Column#4)->Column#3",
-		"└─TableReader 1.00 root  data:HashAgg",
-		"  └─HashAgg 1.00 batchCop[tiflash]  funcs:approx_count_distinct(test.t.a)->Column#4",
+		"StreamAgg 1.00 root  funcs:approx_count_distinct(Column#5)->Column#3",
+		"└─TableReader 1.00 root  data:StreamAgg",
+		"  └─StreamAgg 1.00 batchCop[tiflash]  funcs:approx_count_distinct(test.t.a)->Column#5",
 		"    └─TableFullScan 10000.00 batchCop[tiflash] table:t keep order:false, stats:pseudo"))
 
 	tk.MustExec("set @@session.tidb_isolation_read_engines = 'tikv'")
@@ -6695,7 +6695,7 @@ func TestAggPushToCopForCachedTable(t *testing.T) {
 	tk.MustExec("alter table t32157 cache")
 
 	tk.MustQuery("explain format = 'brief' select /*+AGG_TO_COP()*/ count(*) from t32157 ignore index(primary) where process_code = 'GDEP0071'").Check(testkit.Rows(
-		"HashAgg 1.00 root  funcs:count(1)->Column#8]\n" +
+		"StreamAgg 1.00 root  funcs:count(1)->Column#8]\n" +
 			"[└─UnionScan 10.00 root  eq(test.t32157.process_code, \"GDEP0071\")]\n" +
 			"[  └─TableReader 10.00 root  data:Selection]\n" +
 			"[    └─Selection 10.00 cop[tikv]  eq(test.t32157.process_code, \"GDEP0071\")]\n" +
@@ -7963,10 +7963,10 @@ func TestNullConditionForPrefixIndex(t *testing.T) {
 	ps := []*util.ProcessInfo{tkProcess}
 	tk.Session().SetSessionManager(&testkit.MockSessionManager{PS: ps})
 	tk.MustQuery(fmt.Sprintf("explain for connection %d", tkProcess.ID)).Check(testkit.Rows(
-		"HashAgg_12 1.00 root  funcs:count(Column#6)->Column#5",
-		"└─IndexReader_13 1.00 root  index:HashAgg_6",
-		"  └─HashAgg_6 1.00 cop[tikv]  funcs:count(1)->Column#6",
-		"    └─IndexRangeScan_11 99.90 cop[tikv] table:t1, index:idx2(c1, c2) range:[\"0xfff\" -inf,\"0xfff\" +inf], keep order:false, stats:pseudo"))
+		"StreamAgg_18 1.00 root  funcs:count(Column#7)->Column#5",
+		"└─IndexReader_19 1.00 root  index:StreamAgg_9",
+		"  └─StreamAgg_9 1.00 cop[tikv]  funcs:count(1)->Column#7",
+		"    └─IndexRangeScan_17 99.90 cop[tikv] table:t1, index:idx2(c1, c2) range:[\"0xfff\" -inf,\"0xfff\" +inf], keep order:false, stats:pseudo"))
 }
 
 func TestAutoIncrementCheckWithCheckConstraint(t *testing.T) {

planner/core/physical_plan_trace_test.go

@@ -61,7 +61,7 @@ func TestPhysicalOptimizeWithTraceEnabled(t *testing.T) {
 				"Limit_20",
 				"IndexReader_21",
 				"Limit_14",
-				"HashAgg_9",
+				"StreamAgg_10",
 				"Projection_8",
 			},
 		},

planner/core/plan_cost_ver2.go

@@ -256,14 +256,17 @@ func (p *PhysicalIndexLookUpReader) getPlanCostVer2(taskType property.TaskType,
 	}
 	tableSideCost := divCostVer2(sumCostVer2(tableNetCost, tableChildCost), distConcurrency)
 
-	// double-read: assume at least 1 row to double-read to avoid 0 double-read cost.
-	doubleReadRows := math.Max(indexRows, 1)
+	doubleReadRows := indexRows
 	doubleReadCPUCost := newCostVer2(option, cpuFactor,
 		indexRows*cpuFactor.Value,
 		func() string { return fmt.Sprintf("double-read-cpu(%v*%v)", doubleReadRows, cpuFactor) })
 	batchSize := float64(p.ctx.GetSessionVars().IndexLookupSize)
 	taskPerBatch := 32.0 // TODO: remove this magic number
 	doubleReadTasks := doubleReadRows / batchSize * taskPerBatch
+	if doubleReadRows <= 1 {
+		// prefer to use IndexLookup if double-read data-size is very small
+		doubleReadTasks = 0
+	}
 	doubleReadRequestCost := doubleReadCostVer2(option, doubleReadTasks, requestFactor)
 	doubleReadCost := sumCostVer2(doubleReadCPUCost, doubleReadRequestCost)
 
@@ -448,13 +451,16 @@ func (p *PhysicalHashAgg) getPlanCostVer2(taskType property.TaskType, option *Pl
 	groupCost := groupCostVer2(option, inputRows, p.GroupByItems, cpuFactor)
 	hashBuildCost := hashBuildCostVer2(option, outputRows, outputRowSize, float64(len(p.GroupByItems)), cpuFactor, memFactor)
 	hashProbeCost := hashProbeCostVer2(option, inputRows, float64(len(p.GroupByItems)), cpuFactor)
+	startCost := newCostVer2(option, cpuFactor,
+		10*3*cpuFactor.Value,
+		func() string { return fmt.Sprintf("cpu(10*3*%v)", cpuFactor) })
 
 	childCost, err := p.children[0].getPlanCostVer2(taskType, option)
 	if err != nil {
 		return zeroCostVer2, err
 	}
 
-	p.planCostVer2 = sumCostVer2(childCost, divCostVer2(sumCostVer2(aggCost, groupCost, hashBuildCost, hashProbeCost), concurrency))
+	p.planCostVer2 = sumCostVer2(startCost, childCost, divCostVer2(sumCostVer2(aggCost, groupCost, hashBuildCost, hashProbeCost), concurrency))
 	p.planCostInit = true
 	return p.planCostVer2, nil
 }
@@ -532,7 +538,10 @@ func (p *PhysicalHashJoin) getPlanCostVer2(taskType property.TaskType, option *P
 		p.planCostVer2 = sumCostVer2(buildChildCost, probeChildCost,
 			divCostVer2(sumCostVer2(buildHashCost, buildFilterCost, probeHashCost, probeFilterCost), mppConcurrency))
 	} else { // TiDB HashJoin
-		p.planCostVer2 = sumCostVer2(buildChildCost, probeChildCost, buildHashCost, buildFilterCost,
+		startCost := newCostVer2(option, cpuFactor,
+			10*3*cpuFactor.Value,
+			func() string { return fmt.Sprintf("cpu(10*3*%v)", cpuFactor) })
+		p.planCostVer2 = sumCostVer2(startCost, buildChildCost, probeChildCost, buildHashCost, buildFilterCost,
 			divCostVer2(sumCostVer2(probeFilterCost, probeHashCost), tidbConcurrency))
 	}
 	p.planCostInit = true
@@ -563,6 +572,9 @@ func (p *PhysicalIndexJoin) getIndexJoinCostVer2(taskType property.TaskType, opt
 	buildTaskCost := newCostVer2(option, cpuFactor,
 		buildRows*10*cpuFactor.Value,
 		func() string { return fmt.Sprintf("cpu(%v*10*%v)", buildRows, cpuFactor) })
+	startCost := newCostVer2(option, cpuFactor,
+		10*3*cpuFactor.Value,
+		func() string { return fmt.Sprintf("cpu(10*3*%v)", cpuFactor) })
 
 	probeFilterCost := filterCostVer2(option, probeRowsTot, probeFilters, cpuFactor)
 	probeChildCost, err := probe.getPlanCostVer2(taskType, option)
@@ -587,7 +599,7 @@ func (p *PhysicalIndexJoin) getIndexJoinCostVer2(taskType property.TaskType, opt
 	batchRatio := 6.0
 	probeCost := divCostVer2(mulCostVer2(probeChildCost, buildRows), batchRatio)
 
-	p.planCostVer2 = sumCostVer2(buildChildCost, buildFilterCost, buildTaskCost, divCostVer2(sumCostVer2(probeCost, probeFilterCost, hashTableCost), probeConcurrency))
+	p.planCostVer2 = sumCostVer2(startCost, buildChildCost, buildFilterCost, buildTaskCost, divCostVer2(sumCostVer2(probeCost, probeFilterCost, hashTableCost), probeConcurrency))
 	p.planCostInit = true
 	return p.planCostVer2, nil
 }

planner/core/point_get_plan_test.go

@@ -189,7 +189,7 @@ func TestGetExtraColumn(t *testing.T) {
 	tk.MustQuery(`explain format='brief' select t.*, _tidb_rowid from t where a = 1`).Check(testkit.Rows(`Point_Get 1.00 root table:t, index:idx(a) `))
 	tk.MustExec(`commit`)
 	tk.MustQuery(`explain format='brief' select count(_tidb_rowid) from t where a=1`).Check(testkit.Rows(
-		`HashAgg 1.00 root  funcs:count(test.t._tidb_rowid)->Column#4`,
+		`StreamAgg 1.00 root  funcs:count(test.t._tidb_rowid)->Column#4`,
 		`└─Point_Get 1.00 root table:t, index:idx(a) `))
 	tk.MustQuery(`explain format='brief' select *, date_format(b, "") from t where a =1 for update`).Check(testkit.Rows(
 		`Projection 1.00 root  test.t.a, test.t.b, date_format(cast(test.t.b, datetime BINARY), )->Column#4`,

planner/core/testdata/analyze_suite_out.json

@@ -383,7 +383,7 @@
         "└─Apply 10.00 root  CARTESIAN left outer semi join, other cond:eq(test.t.c, Column#21)",
         "  ├─IndexReader(Build) 10.00 root  index:IndexFullScan",
         "  │ └─IndexFullScan 10.00 cop[tikv] table:t, index:idx(c, b, a) keep order:false",
-        "  └─HashAgg(Probe) 10.00 root  funcs:count(1)->Column#21",
+        "  └─StreamAgg(Probe) 10.00 root  funcs:count(1)->Column#21",
         "    └─HashJoin 10.00 root  inner join, equal:[eq(test.t.a, test.t.a)]",
         "      ├─IndexReader(Build) 10.00 root  index:Selection",
         "      │ └─Selection 10.00 cop[tikv]  eq(test.t.a, test.t.a), not(isnull(test.t.a))",
@@ -411,7 +411,7 @@
       {
         "SQL": "explain format = 'brief' select max(e) from t where a='T3382' and b='ECO' and c='TOPIC' and d='23660fa1ace9455cb7f3ee831e14a342'",
         "Plan": [
-          "HashAgg 1.00 root  funcs:max(test.t.e)->Column#7",
+          "StreamAgg 1.00 root  funcs:max(test.t.e)->Column#7",
           "└─TopN 0.00 root  test.t.e:desc, offset:0, count:1",
           "  └─IndexLookUp 0.00 root  ",
           "    ├─IndexRangeScan(Build) 0.01 cop[tikv] table:t, index:idx1(d, a) range:[\"23660fa1ace9455cb7f3ee831e14a342\" \"T3382\",\"23660fa1ace9455cb7f3ee831e14a342\" \"T3382\"], keep order:false",
@@ -435,7 +435,7 @@
     "Cases": [
       "IndexReader(Index(t.e)[[NULL,+inf]]->StreamAgg)->StreamAgg",
       "IndexReader(Index(t.e)[[-inf,10]]->StreamAgg)->StreamAgg",
-      "IndexReader(Index(t.e)[[-inf,50]]->HashAgg)->HashAgg",
+      "IndexReader(Index(t.e)[[-inf,50]]->StreamAgg)->StreamAgg",
       "IndexReader(Index(t.b_c)[[NULL,+inf]]->Sel([gt(test.t.c, 1)])->StreamAgg)->StreamAgg",
       "IndexLookUp(Index(t.e)[[1,1]], Table(t))->HashAgg",
       "TableReader(Table(t)->Sel([gt(test.t.e, 1)])->HashAgg)->HashAgg",

planner/core/testdata/binary_plan_suite_out.json

@@ -123,23 +123,23 @@
                         "operator_info": "data:TableFullScan_16"
                       }
                     ],
-                    "cost": 975351.9825195674,
+                    "cost": 976848.9825195674,
                     "est_rows": 100000000,
                     "act_rows": 8,
                     "task_type": 1,
                     "store_type": 1,
                     "operator_info": "CARTESIAN inner join"
                   }
                 ],
-                "cost": 998975351.9825196,
+                "cost": 998976848.9825196,
                 "est_rows": 100000000,
                 "act_rows": 8,
                 "task_type": 1,
                 "store_type": 1,
                 "operator_info": "cast(test.t.a, decimal(10,0) BINARY)->Column#8"
               }
             ],
-            "cost": 1996975363.5625196,
+            "cost": 1996978357.5625196,
             "est_rows": 1,
             "act_rows": 1,
             "task_type": 1,
@@ -283,7 +283,7 @@
                 "operator_info": "data:Selection_33"
               }
             ],
-            "cost": 1798740.0717058014,
+            "cost": 1800237.0717058014,
             "est_rows": 12487.5,
             "task_type": 1,
             "store_type": 1,