Merge pull request pingcap#4 from windtalker/hanfei/join-merge

basic cbo support for broadcast join

Author: windtalker

planner/core/exhaust_physical_plans.go

@@ -1479,13 +1479,15 @@ func (p *LogicalJoin) exhaustPhysicalPlans(prop *property.PhysicalProperty) []Ph
 	return joins
 }
 
-func getAllDataSourceRowCount(plan LogicalPlan) int64 {
+func getAllDataSourceTotalRowSize(plan LogicalPlan) float64 {
 	if ds, ok := plan.(*DataSource); ok {
-		return ds.statsInfo().Count()
+		rowCount := ds.statsInfo().Count()
+		rowSize := ds.TblColHists.GetTableAvgRowSize(ds.ctx, ds.schema.Columns, kv.StoreType(ds.preferStoreType), ds.handleCol != nil)
+		return float64(rowCount) * rowSize
 	}
-	ret := int64(0)
+	ret := float64(0)
 	for _, child := range plan.Children() {
-		ret += getAllDataSourceRowCount(child)
+		ret += getAllDataSourceTotalRowSize(child)
 	}
 	return ret
 }
@@ -1511,28 +1513,30 @@ func (p *LogicalJoin) tryToGetBroadCastJoin(prop *property.PhysicalProperty) []P
 		LeftJoinKeys:    lkeys,
 		RightJoinKeys:   rkeys,
 	}
-	// todo: currently, build side is the one has less rowcont and global read side
-	//  is the one has less datasource row count(which mean less remote read), need
-	//  to use cbo to decide the build side and global read side
-	if p.children[0].statsInfo().Count() < p.children[1].statsInfo().Count() {
-		baseJoin.InnerChildIdx = 0
-	} else {
-		baseJoin.InnerChildIdx = 1
+
+	preferredBuildIndex := 0
+	if p.children[0].statsInfo().Count() > p.children[1].statsInfo().Count() {
+		preferredBuildIndex = 1
 	}
-	globalIndex := baseJoin.InnerChildIdx
-	if prop.TaskTp != property.CopTiFlashGlobalReadTaskType && getAllDataSourceRowCount(p.children[globalIndex]) > getAllDataSourceRowCount(p.children[1-globalIndex]) {
-		globalIndex = 1 - globalIndex
+	preferredGlobalIndex := preferredBuildIndex
+	if prop.TaskTp != property.CopTiFlashGlobalReadTaskType && getAllDataSourceTotalRowSize(p.children[preferredGlobalIndex]) > getAllDataSourceTotalRowSize(p.children[1 - preferredGlobalIndex]) {
+		preferredGlobalIndex = 1 - preferredGlobalIndex
 	}
+	// todo: currently, build side is the one has less rowcount and global read side
+	//  is the one has less datasource row size(which mean less remote read), need
+	//  to use cbo to decide the build side and global read side if preferred build index
+	//  is not equal to preferred global index
+	baseJoin.InnerChildIdx = preferredBuildIndex
 	childrenReqProps := make([]*property.PhysicalProperty, 2)
-	childrenReqProps[globalIndex] = &property.PhysicalProperty{TaskTp: property.CopTiFlashGlobalReadTaskType}
+	childrenReqProps[preferredGlobalIndex] = &property.PhysicalProperty{TaskTp: property.CopTiFlashGlobalReadTaskType}
 	if prop.TaskTp == property.CopTiFlashGlobalReadTaskType {
-		childrenReqProps[1-globalIndex] = &property.PhysicalProperty{TaskTp: property.CopTiFlashGlobalReadTaskType}
+		childrenReqProps[1-preferredGlobalIndex] = &property.PhysicalProperty{TaskTp: property.CopTiFlashGlobalReadTaskType}
 	} else {
-		childrenReqProps[1-globalIndex] = &property.PhysicalProperty{TaskTp: property.CopTiFlashLocalReadTaskType}
+		childrenReqProps[1-preferredGlobalIndex] = &property.PhysicalProperty{TaskTp: property.CopTiFlashLocalReadTaskType}
 	}
 	join := PhysicalBroadCastJoin{
 		basePhysicalJoin: baseJoin,
-		globalChildIndex: globalIndex,
+		globalChildIndex: preferredGlobalIndex,
 	}.Init(p.ctx, p.stats, p.blockOffset, childrenReqProps...)
 	results := make([]PhysicalPlan, 0, 1)
 	results = append(results, join)

planner/core/find_best_task.go

@@ -1345,6 +1345,9 @@ func (ds *DataSource) getOriginalPhysicalTableScan(prop *property.PhysicalProper
 	}
 	sessVars := ds.ctx.GetSessionVars()
 	cost := rowCount * rowSize * sessVars.ScanFactor
+	if ts.IsGlobalRead {
+		cost += rowCount * sessVars.NetworkFactor * rowSize
+	}
 	if isMatchProp {
 		if prop.Items[0].Desc {
 			ts.Desc = true

planner/core/integration_test.go

@@ -341,6 +341,68 @@ func (s *testIntegrationSerialSuite) TestSelPushDownTiFlash(c *C) {
 	}
 }
 
+func (s *testIntegrationSerialSuite) TestBroadcastJoin(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists d1_t")
+	tk.MustExec("create table d1_t(d1_k int, value int)")
+	tk.MustExec("insert into d1_t values(1,2),(2,3)")
+	tk.MustExec("analyze table d1_t")
+	tk.MustExec("drop table if exists d2_t")
+	tk.MustExec("create table d2_t(d2_k decimal(10,2), value int)")
+	tk.MustExec("insert into d2_t values(10.11,2),(10.12,3)")
+	tk.MustExec("analyze table d2_t")
+	tk.MustExec("drop table if exists d3_t")
+	tk.MustExec("create table d3_t(d3_k date, value int)")
+	tk.MustExec("insert into d3_t values(date'2010-01-01',2),(date'2010-01-02',3)")
+	tk.MustExec("analyze table d3_t")
+	tk.MustExec("drop table if exists fact_t")
+	tk.MustExec("create table fact_t(d1_k int, d2_k decimal(10,2), d3_k date, col1 int, col2 int, col3 int)")
+	tk.MustExec("insert into fact_t values(1,10.11,date'2010-01-01',1,2,3),(1,10.11,date'2010-01-02',1,2,3),(1,10.12,date'2010-01-01',1,2,3),(1,10.12,date'2010-01-02',1,2,3)")
+	tk.MustExec("insert into fact_t values(2,10.11,date'2010-01-01',1,2,3),(2,10.11,date'2010-01-02',1,2,3),(2,10.12,date'2010-01-01',1,2,3),(2,10.12,date'2010-01-02',1,2,3)")
+	tk.MustExec("analyze table fact_t")
+
+	// Create virtual tiflash replica info.
+	dom := domain.GetDomain(tk.Se)
+	is := dom.InfoSchema()
+	db, exists := is.SchemaByName(model.NewCIStr("test"))
+	c.Assert(exists, IsTrue)
+	for _, tblInfo := range db.Tables {
+		if tblInfo.Name.L == "fact_t" || tblInfo.Name.L == "d1_t" || tblInfo.Name.L == "d2_t" || tblInfo.Name.L == "d3_t" {
+			tblInfo.TiFlashReplica = &model.TiFlashReplicaInfo{
+				Count:     1,
+				Available: true,
+			}
+		}
+	}
+
+	tk.MustExec("set @@session.tidb_isolation_read_engines = 'tiflash'")
+	tk.MustExec("set @@session.tidb_opt_broadcast_join = 1")
+	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())
+		})
+		res := tk.MustQuery(tt)
+		res.Check(testkit.Rows(output[i].Plan...))
+	}
+
+	// out join not supported
+	_, err := tk.Exec("explain select /*+ tidb_bcj(fact_t, d1_t) */ count(*) from fact_t left join d1_t on fact_t.d1_k = d1_t.d1_k")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[planner:1815]Internal : Can't find a proper physical plan for this query")
+	// join with non-equal condition not supported
+	_, err = tk.Exec("explain select /*+ tidb_bcj(fact_t, d1_t) */ count(*) from fact_t join d1_t on fact_t.d1_k = d1_t.d1_k and fact_t.col1 > d1_t.value")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[planner:1815]Internal : Can't find a proper physical plan for this query")
+}
+
 func (s *testIntegrationSerialSuite) TestIssue15110(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")

planner/core/task.go

@@ -28,6 +28,7 @@ import (
 	"github.com/pingcap/tidb/planner/property"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/statistics"
+	"github.com/pingcap/tidb/store/tikv"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/chunk"
 	"github.com/pingcap/tidb/util/plancodec"
@@ -533,9 +534,45 @@ func (p *PhysicalHashJoin) attach2Task(tasks ...task) task {
 	return task
 }
 
+// GetCost computes cost of broadcast join operator itself.
+func (p *PhysicalBroadCastJoin) GetCost(lCnt, rCnt float64) float64 {
+	buildCnt := lCnt
+	if p.InnerChildIdx == 1 {
+		buildCnt = rCnt
+	}
+	sessVars := p.ctx.GetSessionVars()
+	// Cost of building hash table.
+	cpuCost := buildCnt * sessVars.CopCPUFactor
+	memoryCost := buildCnt * sessVars.MemoryFactor
+	// Number of matched row pairs regarding the equal join conditions.
+	helper := &fullJoinRowCountHelper{
+		cartesian:     false,
+		leftProfile:   p.children[0].statsInfo(),
+		rightProfile:  p.children[1].statsInfo(),
+		leftJoinKeys:  p.LeftJoinKeys,
+		rightJoinKeys: p.RightJoinKeys,
+		leftSchema:    p.children[0].Schema(),
+		rightSchema:   p.children[1].Schema(),
+	}
+	numPairs := helper.estimate()
+	probeCost := numPairs * sessVars.CopCPUFactor
+	// should divided by the cop concurrency, which is decide by TiFlash, but TiDB
+	// can not get the information from TiFlash, so just use `sessVars.HashJoinConcurrency`
+	// as a workaround
+	probeCost /= float64(sessVars.HashJoinConcurrency)
+	cpuCost += probeCost
+
+	// todo since TiFlash join is significant faster than TiDB join, maybe
+	//  need to add a variable like 'tiflash_accelerate_factor', and divide
+	//  the final cost by that factor
+	return cpuCost + memoryCost
+}
+
 func (p *PhysicalBroadCastJoin) attach2Task(tasks ...task) task {
 	lTask, lok := tasks[0].(*copTask)
 	rTask, rok := tasks[1].(*copTask)
+	lGlobalRead := p.childrenReqProps[0].TaskTp == property.CopTiFlashGlobalReadTaskType
+	rGlobalRead := p.childrenReqProps[1].TaskTp == property.CopTiFlashGlobalReadTaskType
 	if !lok || !rok || (lTask.getStoreType() != kv.TiFlash && rTask.getStoreType() != kv.TiFlash) {
 		return invalidTask
 	}
@@ -547,11 +584,30 @@ func (p *PhysicalBroadCastJoin) attach2Task(tasks ...task) task {
 	if !rTask.indexPlanFinished {
 		rTask.finishIndexPlan()
 	}
-	task := &copTask{
-		tblColHists:       rTask.tblColHists,
+
+	lCost := lTask.cost()
+	rCost := rTask.cost()
+	if !(lGlobalRead && rGlobalRead) {
+		// the cost model for top level broadcast join is
+		// globalReadSideCost * copTaskNumber + localReadSideCost + broadcast operator cost
+		// because for broadcast join, the global side is executed in every copTask.
+		copTaskNumber := int32(1)
+		copClient, ok := p.ctx.GetClient().(*tikv.CopClient)
+		if ok {
+			copTaskNumber = copClient.GetBatchCopTaskNumber()
+		}
+		if lGlobalRead {
+			lCost = lCost * float64(copTaskNumber)
+		} else {
+			rCost = rCost * float64(copTaskNumber)
+		}
+	}
+
+	task := & copTask {
+		tblColHists: rTask.tblColHists,
 		indexPlanFinished: true,
-		tablePlan:         p,
-		cst:               lTask.cost() + rTask.cost(),
+		tablePlan: p,
+		cst:  lCost + rCost + p.GetCost(lTask.count(), rTask.count()),
 	}
 	logutil.BgLogger().Info("bc join cost", zap.Float64("bc cost", task.cst))
 	return task
@@ -708,6 +764,11 @@ func finishCopTask(ctx sessionctx.Context, task task) task {
 	// is Min(DistSQLScanConcurrency, numRegionsInvolvedInScan), since we cannot infer
 	// the number of regions involved, we simply use DistSQLScanConcurrency.
 	copIterWorkers := float64(t.plan().SCtx().GetSessionVars().DistSQLScanConcurrency)
+	if t.tablePlan != nil && t.tablePlan.TP() == plancodec.TypeBroadcastJoin {
+		if copClient, ok := ctx.GetClient().(*tikv.CopClient); ok {
+			copIterWorkers = math.Min(float64(copClient.GetBatchCopTaskNumber()), copIterWorkers)
+		}
+	}
 	t.finishIndexPlan()
 	// Network cost of transferring rows of table scan to TiDB.
 	if t.tablePlan != nil {

planner/core/testdata/integration_serial_suite_in.json

@@ -6,6 +6,13 @@
       "explain select * from t where cast(t.a as float) + 3 = 5.1"
     ]
   },
+  {
+    "name": "TestBroadcastJoin",
+    "cases": [
+      "explain select /*+ tidb_bcj(fact_t,d1_t) */ count(*) from fact_t, d1_t where fact_t.d1_k = d1_t.d1_k",
+      "explain select /*+ tidb_bcj(fact_t,d1_t,d2_t,d3_t) */ count(*) from fact_t, d1_t, d2_t, d3_t where fact_t.d1_k = d1_t.d1_k and fact_t.d2_k = d2_t.d2_k and fact_t.d3_k = d3_t.d3_k"
+    ]
+  },
   {
     "name": "TestReadFromStorageHint",
     "cases": [

planner/core/testdata/integration_serial_suite_out.json

@@ -20,6 +20,43 @@
       }
     ]
   },
+  {
+    "Name": "TestBroadcastJoin",
+    "Cases": [
+      {
+        "SQL": "explain select /*+ tidb_bcj(fact_t,d1_t) */ count(*) from fact_t, d1_t where fact_t.d1_k = d1_t.d1_k",
+        "Plan": [
+          "StreamAgg_25 1.00 root  funcs:count(Column#14)->Column#11",
+          "└─TableReader_26 1.00 root  data:StreamAgg_13",
+          "  └─StreamAgg_13 1.00 cop[tiflash]  funcs:count(1)->Column#14",
+          "    └─TypeBroadcastJoin_24 8.00 cop[tiflash]  ",
+          "      ├─Selection_18(Build) 0.00 cop[tiflash]  not(isnull(test.d1_t.d1_k))",
+          "      │ └─TableFullScan_17 0.00 cop[tiflash] table:d1_t keep order:false, global read",
+          "      └─Selection_16(Probe) 0.00 cop[tiflash]  not(isnull(test.fact_t.d1_k))",
+          "        └─TableFullScan_15 0.00 cop[tiflash] table:fact_t keep order:false"
+        ]
+      },
+      {
+        "SQL": "explain select /*+ tidb_bcj(fact_t,d1_t,d2_t,d3_t) */ count(*) from fact_t, d1_t, d2_t, d3_t where fact_t.d1_k = d1_t.d1_k and fact_t.d2_k = d2_t.d2_k and fact_t.d3_k = d3_t.d3_k",
+        "Plan": [
+          "StreamAgg_35 1.00 root  funcs:count(Column#20)->Column#17",
+          "└─TableReader_36 1.00 root  data:StreamAgg_17",
+          "  └─StreamAgg_17 1.00 cop[tiflash]  funcs:count(1)->Column#20",
+          "    └─TypeBroadcastJoin_34 8.00 cop[tiflash]  ",
+          "      ├─Selection_28(Build) 0.00 cop[tiflash]  not(isnull(test.d3_t.d3_k))",
+          "      │ └─TableFullScan_27 0.00 cop[tiflash] table:d3_t keep order:false, global read",
+          "      └─TypeBroadcastJoin_19(Probe) 8.00 cop[tiflash]  ",
+          "        ├─Selection_26(Build) 0.00 cop[tiflash]  not(isnull(test.d2_t.d2_k))",
+          "        │ └─TableFullScan_25 0.00 cop[tiflash] table:d2_t keep order:false, global read",
+          "        └─TypeBroadcastJoin_20(Probe) 8.00 cop[tiflash]  ",
+          "          ├─Selection_24(Build) 0.00 cop[tiflash]  not(isnull(test.d1_t.d1_k))",
+          "          │ └─TableFullScan_23 0.00 cop[tiflash] table:d1_t keep order:false, global read",
+          "          └─Selection_22(Probe) 0.00 cop[tiflash]  not(isnull(test.fact_t.d1_k)), not(isnull(test.fact_t.d2_k)), not(isnull(test.fact_t.d3_k))",
+          "            └─TableFullScan_21 8.00 cop[tiflash] table:fact_t keep order:false"
+        ]
+      }
+    ]
+  },
   {
     "Name": "TestReadFromStorageHint",
     "Cases": [

store/tikv/coprocessor.go

@@ -55,6 +55,14 @@ type CopClient struct {
 	replicaReadSeed uint32
 }
 
+func (c *CopClient) GetBatchCopTaskNumber() (ret int32) {
+	ret = c.store.regionCache.storeMu.flashStoreNumber
+	if ret <= 0 {
+		ret = 1
+	}
+	return ret
+}
+
 // Send builds the request and gets the coprocessor iterator response.
 func (c *CopClient) Send(ctx context.Context, req *kv.Request, vars *kv.Variables) kv.Response {
 	if req.StoreType == kv.TiFlash && req.BatchCop {

store/tikv/region_cache.go

@@ -217,6 +217,7 @@ type RegionCache struct {
 	storeMu struct {
 		sync.RWMutex
 		stores map[uint64]*Store
+		flashStoreNumber int32
 	}
 	notifyDieCh   chan []string
 	notifyCheckCh chan struct{}
@@ -231,6 +232,7 @@ func NewRegionCache(pdClient pd.Client) *RegionCache {
 	c.mu.regions = make(map[RegionVerID]*Region)
 	c.mu.sorted = btree.New(btreeDegree)
 	c.storeMu.stores = make(map[uint64]*Store)
+	c.storeMu.flashStoreNumber = 0
 	c.notifyCheckCh = make(chan struct{}, 1)
 	c.notifyDieCh = make(chan []string, 1)
 	c.closeCh = make(chan struct{})
@@ -951,6 +953,13 @@ func (c *RegionCache) getStoreAddr(bo *Backoffer, region *Region, store *Store,
 		return
 	case unresolved:
 		addr, err = store.initResolve(bo, c)
+		if store.storeType == kv.TiFlash {
+			c.storeMu.Lock()
+			if _,exists := c.storeMu.stores[store.storeID]; exists {
+				c.storeMu.flashStoreNumber++
+			}
+			c.storeMu.Unlock()
+		}
 		return
 	case deleted:
 		addr = c.changeToActiveStore(region, store, storeIdx)
@@ -1371,7 +1380,14 @@ func (s *Store) reResolve(c *RegionCache) {
 		newStore := &Store{storeID: s.storeID, addr: addr, storeType: storeType}
 		newStore.state = *(*uint64)(unsafe.Pointer(&state))
 		c.storeMu.Lock()
+		orgStore,exists := c.storeMu.stores[newStore.storeID]
+		if exists && orgStore.storeType == kv.TiFlash {
+			c.storeMu.flashStoreNumber--
+		}
 		c.storeMu.stores[newStore.storeID] = newStore
+		if newStore.storeType == kv.TiFlash {
+			c.storeMu.flashStoreNumber++
+		}
 		c.storeMu.Unlock()
 	retryMarkDel:
 		// all region used those