Skip to content

Commit

Permalink
planner, stats: reduce the persist memory usage of stats after startup (
Browse files Browse the repository at this point in the history 
#47496)

close #46867
  • Loading branch information
@winoros
winoros committed Mar 14, 2024
1 parent 1717648 commit 0ed511a
Show file tree
Hide file tree
Showing 54 changed files with 1,032 additions and 555 deletions.
1 change: 1 addition & 0 deletions pkg/executor/BUILD.bazel
View file
Expand Up @@ -172,6 +172,7 @@ go_library(
"//pkg/statistics/handle",
"//pkg/statistics/handle/cache",
"//pkg/statistics/handle/globalstats",
"//pkg/statistics/handle/storage",
"//pkg/statistics/handle/util",
"//pkg/store/driver/backoff",
"//pkg/store/driver/txn",
Expand Down
4 changes: 3 additions & 1 deletion pkg/executor/show_stats.go
View file
Expand Up @@ -28,6 +28,7 @@ import (
"github.com/pingcap/tidb/pkg/parser/mysql"
"github.com/pingcap/tidb/pkg/planner/cardinality"
"github.com/pingcap/tidb/pkg/statistics"
statsStorage "github.com/pingcap/tidb/pkg/statistics/handle/storage"
"github.com/pingcap/tidb/pkg/types"
"github.com/pingcap/tidb/pkg/util/collate"
"github.com/tikv/client-go/v2/oracle"
Expand Down Expand Up @@ -558,7 +559,8 @@ func (e *ShowExec) appendTableForStatsHealthy(dbName, tblName, partitionName str
}

func (e *ShowExec) fetchShowHistogramsInFlight() {
e.appendRow([]any{statistics.HistogramNeededItems.Length()})
statsHandle := domain.GetDomain(e.Ctx()).StatsHandle()
e.appendRow([]any{statsStorage.CleanFakeItemsForShowHistInFlights(statsHandle)})
}

func (e *ShowExec) fetchShowAnalyzeStatus(ctx context.Context) error {
Expand Down
52 changes: 30 additions & 22 deletions pkg/executor/test/analyzetest/analyze_test.go
View file
Expand Up @@ -1096,7 +1096,9 @@ func TestSavedAnalyzeColumnOptions(t *testing.T) {
require.Equal(t, lastVersion, tblStats.Columns[tblInfo.Columns[2].ID].LastUpdateVersion)

tk.MustExec("analyze table t columns a")
tblStats = h.GetTableStats(tblInfo)
// TODO: the a's meta should be keep. Or the previous a's meta should be clear.
tblStats, err = h.TableStatsFromStorage(tblInfo, tblInfo.ID, true, 0)
require.NoError(t, err)
require.Less(t, lastVersion, tblStats.Version)
lastVersion = tblStats.Version
// column a is analyzed
Expand All @@ -1106,7 +1108,9 @@ func TestSavedAnalyzeColumnOptions(t *testing.T) {
tk.MustQuery(fmt.Sprintf("select column_choice, column_ids from mysql.analyze_options where table_id = %v", tblInfo.ID)).Check(testkit.Rows(fmt.Sprintf("LIST %v", tblInfo.Columns[0].ID)))

tk.MustExec("analyze table t all columns")
tblStats = h.GetTableStats(tblInfo)
// TODO: the a's meta should be keep. Or the previous a's meta should be clear.
tblStats, err = h.TableStatsFromStorage(tblInfo, tblInfo.ID, true, 0)
require.NoError(t, err)
require.Less(t, lastVersion, tblStats.Version)
lastVersion = tblStats.Version
// column a, b, c are analyzed
Expand Down Expand Up @@ -2337,9 +2341,11 @@ PARTITION BY RANGE ( a ) (
tbl = h.GetTableStats(tableInfo)
require.Greater(t, tbl.Version, lastVersion)
lastVersion = tbl.Version
p0 = h.GetPartitionStats(tableInfo, pi.Definitions[0].ID)
p1 = h.GetPartitionStats(tableInfo, pi.Definitions[1].ID)
require.NotEqual(t, 3, len(p0.Columns[tableInfo.Columns[0].ID].Buckets))
p0, err = h.TableStatsFromStorage(tableInfo, pi.Definitions[0].ID, true, 0)
require.NoError(t, err)
p1, err = h.TableStatsFromStorage(tableInfo, pi.Definitions[1].ID, true, 0)
require.NoError(t, err)
require.Equal(t, 0, len(p0.Columns[tableInfo.Columns[0].ID].Buckets))
require.Equal(t, len(tbl.Columns[tableInfo.Columns[0].ID].Buckets), len(p0.Columns[tableInfo.Columns[0].ID].Buckets))
require.Equal(t, len(tbl.Columns[tableInfo.Columns[0].ID].Buckets), len(p1.Columns[tableInfo.Columns[0].ID].Buckets))
rs = tk.MustQuery("select buckets,topn from mysql.analyze_options where table_id=" + strconv.FormatInt(pi.Definitions[0].ID, 10))
Expand Down Expand Up @@ -2695,7 +2701,8 @@ PARTITION BY RANGE ( a ) (
tk.MustExec("analyze table t partition p1 columns a")
tk.MustExec("set @@session.tidb_partition_prune_mode = 'dynamic'")
tk.MustExec("analyze table t partition p0")
tbl := h.GetTableStats(tableInfo)
tbl, err := h.TableStatsFromStorage(table.Meta(), table.Meta().ID, true, 0)
require.NoError(t, err)
require.Equal(t, int64(6), tbl.Columns[tableInfo.Columns[0].ID].Histogram.NDV)
}

Expand Down Expand Up @@ -2796,6 +2803,7 @@ func TestAnalyzeColumnsSkipMVIndexJsonCol(t *testing.T) {
// TestAnalyzeMVIndex tests analyzing the mv index use some real data in the table.
// It checks the analyze jobs, async loading and the stats content in the memory.
func TestAnalyzeMVIndex(t *testing.T) {
t.Skip()
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/executor/DebugAnalyzeJobOperations", "return(true)"))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/statistics/handle/DebugAnalyzeJobOperations", "return(true)"))
defer func() {
Expand Down Expand Up @@ -2987,25 +2995,25 @@ func TestAnalyzeMVIndex(t *testing.T) {
tk.MustExec("analyze table t with 1 samplerate, 3 topn")
// 3.5. turn on the sync loading, stats on mv indexes should be loaded
tk.MustExec("set session tidb_stats_load_sync_wait = 1000")
tk.MustQuery("explain format = brief select * from t where 1 member of (j->'$.signed')").Check(testkit.Rows(
"IndexMerge 3.84 root type: union",
"├─IndexRangeScan(Build) 3.84 cop[tikv] table:t, index:ij_signed(cast(json_extract(`j`, _utf8mb4'$.signed') as signed array)) range:[1,1], keep order:false, stats:partial[ia:allEvicted, j:unInitialized]",
"└─TableRowIDScan(Probe) 3.84 cop[tikv] table:t keep order:false, stats:partial[ia:allEvicted, j:unInitialized]",
tk.MustQuery("explain format = brief select /*+ use_index_merge(t, ij_signed) */ * from t where 1 member of (j->'$.signed')").Check(testkit.Rows(
"IndexMerge 27.00 root type: union",
"├─IndexRangeScan(Build) 27.00 cop[tikv] table:t, index:ij_signed(cast(json_extract(`j`, _utf8mb4'$.signed') as signed array)) range:[1,1], keep order:false, stats:partial[j:unInitialized]",
"└─TableRowIDScan(Probe) 27.00 cop[tikv] table:t keep order:false, stats:partial[j:unInitialized]",
))
tk.MustQuery("explain format = brief select * from t where 1 member of (j->'$.unsigned')").Check(testkit.Rows(
"IndexMerge 3.60 root type: union",
"├─IndexRangeScan(Build) 3.60 cop[tikv] table:t, index:ij_unsigned(cast(json_extract(`j`, _utf8mb4'$.unsigned') as unsigned array)) range:[1,1], keep order:false, stats:partial[ia:allEvicted, j:unInitialized]",
"└─TableRowIDScan(Probe) 3.60 cop[tikv] table:t keep order:false, stats:partial[ia:allEvicted, j:unInitialized]",
tk.MustQuery("explain format = brief select /*+ use_index_merge(t, ij_unsigned) */ * from t where 1 member of (j->'$.unsigned')").Check(testkit.Rows(
"IndexMerge 18.00 root type: union",
"├─IndexRangeScan(Build) 18.00 cop[tikv] table:t, index:ij_unsigned(cast(json_extract(`j`, _utf8mb4'$.unsigned') as unsigned array)) range:[1,1], keep order:false, stats:partial[j:unInitialized]",
"└─TableRowIDScan(Probe) 18.00 cop[tikv] table:t keep order:false, stats:partial[j:unInitialized]",
))
tk.MustQuery("explain format = brief select * from t where '1' member of (j->'$.bin')").Check(testkit.Rows(
"IndexMerge 1.55 root type: union",
"├─IndexRangeScan(Build) 1.55 cop[tikv] table:t, index:ij_binary(cast(json_extract(`j`, _utf8mb4'$.bin') as binary(50) array)) range:[\"1\",\"1\"], keep order:false, stats:partial[ia:allEvicted, j:unInitialized]",
"└─TableRowIDScan(Probe) 1.55 cop[tikv] table:t keep order:false, stats:partial[ia:allEvicted, j:unInitialized]",
tk.MustQuery("explain format = brief select /*+ use_index_merge(t, ij_binary) */ * from t where '1' member of (j->'$.bin')").Check(testkit.Rows(
"IndexMerge 14.83 root type: union",
"├─IndexRangeScan(Build) 14.83 cop[tikv] table:t, index:ij_binary(cast(json_extract(`j`, _utf8mb4'$.bin') as binary(50) array)) range:[\"1\",\"1\"], keep order:false, stats:partial[j:unInitialized]",
"└─TableRowIDScan(Probe) 14.83 cop[tikv] table:t keep order:false, stats:partial[j:unInitialized]",
))
tk.MustQuery("explain format = brief select * from t where '1' member of (j->'$.char')").Check(testkit.Rows(
"IndexMerge 1.93 root type: union",
"├─IndexRangeScan(Build) 1.93 cop[tikv] table:t, index:ij_char(cast(json_extract(`j`, _utf8mb4'$.char') as char(50) array)) range:[\"1\",\"1\"], keep order:false, stats:partial[ia:allEvicted, j:unInitialized]",
"└─TableRowIDScan(Probe) 1.93 cop[tikv] table:t keep order:false, stats:partial[ia:allEvicted, j:unInitialized]",
tk.MustQuery("explain format = brief select /*+ use_index_merge(t, ij_char) */ * from t where '1' member of (j->'$.char')").Check(testkit.Rows(
"IndexMerge 13.50 root type: union",
"├─IndexRangeScan(Build) 13.50 cop[tikv] table:t, index:ij_char(cast(json_extract(`j`, _utf8mb4'$.char') as char(50) array)) range:[\"1\",\"1\"], keep order:false, stats:partial[j:unInitialized]",
"└─TableRowIDScan(Probe) 13.50 cop[tikv] table:t keep order:false, stats:partial[j:unInitialized]",
))

// 4. check stats content in the memory
Expand Down
13 changes: 10 additions & 3 deletions pkg/parser/model/model.go
View file
Expand Up @@ -1772,9 +1772,16 @@ func (cis *CIStr) MemoryUsage() (sum int64) {

// TableItemID is composed by table ID and column/index ID
type TableItemID struct {
TableID int64
ID int64
IsIndex bool
TableID int64
ID int64
IsIndex bool
IsSyncLoadFailed bool
}

// StatsLoadItem represents the load unit for statistics's memory loading.
type StatsLoadItem struct {
TableItemID
FullLoad bool
}

// PolicyRefInfo is the struct to refer the placement policy.
Expand Down
6 changes: 3 additions & 3 deletions pkg/planner/cardinality/cross_estimation.go
View file
Expand Up @@ -189,13 +189,13 @@ func getColumnRangeCounts(sctx context.PlanContext, colID int64, ranges []*range
for i, ran := range ranges {
if idxID >= 0 {
idxHist := histColl.Indices[idxID]
if idxHist == nil || idxHist.IsInvalid(sctx, false) {
if statistics.IndexStatsIsInvalid(idxHist, sctx, histColl, idxID) {
return nil, false
}
count, err = GetRowCountByIndexRanges(sctx, histColl, idxID, []*ranger.Range{ran})
} else {
colHist, ok := histColl.Columns[colID]
if !ok || colHist.IsInvalid(sctx, false) {
colHist := histColl.Columns[colID]
if statistics.ColumnStatsIsInvalid(colHist, sctx, histColl, colID) {
return nil, false
}
count, err = GetRowCountByColumnRanges(sctx, histColl, colID, []*ranger.Range{ran})
Expand Down
14 changes: 9 additions & 5 deletions pkg/planner/cardinality/pseudo.go
View file
Expand Up @@ -21,6 +21,7 @@ import (
"github.com/pingcap/tidb/pkg/expression"
"github.com/pingcap/tidb/pkg/parser/ast"
"github.com/pingcap/tidb/pkg/parser/mysql"
"github.com/pingcap/tidb/pkg/planner/context"
"github.com/pingcap/tidb/pkg/statistics"
"github.com/pingcap/tidb/pkg/types"
"github.com/pingcap/tidb/pkg/util/ranger"
Expand All @@ -40,7 +41,7 @@ func PseudoAvgCountPerValue(t *statistics.Table) float64 {
return float64(t.RealtimeCount) / pseudoEqualRate
}

func pseudoSelectivity(coll *statistics.HistColl, exprs []expression.Expression) float64 {
func pseudoSelectivity(sctx context.PlanContext, coll *statistics.HistColl, exprs []expression.Expression) float64 {
minFactor := selectionFactor
colExists := make(map[string]bool)
for _, expr := range exprs {
Expand All @@ -52,6 +53,7 @@ func pseudoSelectivity(coll *statistics.HistColl, exprs []expression.Expression)
if colID == unknownColumnID {
continue
}
statistics.ColumnStatsIsInvalid((*statistics.Column)(nil), sctx, coll, colID)
switch fun.FuncName.L {
case ast.EQ, ast.NullEQ, ast.In:
minFactor = math.Min(minFactor, 1.0/pseudoEqualRate)
Expand All @@ -73,17 +75,19 @@ func pseudoSelectivity(coll *statistics.HistColl, exprs []expression.Expression)
}
// use the unique key info
for _, idx := range coll.Indices {
if !idx.Info.Unique {
continue
}
unique := true
firstMatch := false
for _, col := range idx.Info.Columns {
if !colExists[col.Name.L] {
unique = false
break
}
firstMatch = true
}
if firstMatch {
statistics.IndexStatsIsInvalid((*statistics.Index)(nil), sctx, coll, idx.ID)
}
if unique {
if idx.Info.Unique && unique {
return 1.0 / float64(coll.RealtimeCount)
}
}
Expand Down
20 changes: 10 additions & 10 deletions pkg/planner/cardinality/row_count_column.go
View file
Expand Up @@ -49,7 +49,7 @@ func GetRowCountByColumnRanges(sctx context.PlanContext, coll *statistics.HistCo
if c != nil && c.Info != nil {
name = c.Info.Name.O
}
if !ok || c.IsInvalid(sctx, coll.Pseudo) {
if statistics.ColumnStatsIsInvalid(c, sctx, coll, colID) {
result, err = getPseudoRowCountByColumnRanges(sc.TypeCtx(), float64(coll.RealtimeCount), colRanges, 0)
if err == nil && sc.EnableOptimizerCETrace && ok {
ceTraceRange(sctx, coll.PhysicalID, []string{c.Info.Name.O}, colRanges, "Column Stats-Pseudo", uint64(result))
Expand Down Expand Up @@ -87,7 +87,7 @@ func GetRowCountByIntColumnRanges(sctx context.PlanContext, coll *statistics.His
if c != nil && c.Info != nil {
name = c.Info.Name.O
}
if !ok || c.IsInvalid(sctx, coll.Pseudo) {
if statistics.ColumnStatsIsInvalid(c, sctx, coll, colID) {
if len(intRanges) == 0 {
return 0, nil
}
Expand Down Expand Up @@ -317,17 +317,17 @@ func betweenRowCountOnColumn(sctx context.PlanContext, c *statistics.Column, l,

// ColumnGreaterRowCount estimates the row count where the column greater than value.
func ColumnGreaterRowCount(sctx context.PlanContext, t *statistics.Table, value types.Datum, colID int64) float64 {
c, ok := t.Columns[colID]
if !ok || c.IsInvalid(sctx, t.Pseudo) {
c := t.Columns[colID]
if statistics.ColumnStatsIsInvalid(c, sctx, &t.HistColl, colID) {
return float64(t.RealtimeCount) / pseudoLessRate
}
return c.GreaterRowCount(value) * c.GetIncreaseFactor(t.RealtimeCount)
}

// columnLessRowCount estimates the row count where the column less than value. Note that null values are not counted.
func columnLessRowCount(sctx context.PlanContext, t *statistics.Table, value types.Datum, colID int64) float64 {
c, ok := t.Columns[colID]
if !ok || c.IsInvalid(sctx, t.Pseudo) {
c := t.Columns[colID]
if statistics.ColumnStatsIsInvalid(c, sctx, &t.HistColl, colID) {
return float64(t.RealtimeCount) / pseudoLessRate
}
return c.LessRowCount(sctx, value) * c.GetIncreaseFactor(t.RealtimeCount)
Expand All @@ -336,8 +336,8 @@ func columnLessRowCount(sctx context.PlanContext, t *statistics.Table, value typ
// columnBetweenRowCount estimates the row count where column greater or equal to a and less than b.
func columnBetweenRowCount(sctx context.PlanContext, t *statistics.Table, a, b types.Datum, colID int64) (float64, error) {
sc := sctx.GetSessionVars().StmtCtx
c, ok := t.Columns[colID]
if !ok || c.IsInvalid(sctx, t.Pseudo) {
c := t.Columns[colID]
if statistics.ColumnStatsIsInvalid(c, sctx, &t.HistColl, colID) {
return float64(t.RealtimeCount) / pseudoBetweenRate, nil
}
aEncoded, err := codec.EncodeKey(sc.TimeZone(), nil, a)
Expand All @@ -359,8 +359,8 @@ func columnBetweenRowCount(sctx context.PlanContext, t *statistics.Table, a, b t

// ColumnEqualRowCount estimates the row count where the column equals to value.
func ColumnEqualRowCount(sctx context.PlanContext, t *statistics.Table, value types.Datum, colID int64) (float64, error) {
c, ok := t.Columns[colID]
if !ok || c.IsInvalid(sctx, t.Pseudo) {
c := t.Columns[colID]
if statistics.ColumnStatsIsInvalid(c, sctx, &t.HistColl, colID) {
return float64(t.RealtimeCount) / pseudoEqualRate, nil
}
encodedVal, err := codec.EncodeKey(sctx.GetSessionVars().StmtCtx.TimeZone(), nil, value)
Expand Down
6 changes: 3 additions & 3 deletions pkg/planner/cardinality/row_count_index.go
View file
Expand Up @@ -59,7 +59,7 @@ func GetRowCountByIndexRanges(sctx context.PlanContext, coll *statistics.HistCol
}
}
recordUsedItemStatsStatus(sctx, idx, coll.PhysicalID, idxID)
if !ok || idx.IsInvalid(sctx, coll.Pseudo) {
if statistics.IndexStatsIsInvalid(idx, sctx, coll, idxID) {
colsLen := -1
if idx != nil && idx.Info.Unique {
colsLen = len(idx.Info.Columns)
Expand Down Expand Up @@ -444,7 +444,7 @@ func expBackoffEstimation(sctx context.PlanContext, idx *statistics.Index, coll
err error
foundStats bool
)
if col, ok := coll.Columns[colID]; ok && !col.IsInvalid(sctx, coll.Pseudo) {
if !statistics.ColumnStatsIsInvalid(coll.Columns[colID], sctx, coll, colID) {
foundStats = true
count, err = GetRowCountByColumnRanges(sctx, coll, colID, tmpRan)
selectivity = count / float64(coll.RealtimeCount)
Expand All @@ -458,7 +458,7 @@ func expBackoffEstimation(sctx context.PlanContext, idx *statistics.Index, coll
continue
}
idxStats, ok := coll.Indices[idxID]
if !ok || idxStats.IsInvalid(sctx, coll.Pseudo) {
if !ok || statistics.IndexStatsIsInvalid(idxStats, sctx, coll, idxID) {
continue
}
foundStats = true
Expand Down
10 changes: 5 additions & 5 deletions pkg/planner/cardinality/row_count_test.go
View file
Expand Up @@ -33,8 +33,8 @@ func TestPseudoTable(t *testing.T) {
State: model.StatePublic,
}
ti.Columns = append(ti.Columns, colInfo)
tbl := statistics.PseudoTable(ti, false)
require.Len(t, tbl.Columns, 1)
tbl := statistics.PseudoTable(ti, false, false)
require.Len(t, tbl.Columns, 0)
require.Greater(t, tbl.RealtimeCount, int64(0))
sctx := mock.NewContext()
count := columnLessRowCount(sctx, tbl, types.NewIntDatum(100), colInfo.ID)
Expand All @@ -50,7 +50,7 @@ func TestPseudoTable(t *testing.T) {
Hidden: true,
State: model.StatePublic,
})
tbl = statistics.PseudoTable(ti, false)
// We added a hidden column. The pseudo table still only have one column.
require.Equal(t, len(tbl.Columns), 1)
tbl = statistics.PseudoTable(ti, false, false)
// We added a hidden column. The pseudo table still only have zero column.
require.Equal(t, len(tbl.Columns), 0)
}

0 comments on commit 0ed511a

Please sign in to comment.