planner, statistics: maintain histogram for inner join

planner/core/cbo_test.go

@@ -747,6 +747,32 @@ func (s *testAnalyzeSuite) TestInconsistentEstimation(c *C) {
 		))
 }
 
+func (s *testAnalyzeSuite) TestJoinWithHistogram(c *C) {
+	defer testleak.AfterTest(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("create table t(a int primary key, b int, index idx(b))")
+	tk.MustExec("create table tt(a int primary key, b int, index idx(b))")
+	tk.MustExec("insert into t values(1, 1), (2, 1), (3, 1), (4, 2), (5, 2), (6, 2), (7, 3), (8, 4), (9, 5)")
+	tk.MustExec("insert into tt values(1, 1), (3, 1), (5, 1), (7, 2), (9, 3), (15, 4)")
+	tk.MustExec("analyze table t, tt")
+	tk.MustExec("set @@session.tidb_optimizer_selectivity_level=1")
+	tk.MustQuery("explain select * from t t1 join tt t2 where t1.a=t2.a").Check(testkit.Rows(
+		"Projection_7 5.00 root t1.a, t1.b, t2.a, t2.b",
+		"└─IndexJoin_11 5.00 root inner join, inner:TableReader_10, outer key:t2.a, inner key:t1.a",
+		"  ├─TableReader_24 6.00 root data:TableScan_23",
+		"  │ └─TableScan_23 6.00 cop table:t2, range:[-inf,+inf], keep order:false",
+		"  └─TableReader_10 1.00 root data:TableScan_9",
+		"    └─TableScan_9 1.00 cop table:t1, range: decided by [t2.a], keep order:false",
+	))
+}
+
 func newStoreWithBootstrap() (kv.Storage, *domain.Domain, error) {
 	store, err := mockstore.NewMockTikvStore()
 	if err != nil {

planner/core/stats.go

@@ -75,6 +75,15 @@ func (p *baseLogicalPlan) DeriveStats(childStats []*property.StatsInfo) (*proper
 	return profile, nil
 }
 
+func allStatsHasHistColl(stats []*property.StatsInfo) bool {
+	for _, s := range stats {
+		if s.HistColl == nil {
+			return false
+		}
+	}
+	return true
+}
+
 func (ds *DataSource) getStatsByFilter(conds expression.CNFExprs) (*property.StatsInfo, *statistics.HistColl) {
 	profile := &property.StatsInfo{
 		RowCount:       float64(ds.statisticTable.Count),
@@ -281,6 +290,9 @@ func (p *LogicalJoin) DeriveStats(childStats []*property.StatsInfo) (*property.S
 		leftKeys = append(leftKeys, eqCond.GetArgs()[0].(*expression.Column))
 		rightKeys = append(rightKeys, eqCond.GetArgs()[1].(*expression.Column))
 	}
+	if p.JoinType == InnerJoin && p.ctx.GetSessionVars().OptimizerSelectivityLevel >= 1 && allStatsHasHistColl(childStats) {
+		return p.deriveInnerJoinStatsWithHist(leftKeys, rightKeys, childStats)
+	}
 	leftKeyCardinality := getCardinality(leftKeys, p.children[0].Schema(), leftProfile)
 	rightKeyCardinality := getCardinality(rightKeys, p.children[1].Schema(), rightProfile)
 	count := leftProfile.RowCount * rightProfile.RowCount / math.Max(leftKeyCardinality, rightKeyCardinality)
@@ -302,6 +314,97 @@ func (p *LogicalJoin) DeriveStats(childStats []*property.StatsInfo) (*property.S
 	return p.stats, nil
 }
 
+func (p *LogicalJoin) deriveInnerJoinStatsWithHist(leftKeys, rightKeys []*expression.Column, childStats []*property.StatsInfo) (*property.StatsInfo, error) {
+	leftChild, rightChild := p.children[0], p.children[1]
+	leftProfile, rightProfile := childStats[0], childStats[1]
+
+	cardinality := make([]float64, 0, p.schema.Len())
+	cardinality = append(cardinality, leftProfile.Cardinality...)
+	cardinality = append(cardinality, rightProfile.Cardinality...)
+
+	finalNdv, leftNdv, rightNdv := math.MaxFloat64, float64(1), float64(1)
+	newColID2Hist := make(map[int64]*statistics.Column)
+
+	// TODO: Support using index histogram to calculate the NDV after join and the final row count.
+	for i := range leftKeys {
+		leftHist, ok1 := leftProfile.HistColl.Columns[leftKeys[i].UniqueID]
+		rightHist, ok2 := rightProfile.HistColl.Columns[rightKeys[i].UniqueID]
+		if ok1 {
+			ok1 = !leftHist.IsInvalid(p.ctx.GetSessionVars().StmtCtx, leftProfile.HistColl.Pseudo)
+		}
+		if ok2 {
+			ok2 = !rightHist.IsInvalid(p.ctx.GetSessionVars().StmtCtx, rightProfile.HistColl.Pseudo)
+		}
+		lPos := leftChild.Schema().ColumnIndex(leftKeys[i])
+		rPos := rightChild.Schema().ColumnIndex(rightKeys[i])
+		var keyNdv float64
+		if ok1 && ok2 {
+			newHist := statistics.MergeHistogramForInnerJoin(&leftHist.Histogram, &rightHist.Histogram, leftKeys[i].RetType)
+			leftCol := &statistics.Column{Info: leftHist.Info, Histogram: *newHist}
+			rightCol := &statistics.Column{Info: rightHist.Info, Histogram: *newHist}
+			lIncreaseFactor := leftHist.GetIncreaseFactor(leftProfile.HistColl.Count)
+			// The factor is used to scale the NDV. When it's higher than one. NDV doesn't need to be changed.
+			if lIncreaseFactor > 1 {
+				lIncreaseFactor = 1
+			}
+			rIncreaseFactor := rightHist.GetIncreaseFactor(rightProfile.HistColl.Count)
+			if rIncreaseFactor > 1 {
+				rIncreaseFactor = 1
+			}
+			keyNdv = float64(newHist.NDV) * lIncreaseFactor * rIncreaseFactor
+			lPosNew := p.schema.ColumnIndex(leftKeys[i])
+			rPosNew := p.schema.ColumnIndex(rightKeys[i])
+			cardinality[lPosNew] = float64(newHist.NDV)
+			cardinality[rPosNew] = float64(newHist.NDV)
+			newColID2Hist[leftKeys[i].UniqueID] = leftCol
+			newColID2Hist[rightKeys[i].UniqueID] = rightCol
+		} else {
+			keyNdv = math.Min(leftChild.statsInfo().Cardinality[lPos], rightChild.statsInfo().Cardinality[rPos])
+		}
+		// Update the finalNdv using the most selective join key.
+		// TODO: support using index to enhance its accuracy
+		if finalNdv > keyNdv {
+			finalNdv = keyNdv
+			leftNdv = leftProfile.Cardinality[lPos]
+			rightNdv = rightProfile.Cardinality[rPos]
+		}
+	}
+	var count float64 = 0
+	if leftNdv != 0 || rightNdv != 0 {
+		count = leftProfile.RowCount / leftNdv * rightProfile.RowCount / rightNdv * finalNdv
+	}
+
+	// Update left column map in `HistColl`.
+	for uniqID, colHist := range leftProfile.HistColl.Columns {
+		_, ok := newColID2Hist[uniqID]
+		if ok {
+			continue
+		}
+		newColID2Hist[uniqID] = colHist
+	}
+
+	// Update right column map in `HistColl`.
+	for uniqID, colHist := range rightProfile.HistColl.Columns {
+		_, ok := newColID2Hist[uniqID]
+		if ok {
+			continue
+		}
+		newColID2Hist[uniqID] = colHist
+	}
+
+	// TODO: support calculate index histogram.
+	newHistColl := &statistics.HistColl{
+		Count:   int64(count),
+		Columns: newColID2Hist,
+	}
+	p.stats = &property.StatsInfo{
+		RowCount:    count,
+		Cardinality: cardinality,
+		HistColl:    newHistColl,
+	}
+	return p.stats, nil
+}
+
 // DeriveStats implement LogicalPlan DeriveStats interface.
 func (la *LogicalApply) DeriveStats(childStats []*property.StatsInfo) (*property.StatsInfo, error) {
 	leftProfile := childStats[0]

planner/core/stats_test.go

@@ -0,0 +1,150 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package core
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"path/filepath"
+
+	. "github.com/pingcap/check"
+	"github.com/pingcap/errors"
+	"github.com/pingcap/parser"
+	"github.com/pingcap/parser/ast"
+	"github.com/pingcap/parser/model"
+	"github.com/pingcap/tidb/ddl"
+	"github.com/pingcap/tidb/expression"
+	"github.com/pingcap/tidb/planner/property"
+	"github.com/pingcap/tidb/sessionctx"
+	"github.com/pingcap/tidb/statistics"
+	"github.com/pingcap/tidb/util/mock"
+)
+
+var _ = Suite(&testStatsSuite{})
+
+type testStatsSuite struct {
+	ctx sessionctx.Context
+}
+
+func (s *testStatsSuite) SetUpTest(c *C) {
+	s.ctx = mock.NewContext()
+}
+
+func (s *testStatsSuite) loadStatsByFileName(t string, tbl *model.TableInfo) (*statistics.HistColl, error) {
+	statsTbl := &statistics.JSONTable{}
+	path := filepath.Join("testdata", t)
+	bytes, err := ioutil.ReadFile(path)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	err = json.Unmarshal(bytes, statsTbl)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	coll, err := statistics.HistCollFromJSON(tbl, tbl.ID, statsTbl)
+
+	return coll, err
+}
+
+func (s *testStatsSuite) mockTableInfo(sql string, tblID int64) (*model.TableInfo, error) {
+	stmt, err := parser.New().ParseOneStmt(sql, "", "")
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	mockTbl, err := ddl.MockTableInfo(mock.NewContext(), stmt.(*ast.CreateTableStmt), tblID)
+	return mockTbl, err
+}
+
+func (s *testStatsSuite) TestInnerJoinStats(c *C) {
+	// t1's values are (i, i) where i = 1..9. Each pair appears once.
+	t1Tbl, err := s.mockTableInfo("create table t1(a int, b int, index a(a, b))", 1)
+	c.Assert(err, IsNil)
+	t1Coll, err := s.loadStatsByFileName("t1.json", t1Tbl)
+	c.Assert(err, IsNil)
+	t1ExprCols := expression.ColumnInfos2ColumnsWithDBName(s.ctx, model.NewCIStr("test"), t1Tbl.Name, t1Tbl.Columns)
+	t1FinalColl := t1Coll.GenerateHistCollFromColumnInfo(t1Tbl.Columns, t1ExprCols)
+	t1StatsInfo := &property.StatsInfo{
+		RowCount:    float64(t1Coll.Count),
+		Cardinality: make([]float64, len(t1ExprCols)),
+		HistColl:    t1FinalColl,
+	}
+	for i := range t1ExprCols {
+		t1StatsInfo.Cardinality[i] = float64(t1FinalColl.Columns[t1ExprCols[i].UniqueID].NDV)
+	}
+	t1Child := DataSource{}.Init(s.ctx)
+	t1Child.schema = expression.NewSchema(t1ExprCols...)
+	t1Child.stats = t1StatsInfo
+
+	t2Tbl, err := s.mockTableInfo("create table t2(a int, b int, index a(a, b))", 2)
+	c.Assert(err, IsNil)
+	// t2's values are (i, i) where i = 8..15. Each pair appears once.
+	t2Coll, err := s.loadStatsByFileName("t2.json", t2Tbl)
+	c.Assert(err, IsNil)
+	t2ExprCols := expression.ColumnInfos2ColumnsWithDBName(s.ctx, model.NewCIStr("test"), t2Tbl.Name, t2Tbl.Columns)
+	t2FinalColl := t2Coll.GenerateHistCollFromColumnInfo(t2Tbl.Columns, t2ExprCols)
+	t2StatsInfo := &property.StatsInfo{
+		RowCount:    float64(t2Coll.Count),
+		Cardinality: make([]float64, len(t2ExprCols)),
+		HistColl:    t2FinalColl,
+	}
+	for i := range t2ExprCols {
+		t2StatsInfo.Cardinality[i] = float64(t2FinalColl.Columns[t2ExprCols[i].UniqueID].NDV)
+	}
+	t2Child := DataSource{}.Init(s.ctx)
+	t2Child.schema = expression.NewSchema(t2ExprCols...)
+	t2Child.stats = t2StatsInfo
+
+	join := LogicalJoin{}.Init(s.ctx)
+	join.SetChildren(t1Child, t2Child)
+	join.schema = expression.MergeSchema(t1Child.schema, t2Child.schema)
+	finalStats, err := join.deriveInnerJoinStatsWithHist([]*expression.Column{t1Child.schema.Columns[0]}, []*expression.Column{t2Child.schema.Columns[0]}, []*property.StatsInfo{t1StatsInfo, t2StatsInfo})
+	c.Assert(err, IsNil)
+	c.Assert(finalStats.RowCount, Equals, float64(2))
+	c.Assert(len(finalStats.HistColl.Columns), Equals, 4)
+	ans1 := `column:0 ndv:2 totColSize:0
+num: 1 lower_bound: 8 upper_bound: 8 repeats: 1
+num: 1 lower_bound: 9 upper_bound: 9 repeats: 1`
+	c.Assert(finalStats.HistColl.Columns[1].String(), Equals, ans1)
+	ans2 := `column:2 ndv:9 totColSize:9
+num: 1 lower_bound: 1 upper_bound: 1 repeats: 1
+num: 1 lower_bound: 2 upper_bound: 2 repeats: 1
+num: 1 lower_bound: 3 upper_bound: 3 repeats: 1
+num: 1 lower_bound: 4 upper_bound: 4 repeats: 1
+num: 1 lower_bound: 5 upper_bound: 5 repeats: 1
+num: 1 lower_bound: 6 upper_bound: 6 repeats: 1
+num: 1 lower_bound: 7 upper_bound: 7 repeats: 1
+num: 1 lower_bound: 8 upper_bound: 8 repeats: 1
+num: 1 lower_bound: 9 upper_bound: 9 repeats: 1`
+	c.Assert(finalStats.HistColl.Columns[2].String(), Equals, ans2)
+	c.Assert(finalStats.HistColl.Columns[3].String(), Equals, ans1)
+	ans4 := `column:2 ndv:8 totColSize:8
+num: 1 lower_bound: 8 upper_bound: 8 repeats: 1
+num: 1 lower_bound: 9 upper_bound: 9 repeats: 1
+num: 1 lower_bound: 10 upper_bound: 10 repeats: 1
+num: 1 lower_bound: 11 upper_bound: 11 repeats: 1
+num: 1 lower_bound: 12 upper_bound: 12 repeats: 1
+num: 1 lower_bound: 13 upper_bound: 13 repeats: 1
+num: 1 lower_bound: 14 upper_bound: 14 repeats: 1
+num: 1 lower_bound: 15 upper_bound: 15 repeats: 1`
+	c.Assert(finalStats.HistColl.Columns[4].String(), Equals, ans4)
+	t2StatsInfo.RowCount /= 2
+	t2StatsInfo.HistColl.Count /= 2
+	for i := range t2StatsInfo.Cardinality {
+		t2StatsInfo.Cardinality[i] /= 2
+	}
+	finalStats, err = join.deriveInnerJoinStatsWithHist([]*expression.Column{t1Child.schema.Columns[0]}, []*expression.Column{t2Child.schema.Columns[0]}, []*property.StatsInfo{t1StatsInfo, t2StatsInfo})
+	c.Assert(err, IsNil)
+	c.Assert(finalStats.RowCount, Equals, float64(1))
+}

statistics/dump.go

@@ -171,17 +171,28 @@ func (h *Handle) loadStatsFromJSON(tableInfo *model.TableInfo, physicalID int64,
 
 // TableStatsFromJSON loads statistic from JSONTable and return the Table of statistic.
 func TableStatsFromJSON(tableInfo *model.TableInfo, physicalID int64, jsonTbl *JSONTable) (*Table, error) {
-	newHistColl := HistColl{
+	coll, err := HistCollFromJSON(tableInfo, physicalID, jsonTbl)
+	if err != nil {
+		return nil, err
+	}
+
+	tbl := &Table{
+		HistColl: *coll,
+	}
+
+	return tbl, nil
+}
+
+// HistCollFromJSON build HistColl from JSONTable.
+func HistCollFromJSON(tableInfo *model.TableInfo, physicalID int64, jsonTbl *JSONTable) (*HistColl, error) {
+	newHistColl := &HistColl{
 		PhysicalID:     physicalID,
 		HavePhysicalID: true,
 		Count:          jsonTbl.Count,
 		ModifyCount:    jsonTbl.ModifyCount,
 		Columns:        make(map[int64]*Column, len(jsonTbl.Columns)),
 		Indices:        make(map[int64]*Index, len(jsonTbl.Indices)),
 	}
-	tbl := &Table{
-		HistColl: newHistColl,
-	}
 	for id, jsonIdx := range jsonTbl.Indices {
 		for _, idxInfo := range tableInfo.Indices {
 			if idxInfo.Name.L != id {
@@ -194,7 +205,7 @@ func TableStatsFromJSON(tableInfo *model.TableInfo, physicalID int64, jsonTbl *J
 				CMSketch:  CMSketchFromProto(jsonIdx.CMSketch),
 				Info:      idxInfo,
 			}
-			tbl.Indices[idx.ID] = idx
+			newHistColl.Indices[idx.ID] = idx
 		}
 	}
 
@@ -208,7 +219,7 @@ func TableStatsFromJSON(tableInfo *model.TableInfo, physicalID int64, jsonTbl *J
 			sc := &stmtctx.StatementContext{TimeZone: time.UTC}
 			hist, err := hist.ConvertTo(sc, &colInfo.FieldType)
 			if err != nil {
-				return nil, errors.Trace(err)
+				return nil, err
 			}
 			hist.ID, hist.NullCount, hist.LastUpdateVersion, hist.TotColSize = colInfo.ID, jsonCol.NullCount, jsonCol.LastUpdateVersion, jsonCol.TotColSize
 			col := &Column{
@@ -219,8 +230,8 @@ func TableStatsFromJSON(tableInfo *model.TableInfo, physicalID int64, jsonTbl *J
 				Count:      count,
 				isHandle:   tableInfo.PKIsHandle && mysql.HasPriKeyFlag(colInfo.Flag),
 			}
-			tbl.Columns[col.ID] = col
+			newHistColl.Columns[col.ID] = col
 		}
 	}
-	return tbl, nil
+	return newHistColl, nil
 }