You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

1055 lines
34 KiB

// Copyright 2019 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 (
"bytes"
"fmt"
"math"
"regexp"
"sort"
"strconv"
"strings"
"time"
"github.com/cznic/mathutil"
"github.com/pingcap/parser/ast"
"github.com/pingcap/parser/mysql"
"github.com/pingcap/tidb/expression"
"github.com/pingcap/tidb/infoschema"
"github.com/pingcap/tidb/sessionctx"
"github.com/pingcap/tidb/types"
"github.com/pingcap/tidb/util/set"
"github.com/pingcap/tidb/util/stringutil"
"github.com/pingcap/tipb/go-tipb"
)
// MemTablePredicateExtractor is used to extract some predicates from `WHERE` clause
// and push the predicates down to the data retrieving on reading memory table stage.
//
// e.g:
// SELECT * FROM cluster_config WHERE type='tikv' AND instance='192.168.1.9:2379'
// We must request all components in the cluster via HTTP API for retrieving
// configurations and filter them by `type/instance` columns.
//
// The purpose of defining a `MemTablePredicateExtractor` is to optimize this
// 1. Define a `ClusterConfigTablePredicateExtractor`
// 2. Extract the `type/instance` columns on the logic optimizing stage and save them via fields.
// 3. Passing the extractor to the `ClusterReaderExecExec` executor
// 4. Executor sends requests to the target components instead of all of the components
type MemTablePredicateExtractor interface {
// Extracts predicates which can be pushed down and returns the remained predicates
Extract(sessionctx.Context, *expression.Schema, []*types.FieldName, []expression.Expression) (remained []expression.Expression)
explainInfo(p *PhysicalMemTable) string
}
// extractHelper contains some common utililty functions for all extractor.
// define an individual struct instead of a bunch of un-exported functions
// to avoid polluting the global scope of current package.
type extractHelper struct{}
func (helper extractHelper) extractColInConsExpr(extractCols map[int64]*types.FieldName, expr *expression.ScalarFunction) (string, []types.Datum) {
args := expr.GetArgs()
col, isCol := args[0].(*expression.Column)
if !isCol {
return "", nil
}
name, found := extractCols[col.UniqueID]
if !found {
return "", nil
}
// All expressions in IN must be a constant
// SELECT * FROM t1 WHERE c IN ('1', '2')
var results []types.Datum
for _, arg := range args[1:] {
constant, ok := arg.(*expression.Constant)
if !ok || constant.DeferredExpr != nil || constant.ParamMarker != nil {
return "", nil
}
results = append(results, constant.Value)
}
return name.ColName.L, results
}
func (helper extractHelper) extractColBinaryOpConsExpr(extractCols map[int64]*types.FieldName, expr *expression.ScalarFunction) (string, []types.Datum) {
args := expr.GetArgs()
var col *expression.Column
var colIdx int
// c = 'rhs'
// 'lhs' = c
for i := 0; i < 2; i++ {
var isCol bool
col, isCol = args[i].(*expression.Column)
if isCol {
colIdx = i
break
}
}
if col == nil {
return "", nil
}
name, found := extractCols[col.UniqueID]
if !found {
return "", nil
}
// The `lhs/rhs` of EQ expression must be a constant
// SELECT * FROM t1 WHERE c='rhs'
// SELECT * FROM t1 WHERE 'lhs'=c
constant, ok := args[1-colIdx].(*expression.Constant)
if !ok || constant.DeferredExpr != nil || constant.ParamMarker != nil {
return "", nil
}
return name.ColName.L, []types.Datum{constant.Value}
}
// extract the OR expression, e.g:
// SELECT * FROM t1 WHERE c1='a' OR c1='b' OR c1='c'
func (helper extractHelper) extractColOrExpr(extractCols map[int64]*types.FieldName, expr *expression.ScalarFunction) (string, []types.Datum) {
args := expr.GetArgs()
lhs, ok := args[0].(*expression.ScalarFunction)
if !ok {
return "", nil
}
rhs, ok := args[1].(*expression.ScalarFunction)
if !ok {
return "", nil
}
// Define an inner function to avoid populate the outer scope
var extract = func(extractCols map[int64]*types.FieldName, fn *expression.ScalarFunction) (string, []types.Datum) {
switch fn.FuncName.L {
case ast.EQ:
return helper.extractColBinaryOpConsExpr(extractCols, fn)
case ast.LogicOr:
return helper.extractColOrExpr(extractCols, fn)
case ast.In:
return helper.extractColInConsExpr(extractCols, fn)
default:
return "", nil
}
}
lhsColName, lhsDatums := extract(extractCols, lhs)
if lhsColName == "" {
return "", nil
}
rhsColName, rhsDatums := extract(extractCols, rhs)
if lhsColName == rhsColName {
return lhsColName, append(lhsDatums, rhsDatums...)
}
return "", nil
}
// merges `lhs` and `datums` with CNF logic
// 1. Returns `datums` set if the `lhs` is an empty set
// 2. Returns the intersection of `datums` and `lhs` if the `lhs` is not an empty set
func (helper extractHelper) merge(lhs set.StringSet, datums []types.Datum, toLower bool) set.StringSet {
tmpNodeTypes := set.NewStringSet()
for _, datum := range datums {
s, err := datum.ToString()
if err != nil {
return nil
}
if toLower {
s = strings.ToLower(s)
}
tmpNodeTypes.Insert(s)
}
if len(lhs) > 0 {
return lhs.Intersection(tmpNodeTypes)
}
return tmpNodeTypes
}
func (helper extractHelper) extractCol(
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
extractColName string,
valueToLower bool,
) (
remained []expression.Expression,
skipRequest bool,
result set.StringSet,
) {
remained = make([]expression.Expression, 0, len(predicates))
result = set.NewStringSet()
extractCols := helper.findColumn(schema, names, extractColName)
if len(extractCols) == 0 {
return predicates, false, result
}
// We should use INTERSECTION of sets because of the predicates is CNF array
for _, expr := range predicates {
fn, ok := expr.(*expression.ScalarFunction)
if !ok {
continue
}
var colName string
var datums []types.Datum // the memory of datums should not be reused, they will be put into result.
switch fn.FuncName.L {
case ast.EQ:
colName, datums = helper.extractColBinaryOpConsExpr(extractCols, fn)
case ast.In:
colName, datums = helper.extractColInConsExpr(extractCols, fn)
case ast.LogicOr:
colName, datums = helper.extractColOrExpr(extractCols, fn)
}
if colName == extractColName {
result = helper.merge(result, datums, valueToLower)
skipRequest = len(result) == 0
} else {
remained = append(remained, expr)
}
// There are no data if the low-level executor skip request, so the filter can be droped
if skipRequest {
remained = remained[:0]
break
}
}
return
}
// extracts the string pattern column, e.g:
// SELECT * FROM t WHERE c LIKE '%a%'
// SELECT * FROM t WHERE c LIKE '%a%' AND c REGEXP '.*xxx.*'
// SELECT * FROM t WHERE c LIKE '%a%' OR c REGEXP '.*xxx.*'
func (helper extractHelper) extractLikePatternCol(
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
extractColName string,
) (
remained []expression.Expression,
patterns []string,
) {
remained = make([]expression.Expression, 0, len(predicates))
extractCols := helper.findColumn(schema, names, extractColName)
if len(extractCols) == 0 {
return predicates, nil
}
// We use a string array to save multiple patterns because the Golang and Rust don't
// support perl-like CNF regular expression: (?=expr1)(?=expr2).
// e.g:
// SELECT * FROM t WHERE c LIKE '%a%' AND c LIKE '%b%' AND c REGEXP 'gc.*[0-9]{10,20}'
for _, expr := range predicates {
fn, ok := expr.(*expression.ScalarFunction)
if !ok {
remained = append(remained, expr)
continue
}
var canBuildPattern bool
var pattern string
// We use '|' to combine DNF regular expression: .*a.*|.*b.*
// e.g:
// SELECT * FROM t WHERE c LIKE '%a%' OR c LIKE '%b%'
if fn.FuncName.L == ast.LogicOr {
canBuildPattern, pattern = helper.extractOrLikePattern(fn, extractColName, extractCols)
} else {
canBuildPattern, pattern = helper.extractLikePattern(fn, extractColName, extractCols)
}
if canBuildPattern {
patterns = append(patterns, pattern)
} else {
remained = append(remained, expr)
}
}
return
}
func (helper extractHelper) extractOrLikePattern(
orFunc *expression.ScalarFunction,
extractColName string,
extractCols map[int64]*types.FieldName,
) (
ok bool,
pattern string,
) {
predicates := expression.SplitDNFItems(orFunc)
if len(predicates) == 0 {
return false, ""
}
patternBuilder := make([]string, 0, len(predicates))
for _, predicate := range predicates {
fn, ok := predicate.(*expression.ScalarFunction)
if !ok {
return false, ""
}
ok, partPattern := helper.extractLikePattern(fn, extractColName, extractCols)
if !ok {
return false, ""
}
patternBuilder = append(patternBuilder, partPattern)
}
return true, strings.Join(patternBuilder, "|")
}
func (helper extractHelper) extractLikePattern(
fn *expression.ScalarFunction,
extractColName string,
extractCols map[int64]*types.FieldName,
) (
ok bool,
pattern string,
) {
var colName string
var datums []types.Datum
switch fn.FuncName.L {
case ast.EQ, ast.Like, ast.Regexp:
colName, datums = helper.extractColBinaryOpConsExpr(extractCols, fn)
}
if colName == extractColName {
switch fn.FuncName.L {
case ast.EQ:
return true, "^" + regexp.QuoteMeta(datums[0].GetString()) + "$"
case ast.Like:
return true, stringutil.CompileLike2Regexp(datums[0].GetString())
case ast.Regexp:
return true, datums[0].GetString()
default:
return false, ""
}
} else {
return false, ""
}
}
func (helper extractHelper) findColumn(schema *expression.Schema, names []*types.FieldName, colName string) map[int64]*types.FieldName {
extractCols := make(map[int64]*types.FieldName)
for i, name := range names {
if name.ColName.L == colName {
extractCols[schema.Columns[i].UniqueID] = name
}
}
return extractCols
}
// getTimeFunctionName is used to get the (time) function name.
// For the expression that push down to the coprocessor, the function name is different with normal compare function,
// Then getTimeFunctionName will do a sample function name convert.
// Currently, this is used to support query `CLUSTER_SLOW_QUERY` at any time.
func (helper extractHelper) getTimeFunctionName(fn *expression.ScalarFunction) string {
switch fn.Function.PbCode() {
case tipb.ScalarFuncSig_GTTime:
return ast.GT
case tipb.ScalarFuncSig_GETime:
return ast.GE
case tipb.ScalarFuncSig_LTTime:
return ast.LT
case tipb.ScalarFuncSig_LETime:
return ast.LE
case tipb.ScalarFuncSig_EQTime:
return ast.EQ
default:
return fn.FuncName.L
}
}
// extracts the time range column, e.g:
// SELECT * FROM t WHERE time='2019-10-10 10:10:10'
// SELECT * FROM t WHERE time>'2019-10-10 10:10:10' AND time<'2019-10-11 10:10:10'
func (helper extractHelper) extractTimeRange(
ctx sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
extractColName string,
timezone *time.Location,
) (
remained []expression.Expression,
// unix timestamp in nanoseconds
startTime int64,
endTime int64,
) {
remained = make([]expression.Expression, 0, len(predicates))
extractCols := helper.findColumn(schema, names, extractColName)
if len(extractCols) == 0 {
return predicates, startTime, endTime
}
for _, expr := range predicates {
fn, ok := expr.(*expression.ScalarFunction)
if !ok {
remained = append(remained, expr)
continue
}
var colName string
var datums []types.Datum
fnName := helper.getTimeFunctionName(fn)
switch fnName {
case ast.GT, ast.GE, ast.LT, ast.LE, ast.EQ:
colName, datums = helper.extractColBinaryOpConsExpr(extractCols, fn)
}
if colName == extractColName {
timeType := types.NewFieldType(mysql.TypeDatetime)
timeType.Decimal = 6
timeDatum, err := datums[0].ConvertTo(ctx.GetSessionVars().StmtCtx, timeType)
if err != nil || timeDatum.Kind() == types.KindNull {
remained = append(remained, expr)
continue
}
mysqlTime := timeDatum.GetMysqlTime()
timestamp := time.Date(mysqlTime.Year(),
time.Month(mysqlTime.Month()),
mysqlTime.Day(),
mysqlTime.Hour(),
mysqlTime.Minute(),
mysqlTime.Second(),
mysqlTime.Microsecond()*1000,
timezone,
).UnixNano()
switch fnName {
case ast.EQ:
startTime = mathutil.MaxInt64(startTime, timestamp)
if endTime == 0 {
endTime = timestamp
} else {
endTime = mathutil.MinInt64(endTime, timestamp)
}
case ast.GT:
// FixMe: add 1ms is not absolutely correct here, just because the log search precision is millisecond.
startTime = mathutil.MaxInt64(startTime, timestamp+int64(time.Millisecond))
case ast.GE:
startTime = mathutil.MaxInt64(startTime, timestamp)
case ast.LT:
if endTime == 0 {
endTime = timestamp - int64(time.Millisecond)
} else {
endTime = mathutil.MinInt64(endTime, timestamp-int64(time.Millisecond))
}
case ast.LE:
if endTime == 0 {
endTime = timestamp
} else {
endTime = mathutil.MinInt64(endTime, timestamp)
}
default:
remained = append(remained, expr)
}
} else {
remained = append(remained, expr)
}
}
return
}
func (helper extractHelper) parseQuantiles(quantileSet set.StringSet) []float64 {
quantiles := make([]float64, 0, len(quantileSet))
for k := range quantileSet {
v, err := strconv.ParseFloat(k, 64)
if err != nil {
// ignore the parse error won't affect result.
continue
}
quantiles = append(quantiles, v)
}
sort.Float64s(quantiles)
return quantiles
}
func (helper extractHelper) extractCols(
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
excludeCols set.StringSet,
valueToLower bool) ([]expression.Expression, bool, map[string]set.StringSet) {
cols := map[string]set.StringSet{}
remained := predicates
skipRequest := false
// Extract the label columns.
for _, name := range names {
if excludeCols.Exist(name.ColName.L) {
continue
}
var values set.StringSet
remained, skipRequest, values = helper.extractCol(schema, names, remained, name.ColName.L, valueToLower)
if skipRequest {
return nil, true, nil
}
if len(values) == 0 {
continue
}
cols[name.ColName.L] = values
}
return remained, skipRequest, cols
}
func (helper extractHelper) convertToTime(t int64) time.Time {
if t == 0 || t == math.MaxInt64 {
return time.Now()
}
return time.Unix(0, t)
}
// ClusterTableExtractor is used to extract some predicates of cluster table.
type ClusterTableExtractor struct {
extractHelper
// SkipRequest means the where clause always false, we don't need to request any component
SkipRequest bool
// NodeTypes represents all components types we should send request to.
// e.g:
// 1. SELECT * FROM cluster_config WHERE type='tikv'
// 2. SELECT * FROM cluster_config WHERE type in ('tikv', 'tidb')
NodeTypes set.StringSet
// Instances represents all components instances we should send request to.
// e.g:
// 1. SELECT * FROM cluster_config WHERE instance='192.168.1.7:2379'
// 2. SELECT * FROM cluster_config WHERE type in ('192.168.1.7:2379', '192.168.1.9:2379')
Instances set.StringSet
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *ClusterTableExtractor) Extract(_ sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) []expression.Expression {
remained, typeSkipRequest, nodeTypes := e.extractCol(schema, names, predicates, "type", true)
remained, addrSkipRequest, instances := e.extractCol(schema, names, remained, "instance", false)
e.SkipRequest = typeSkipRequest || addrSkipRequest
e.NodeTypes = nodeTypes
e.Instances = instances
return remained
}
func (e *ClusterTableExtractor) explainInfo(p *PhysicalMemTable) string {
return ""
}
// ClusterLogTableExtractor is used to extract some predicates of `cluster_config`
type ClusterLogTableExtractor struct {
extractHelper
// SkipRequest means the where clause always false, we don't need to request any component
SkipRequest bool
// NodeTypes represents all components types we should send request to.
// e.g:
// 1. SELECT * FROM cluster_log WHERE type='tikv'
// 2. SELECT * FROM cluster_log WHERE type in ('tikv', 'tidb')
NodeTypes set.StringSet
// Instances represents all components instances we should send request to.
// e.g:
// 1. SELECT * FROM cluster_log WHERE instance='192.168.1.7:2379'
// 2. SELECT * FROM cluster_log WHERE instance in ('192.168.1.7:2379', '192.168.1.9:2379')
Instances set.StringSet
// StartTime represents the beginning time of log message
// e.g: SELECT * FROM cluster_log WHERE time>'2019-10-10 10:10:10.999'
StartTime int64
// EndTime represents the ending time of log message
// e.g: SELECT * FROM cluster_log WHERE time<'2019-10-11 10:10:10.999'
EndTime int64
// Pattern is used to filter the log message
// e.g:
// 1. SELECT * FROM cluster_log WHERE message like '%gc%'
// 2. SELECT * FROM cluster_log WHERE message regexp '.*'
Patterns []string
LogLevels set.StringSet
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *ClusterLogTableExtractor) Extract(
ctx sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) []expression.Expression {
// Extract the `type/instance` columns
remained, typeSkipRequest, nodeTypes := e.extractCol(schema, names, predicates, "type", true)
remained, addrSkipRequest, instances := e.extractCol(schema, names, remained, "instance", false)
remained, levlSkipRequest, logLevels := e.extractCol(schema, names, remained, "level", true)
e.SkipRequest = typeSkipRequest || addrSkipRequest || levlSkipRequest
e.NodeTypes = nodeTypes
e.Instances = instances
e.LogLevels = logLevels
if e.SkipRequest {
return nil
}
remained, startTime, endTime := e.extractTimeRange(ctx, schema, names, remained, "time", time.Local)
// The time unit for search log is millisecond.
startTime = startTime / int64(time.Millisecond)
endTime = endTime / int64(time.Millisecond)
e.StartTime = startTime
e.EndTime = endTime
if startTime != 0 && endTime != 0 {
e.SkipRequest = startTime > endTime
}
if e.SkipRequest {
return nil
}
remained, patterns := e.extractLikePatternCol(schema, names, remained, "message")
e.Patterns = patterns
return remained
}
func (e *ClusterLogTableExtractor) explainInfo(p *PhysicalMemTable) string {
if e.SkipRequest {
return "skip_request: true"
}
r := new(bytes.Buffer)
st, et := e.StartTime, e.EndTime
if st > 0 {
st := time.Unix(0, st*1e6)
r.WriteString(fmt.Sprintf("start_time:%v, ", st.In(p.ctx.GetSessionVars().StmtCtx.TimeZone).Format(MetricTableTimeFormat)))
}
if et > 0 {
et := time.Unix(0, et*1e6)
r.WriteString(fmt.Sprintf("end_time:%v, ", et.In(p.ctx.GetSessionVars().StmtCtx.TimeZone).Format(MetricTableTimeFormat)))
}
if len(e.NodeTypes) > 0 {
r.WriteString(fmt.Sprintf("node_types:[%s], ", extractStringFromStringSet(e.NodeTypes)))
}
if len(e.Instances) > 0 {
r.WriteString(fmt.Sprintf("instances:[%s], ", extractStringFromStringSet(e.Instances)))
}
if len(e.LogLevels) > 0 {
r.WriteString(fmt.Sprintf("log_levels:[%s], ", extractStringFromStringSet(e.LogLevels)))
}
// remove the last ", " in the message info
s := r.String()
if len(s) > 2 {
return s[:len(s)-2]
}
return s
}
// MetricTableExtractor is used to extract some predicates of metrics_schema tables.
type MetricTableExtractor struct {
extractHelper
// SkipRequest means the where clause always false, we don't need to request any component
SkipRequest bool
// StartTime represents the beginning time of metric data.
StartTime time.Time
// EndTime represents the ending time of metric data.
EndTime time.Time
// LabelConditions represents the label conditions of metric data.
LabelConditions map[string]set.StringSet
Quantiles []float64
}
func newMetricTableExtractor() *MetricTableExtractor {
e := &MetricTableExtractor{}
e.StartTime, e.EndTime = e.getTimeRange(0, 0)
return e
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *MetricTableExtractor) Extract(
ctx sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) []expression.Expression {
// Extract the `quantile` columns
remained, skipRequest, quantileSet := e.extractCol(schema, names, predicates, "quantile", true)
e.Quantiles = e.parseQuantiles(quantileSet)
e.SkipRequest = skipRequest
if e.SkipRequest {
return nil
}
// Extract the `time` columns
remained, startTime, endTime := e.extractTimeRange(ctx, schema, names, remained, "time", ctx.GetSessionVars().StmtCtx.TimeZone)
e.StartTime, e.EndTime = e.getTimeRange(startTime, endTime)
e.SkipRequest = e.StartTime.After(e.EndTime)
if e.SkipRequest {
return nil
}
excludeCols := set.NewStringSet("quantile", "time", "value")
_, skipRequest, extractCols := e.extractCols(schema, names, remained, excludeCols, false)
e.SkipRequest = skipRequest
if e.SkipRequest {
return nil
}
e.LabelConditions = extractCols
// For some metric, the metric reader can't use the predicate, so keep all label conditions remained.
return remained
}
func (e *MetricTableExtractor) getTimeRange(start, end int64) (time.Time, time.Time) {
const defaultMetricQueryDuration = 10 * time.Minute
var startTime, endTime time.Time
if start == 0 && end == 0 {
endTime = time.Now()
return endTime.Add(-defaultMetricQueryDuration), endTime
}
if start != 0 {
startTime = e.convertToTime(start)
}
if end != 0 {
endTime = e.convertToTime(end)
}
if start == 0 {
startTime = endTime.Add(-defaultMetricQueryDuration)
}
if end == 0 {
endTime = startTime.Add(defaultMetricQueryDuration)
}
return startTime, endTime
}
func (e *MetricTableExtractor) explainInfo(p *PhysicalMemTable) string {
if e.SkipRequest {
return "skip_request: true"
}
promQL := e.GetMetricTablePromQL(p.ctx, p.Table.Name.L)
startTime, endTime := e.StartTime, e.EndTime
step := time.Second * time.Duration(p.ctx.GetSessionVars().MetricSchemaStep)
return fmt.Sprintf("PromQL:%v, start_time:%v, end_time:%v, step:%v",
promQL,
startTime.In(p.ctx.GetSessionVars().StmtCtx.TimeZone).Format(MetricTableTimeFormat),
endTime.In(p.ctx.GetSessionVars().StmtCtx.TimeZone).Format(MetricTableTimeFormat),
step,
)
}
// GetMetricTablePromQL uses to get the promQL of metric table.
func (e *MetricTableExtractor) GetMetricTablePromQL(sctx sessionctx.Context, lowerTableName string) string {
quantiles := e.Quantiles
def, err := infoschema.GetMetricTableDef(lowerTableName)
if err != nil {
return ""
}
if len(quantiles) == 0 {
quantiles = []float64{def.Quantile}
}
var buf bytes.Buffer
for i, quantile := range quantiles {
promQL := def.GenPromQL(sctx, e.LabelConditions, quantile)
if i > 0 {
buf.WriteByte(',')
}
buf.WriteString(promQL)
}
return buf.String()
}
// MetricSummaryTableExtractor is used to extract some predicates of metrics_schema tables.
type MetricSummaryTableExtractor struct {
extractHelper
// SkipRequest means the where clause always false, we don't need to request any component
SkipRequest bool
MetricsNames set.StringSet
Quantiles []float64
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *MetricSummaryTableExtractor) Extract(
_ sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) (remained []expression.Expression) {
remained, quantileSkip, quantiles := e.extractCol(schema, names, predicates, "quantile", false)
remained, metricsNameSkip, metricsNames := e.extractCol(schema, names, predicates, "metrics_name", true)
e.SkipRequest = quantileSkip || metricsNameSkip
e.Quantiles = e.parseQuantiles(quantiles)
e.MetricsNames = metricsNames
return remained
}
func (e *MetricSummaryTableExtractor) explainInfo(p *PhysicalMemTable) string {
return ""
}
// InspectionResultTableExtractor is used to extract some predicates of `inspection_result`
type InspectionResultTableExtractor struct {
extractHelper
// SkipInspection means the where clause always false, we don't need to request any component
SkipInspection bool
// Rules represents rules applied to, and we should apply all inspection rules if there is no rules specified
// e.g: SELECT * FROM inspection_result WHERE rule in ('ddl', 'config')
Rules set.StringSet
// Items represents items applied to, and we should apply all inspection item if there is no rules specified
// e.g: SELECT * FROM inspection_result WHERE item in ('ddl.lease', 'raftstore.threadpool')
Items set.StringSet
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *InspectionResultTableExtractor) Extract(
_ sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) (remained []expression.Expression) {
// Extract the `rule/item` columns
remained, ruleSkip, rules := e.extractCol(schema, names, predicates, "rule", true)
remained, itemSkip, items := e.extractCol(schema, names, remained, "item", true)
e.SkipInspection = ruleSkip || itemSkip
e.Rules = rules
e.Items = items
return remained
}
func (e *InspectionResultTableExtractor) explainInfo(p *PhysicalMemTable) string {
return ""
}
// InspectionSummaryTableExtractor is used to extract some predicates of `inspection_summary`
type InspectionSummaryTableExtractor struct {
extractHelper
// SkipInspection means the where clause always false, we don't need to request any component
SkipInspection bool
// Rules represents rules applied to, and we should apply all inspection rules if there is no rules specified
// e.g: SELECT * FROM inspection_summary WHERE rule in ('ddl', 'config')
Rules set.StringSet
MetricNames set.StringSet
Quantiles []float64
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *InspectionSummaryTableExtractor) Extract(
_ sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) (remained []expression.Expression) {
// Extract the `rule` columns
remained, ruleSkip, rules := e.extractCol(schema, names, predicates, "rule", true)
// Extract the `metric_name` columns
remained, metricNameSkip, metricNames := e.extractCol(schema, names, predicates, "metrics_name", true)
// Extract the `quantile` columns
remained, quantileSkip, quantileSet := e.extractCol(schema, names, predicates, "quantile", false)
e.SkipInspection = ruleSkip || quantileSkip || metricNameSkip
e.Rules = rules
e.Quantiles = e.parseQuantiles(quantileSet)
e.MetricNames = metricNames
return remained
}
func (e *InspectionSummaryTableExtractor) explainInfo(p *PhysicalMemTable) string {
return ""
}
// InspectionRuleTableExtractor is used to extract some predicates of `inspection_rules`
type InspectionRuleTableExtractor struct {
extractHelper
SkipRequest bool
Types set.StringSet
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *InspectionRuleTableExtractor) Extract(
_ sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) (remained []expression.Expression) {
// Extract the `type` columns
remained, tpSkip, tps := e.extractCol(schema, names, predicates, "type", true)
e.SkipRequest = tpSkip
e.Types = tps
return remained
}
func (e *InspectionRuleTableExtractor) explainInfo(p *PhysicalMemTable) string {
return ""
}
// SlowQueryExtractor is used to extract some predicates of `slow_query`
type SlowQueryExtractor struct {
extractHelper
SkipRequest bool
StartTime time.Time
EndTime time.Time
// Enable is true means the executor should use the time range to locate the slow-log file that need to be parsed.
// Enable is false, means the executor should keep the behavior compatible with before, which is only parse the
// current slow-log file.
Enable bool
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *SlowQueryExtractor) Extract(
ctx sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) []expression.Expression {
remained, startTime, endTime := e.extractTimeRange(ctx, schema, names, predicates, "time", ctx.GetSessionVars().StmtCtx.TimeZone)
e.setTimeRange(startTime, endTime)
e.SkipRequest = e.Enable && e.StartTime.After(e.EndTime)
if e.SkipRequest {
return nil
}
return remained
}
func (e *SlowQueryExtractor) setTimeRange(start, end int64) {
const defaultSlowQueryDuration = 24 * time.Hour
var startTime, endTime time.Time
if start == 0 && end == 0 {
return
}
if start != 0 {
startTime = e.convertToTime(start)
}
if end != 0 {
endTime = e.convertToTime(end)
}
if start == 0 {
startTime = endTime.Add(-defaultSlowQueryDuration)
}
if end == 0 {
endTime = startTime.Add(defaultSlowQueryDuration)
}
e.StartTime, e.EndTime = startTime, endTime
e.Enable = true
}
// TableStorageStatsExtractor is used to extract some predicates of `disk_usage`.
type TableStorageStatsExtractor struct {
extractHelper
// SkipRequest means the where clause always false, we don't need to request any component.
SkipRequest bool
// TableSchema represents tableSchema applied to, and we should apply all table disk usage if there is no schema specified.
// e.g: SELECT * FROM information_schema.disk_usage WHERE table_schema in ('test', 'information_schema').
TableSchema set.StringSet
// TableName represents tableName applied to, and we should apply all table disk usage if there is no table specified.
// e.g: SELECT * FROM information_schema.disk_usage WHERE table in ('schemata', 'tables').
TableName set.StringSet
}
// Extract implements the MemTablePredicateExtractor Extract interface.
func (e *TableStorageStatsExtractor) Extract(
_ sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) []expression.Expression {
// Extract the `table_schema` columns.
remained, schemaSkip, tableSchema := e.extractCol(schema, names, predicates, "table_schema", true)
// Extract the `table_name` columns.
remained, tableSkip, tableName := e.extractCol(schema, names, remained, "table_name", true)
e.SkipRequest = schemaSkip || tableSkip
if e.SkipRequest {
return nil
}
e.TableSchema = tableSchema
e.TableName = tableName
return remained
}
func (e *TableStorageStatsExtractor) explainInfo(p *PhysicalMemTable) string {
if e.SkipRequest {
return "skip_request: true"
}
r := new(bytes.Buffer)
if len(e.TableSchema) > 0 {
r.WriteString(fmt.Sprintf("schema:[%s]", extractStringFromStringSet(e.TableSchema)))
}
if r.Len() > 0 && len(e.TableName) > 0 {
r.WriteString(", ")
}
if len(e.TableName) > 0 {
r.WriteString(fmt.Sprintf("table:[%s]", extractStringFromStringSet(e.TableName)))
}
return r.String()
}
func (e *SlowQueryExtractor) explainInfo(p *PhysicalMemTable) string {
if e.SkipRequest {
return "skip_request: true"
}
if !e.Enable {
return fmt.Sprintf("only search in the current '%v' file", p.ctx.GetSessionVars().SlowQueryFile)
}
startTime := e.StartTime.In(p.ctx.GetSessionVars().StmtCtx.TimeZone)
endTime := e.EndTime.In(p.ctx.GetSessionVars().StmtCtx.TimeZone)
return fmt.Sprintf("start_time:%v, end_time:%v",
types.NewTime(types.FromGoTime(startTime), mysql.TypeDatetime, types.MaxFsp).String(),
types.NewTime(types.FromGoTime(endTime), mysql.TypeDatetime, types.MaxFsp).String())
}
// TiFlashSystemTableExtractor is used to extract some predicates of tiflash system table.
type TiFlashSystemTableExtractor struct {
extractHelper
// SkipRequest means the where clause always false, we don't need to request any component
SkipRequest bool
// TiFlashInstances represents all tiflash instances we should send request to.
// e.g:
// 1. SELECT * FROM information_schema.<table_name> WHERE tiflash_instance='192.168.1.7:3930'
// 2. SELECT * FROM information_schema.<table_name> WHERE tiflash_instance in ('192.168.1.7:3930', '192.168.1.9:3930')
TiFlashInstances set.StringSet
// TidbDatabases represents tidbDatabases applied to, and we should apply all tidb database if there is no database specified.
// e.g: SELECT * FROM information_schema.<table_name> WHERE tidb_database in ('test', 'test2').
TiDBDatabases string
// TidbTables represents tidbTables applied to, and we should apply all tidb table if there is no table specified.
// e.g: SELECT * FROM information_schema.<table_name> WHERE tidb_table in ('t', 't2').
TiDBTables string
}
// Extract implements the MemTablePredicateExtractor Extract interface
func (e *TiFlashSystemTableExtractor) Extract(_ sessionctx.Context,
schema *expression.Schema,
names []*types.FieldName,
predicates []expression.Expression,
) []expression.Expression {
// Extract the `tiflash_instance` columns.
remained, instanceSkip, tiflashInstances := e.extractCol(schema, names, predicates, "tiflash_instance", false)
// Extract the `tidb_database` columns.
remained, databaseSkip, tidbDatabases := e.extractCol(schema, names, remained, "tidb_database", true)
// Extract the `tidb_table` columns.
remained, tableSkip, tidbTables := e.extractCol(schema, names, remained, "tidb_table", true)
e.SkipRequest = instanceSkip || databaseSkip || tableSkip
if e.SkipRequest {
return nil
}
e.TiFlashInstances = tiflashInstances
e.TiDBDatabases = extractStringFromStringSet(tidbDatabases)
e.TiDBTables = extractStringFromStringSet(tidbTables)
return remained
}
func (e *TiFlashSystemTableExtractor) explainInfo(p *PhysicalMemTable) string {
if e.SkipRequest {
return "skip_request:true"
}
r := new(bytes.Buffer)
if len(e.TiFlashInstances) > 0 {
r.WriteString(fmt.Sprintf("tiflash_instances:[%s], ", extractStringFromStringSet(e.TiFlashInstances)))
}
if len(e.TiDBDatabases) > 0 {
r.WriteString(fmt.Sprintf("tidb_databases:[%s], ", e.TiDBDatabases))
}
if len(e.TiDBTables) > 0 {
r.WriteString(fmt.Sprintf("tidb_tables:[%s], ", e.TiDBTables))
}
// remove the last ", " in the message info
s := r.String()
if len(s) > 2 {
return s[:len(s)-2]
}
return s
}