package unstable_plan import ( "fmt" "strings" "github.com/pingcap/errors" schemaUtils "github.com/pingcap/tidb-dashboard/pkg/utils" "github.com/pingcap/tidb-dashboard/util/reflectutil" "github.com/samber/lo" "gorm.io/gorm" "gorm.io/gorm/schema" "github.com/pingcap-inc/compute-engine/pkg/utils" ) const ( // [statement summary table](https://docs.pingcap.com/tidb/stable/statement-summary-tables) model. StatementsTable = "INFORMATION_SCHEMA.CLUSTER_STATEMENTS_SUMMARY_HISTORY" ImpactLow = "LOW" ImpactMedium = "MIDDLE" ImpactHigh = "HIGH" ) var ( schemaCache = schemaUtils.NewSysSchema() gormDefaultNamingStrategy = schema.NamingStrategy{} ) // Model is the model of statement summary table. // Note: all time unit is nano second (1s = 1000000000ns). type Model struct { AggBeginTime int64 `json:"summary_begin_time" agg:"FLOOR(UNIX_TIMESTAMP(MIN(summary_begin_time)))"` AggEndTime int64 `json:"summary_end_time" agg:"FLOOR(UNIX_TIMESTAMP(MAX(summary_end_time)))"` DigestText string `json:"digest_text" agg:"ANY_VALUE(digest_text)"` Digest string `json:"digest" agg:"ANY_VALUE(digest)"` StmtType string `json:"stmt_type" agg:"ANY_VALUE(stmt_type)"` SampleUser string `json:"sample_user" agg:"ANY_VALUE(sample_user)"` QuerySampleText string `json:"query_sample_text" agg:"ANY_VALUE(query_sample_text)"` SchemaName string `json:"schema_name" agg:"ANY_VALUE(schema_name)"` TableNames string `json:"table_names" agg:"ANY_VALUE(table_names)"` IndexNames string `json:"index_names" agg:"ANY_VALUE(index_names)"` PlanDigest string `json:"plan_digest" agg:"ANY_VALUE(plan_digest)"` AggExecCount int64 `json:"exec_count" agg:"SUM(exec_count)"` AggSumErrors int64 `json:"sum_errors" agg:"SUM(sum_errors)"` AggSumWarnings int64 `json:"sum_warnings" agg:"SUM(sum_warnings)"` AggSumLatency int64 `json:"sum_latency" agg:"SUM(sum_latency)"` AggMaxLatency int64 `json:"max_latency" agg:"MAX(max_latency)"` AggMinLatency int64 `json:"min_latency" agg:"MIN(min_latency)"` AggAvgLatency int64 `json:"avg_latency" agg:"CAST(SUM(exec_count * avg_latency) / SUM(exec_count) AS SIGNED)"` AggAvgParseLatency int64 `json:"avg_parse_latency" agg:"CAST(SUM(exec_count * avg_parse_latency) / SUM(exec_count) AS SIGNED)"` AggMaxParseLatency int64 `json:"max_parse_latency" agg:"MAX(max_parse_latency)"` AggAvgCompileLatency int64 `json:"avg_compile_latency" agg:"CAST(SUM(exec_count * avg_compile_latency) / SUM(exec_count) AS SIGNED)"` AggMaxCompileLatency int64 `json:"max_compile_latency" agg:"MAX(max_compile_latency)"` AggSumCopTaskNum int64 `json:"sum_cop_task_num" agg:"SUM(sum_cop_task_num)"` AggAvgCopProcessTime int64 `json:"avg_cop_process_time" agg:"CAST(SUM(exec_count * avg_process_time) / SUM(sum_cop_task_num) AS SIGNED)"` // avg process time per copr task AggMaxCopProcessTime int64 `json:"max_cop_process_time" agg:"MAX(max_cop_process_time)"` // max process time per copr task AggAvgCopWaitTime int64 `json:"avg_cop_wait_time" agg:"CAST(SUM(exec_count * avg_wait_time) / SUM(sum_cop_task_num) AS SIGNED)"` // avg wait time per copr task AggMaxCopWaitTime int64 `json:"max_cop_wait_time" agg:"MAX(max_cop_wait_time)"` // max wait time per copr task AggAvgProcessTime int64 `json:"avg_process_time" agg:"CAST(SUM(exec_count * avg_process_time) / SUM(exec_count) AS SIGNED)"` // avg total process time per sql AggMaxProcessTime int64 `json:"max_process_time" agg:"MAX(max_process_time)"` // max process time per sql AggAvgWaitTime int64 `json:"avg_wait_time" agg:"CAST(SUM(exec_count * avg_wait_time) / SUM(exec_count) AS SIGNED)"` // avg total wait time per sql AggMaxWaitTime int64 `json:"max_wait_time" agg:"MAX(max_wait_time)"` // max wait time per sql AggAvgBackoffTime int64 `json:"avg_backoff_time" agg:"CAST(SUM(exec_count * avg_backoff_time) / SUM(exec_count) AS SIGNED)"` // avg total back off time per sql AggMaxBackoffTime int64 `json:"max_backoff_time" agg:"MAX(max_backoff_time)"` // max back off time per sql AggAvgTotalKeys int64 `json:"avg_total_keys" agg:"CAST(SUM(exec_count * avg_total_keys) / SUM(exec_count) AS SIGNED)"` AggMaxTotalKeys int64 `json:"max_total_keys" agg:"MAX(max_total_keys)"` AggAvgProcessedKeys int64 `json:"avg_processed_keys" agg:"CAST(SUM(exec_count * avg_processed_keys) / SUM(exec_count) AS SIGNED)"` AggMaxProcessedKeys int64 `json:"max_processed_keys" agg:"MAX(max_processed_keys)"` AggAvgPrewriteTime int64 `json:"avg_prewrite_time" agg:"CAST(SUM(exec_count * avg_prewrite_time) / SUM(exec_count) AS SIGNED)"` AggMaxPrewriteTime int64 `json:"max_prewrite_time" agg:"MAX(max_prewrite_time)"` AggAvgCommitTime int64 `json:"avg_commit_time" agg:"CAST(SUM(exec_count * avg_commit_time) / SUM(exec_count) AS SIGNED)"` AggMaxCommitTime int64 `json:"max_commit_time" agg:"MAX(max_commit_time)"` AggAvgGetCommitTsTime int64 `json:"avg_get_commit_ts_time" agg:"CAST(SUM(exec_count * avg_get_commit_ts_time) / SUM(exec_count) AS SIGNED)"` AggMaxGetCommitTsTime int64 `json:"max_get_commit_ts_time" agg:"MAX(max_get_commit_ts_time)"` AggAvgCommitBackoffTime int64 `json:"avg_commit_backoff_time" agg:"CAST(SUM(exec_count * avg_commit_backoff_time) / SUM(exec_count) AS SIGNED)"` AggMaxCommitBackoffTime int64 `json:"max_commit_backoff_time" agg:"MAX(max_commit_backoff_time)"` AggAvgResolveLockTime int64 `json:"avg_resolve_lock_time" agg:"CAST(SUM(exec_count * avg_resolve_lock_time) / SUM(exec_count) AS SIGNED)"` AggMaxResolveLockTime int64 `json:"max_resolve_lock_time" agg:"MAX(max_resolve_lock_time)"` AggAvgLocalLatchWaitTime int64 `json:"avg_local_latch_wait_time" agg:"CAST(SUM(exec_count * avg_local_latch_wait_time) / SUM(exec_count) AS SIGNED)"` AggMaxLocalLatchWaitTime int64 `json:"max_local_latch_wait_time" agg:"MAX(max_local_latch_wait_time)"` AggAvgWriteKeys int64 `json:"avg_write_keys" agg:"CAST(SUM(exec_count * avg_write_keys) / SUM(exec_count) AS SIGNED)"` AggMaxWriteKeys int64 `json:"max_write_keys" agg:"MAX(max_write_keys)"` AggAvgWriteSize int64 `json:"avg_write_size" agg:"CAST(SUM(exec_count * avg_write_size) / SUM(exec_count) AS SIGNED)"` AggMaxWriteSize int64 `json:"max_write_size" agg:"MAX(max_write_size)"` AggAvgPrewriteRegions int64 `json:"avg_prewrite_regions" agg:"CAST(SUM(exec_count * avg_prewrite_regions) / SUM(exec_count) AS SIGNED)"` AggMaxPrewriteRegions int64 `json:"max_prewrite_regions" agg:"MAX(max_prewrite_regions)"` AggAvgTxnRetry int64 `json:"avg_txn_retry" agg:"CAST(SUM(exec_count * avg_txn_retry) / SUM(exec_count) AS SIGNED)"` AggMaxTxnRetry int64 `json:"max_txn_retry" agg:"MAX(max_txn_retry)"` AggSumBackoffTimes int64 `json:"sum_backoff_times" agg:"SUM(sum_backoff_times)"` AggAvgMem int64 `json:"avg_mem" agg:"CAST(SUM(exec_count * avg_mem) / SUM(exec_count) AS SIGNED)"` AggMaxMem int64 `json:"max_mem" agg:"MAX(max_mem)"` AggAvgDisk int64 `json:"avg_disk" agg:"CAST(SUM(exec_count * avg_disk) / SUM(exec_count) AS SIGNED)"` AggMaxDisk int64 `json:"max_disk" agg:"MAX(max_disk)"` AggAvgAffectedRows int64 `json:"avg_affected_rows" agg:"CAST(SUM(exec_count * avg_affected_rows) / SUM(exec_count) AS SIGNED)"` AggFirstSeen int64 `json:"first_seen" agg:"UNIX_TIMESTAMP(MIN(first_seen))"` AggLastSeen int64 `json:"last_seen" agg:"UNIX_TIMESTAMP(MAX(last_seen))"` AggPlanCount int64 `json:"plan_count" agg:"COUNT(DISTINCT plan_digest)" related:"plan_digest"` // Computed fields outside of the database. M float64 `json:"m,omitempty" gorm:"-"` N float64 `json:"n,omitempty" gorm:"-"` Impact string `json:"impact,omitempty" gorm:"-"` HasSuggestedPlan *bool `json:"has_suggested_plan,omitempty" gorm:"-"` DetailPlans []*Model `json:"detail_plans,omitempty" gorm:"-"` // has many SuggestedPlan *Model `json:"suggested_plan,omitempty" gorm:"-"` // could have one IsSuggested *bool `json:"is_suggested,omitempty" gorm:"-"` } func (m *Model) GetKey() string { k := m.SchemaName + "___" + m.Digest return k } type field struct { ColumnName string JSONName string // `related` tag is used to verify a non-existent column, which is aggregated from the columns represented by related. Related []string Aggregation string } func parseModel() (stmtFields []field) { fields := reflectutil.GetFieldsAndTags(Model{}, []string{"related", "agg", "json"}) for _, f := range fields { sf := field{ ColumnName: gormDefaultNamingStrategy.ColumnName("", f.Name), JSONName: f.Tags["json"], Related: []string{}, Aggregation: f.Tags["agg"], } if f.Tags["related"] != "" { sf.Related = strings.Split(f.Tags["related"], ",") } stmtFields = append(stmtFields, sf) } return } func GenSelectStmt(db *gorm.DB) (string, error) { modelFields := parseModel() // Get table column names from real schema. schemaCols, err := schemaCache.GetTableColumnNames(db, StatementsTable) if err != nil { return "", err } // Filter out columns whose json name does not exist in current version TiDB schema. modelFields = lo.Filter(modelFields, func(f field, _ int) bool { var dependentCols []string if len(f.Related) != 0 { dependentCols = f.Related } else { dependentCols = []string{f.JSONName} } // Dependent columns of the requested field must exist in the db schema. Otherwise, the requested field will be ignored. return utils.IsSubsetCaseInsensitive(schemaCols, dependentCols) }) if len(modelFields) == 0 { return "", errors.Errorf("all columns are not included in the current schema") } // Generate select statement with aggregation. stmt := lo.Map(modelFields, func(f field, _ int) string { if f.Aggregation == "" { return f.JSONName } return fmt.Sprintf("%s AS %s", f.Aggregation, f.ColumnName) }) return strings.Join(stmt, ", "), nil }