package topsqlv2 import ( "bytes" "context" "fmt" "net/http" "os" "sort" "strconv" "strings" "text/template" "time" "github.com/gin-gonic/gin" "github.com/pingcap/log" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/topsql" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/topsql/planmeta" "go.uber.org/zap" "sigs.k8s.io/controller-runtime/pkg/client" ) const ( perClusterQPSLimit = 10.0 perClusterLimiterBucketSize = 10 perClusterQpsLimitDuckDB = 10.0 DefaultTimeout = time.Second * 180 ) type DuckDBConfig struct { Vendor string Region string } func InitSQLsApi(g *gin.RouterGroup, s *Service) { g.GET("/topsql/config", s.configHandler) g.GET("/topsql/instances", s.instancesHandler) g.GET("/topsql/available_fields", s.availableFieldsHandler) g.GET("/topsql/summary", s.summaryHandler) } func NewSQLsService(cacheDir string, store storage.Storage, client client.Client, duckDBConfig DuckDBConfig, topsqlService *topsql.Service) *Service { // Initialize metrics InitMetrics() return &Service{ SQLsDuckDBAdapter: newRateLimitedQuery(NewQueryDuckDBAdapter(duckDBConfig.Vendor, duckDBConfig.Region, store, client), perClusterQpsLimitDuckDB, perClusterLimiterBucketSize), SQLsQuery: newRateLimitedQuery(NewSQLsQuery(store), perClusterQPSLimit, perClusterLimiterBucketSize), store: store, topsqlService: topsqlService, } } type Service struct { SQLsDuckDBAdapter duckDBAdapter SQLsQuery duckDBAdapter store storage.Storage topsqlService *topsql.Service } // errorResponse represents the error details returned to the web front. type errorResponse struct { Error bool `json:"error"` Message string `json:"message"` Code string `json:"code"` FullText string `json:"full_text"` } type summaryEnvelope struct { Status string `json:"status"` Data []SummaryItem `json:"data,omitempty"` DataBy []SummaryByItem `json:"data_by,omitempty"` } type instancesEnvelope struct { Status string `json:"status"` Data []InstanceItem `json:"data"` } type configResponse struct { Enable bool `json:"enable"` } // @Summary Get TopSQL configuration // @Tags topsql // @Description Return the TopSQL feature configuration for diagnosis queries. // @Param data_source query string false "Data source: deltalake enables TopSQLv2 logic; otherwise the legacy TopSQL logic" // @Produce json // @Success 200 {object} configResponse // @Router /topsql/config [get] func (s *Service) configHandler(c *gin.Context) { // Check data_source field: if "deltalake", use topsqlv2 logic, otherwise use topsql dataSource := c.Query("data_source") if strings.ToLower(dataSource) == "deltalake" { // Use topsqlv2 logic (current implementation) c.JSON(http.StatusOK, configResponse{true}) return } // Use topsql service (legacy implementation) if s.topsqlService == nil { log.Warn("topsql service is not available") c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "Internal Server Error", FullText: "topsql service is not available", }) return } // Call topsql service handler s.topsqlService.ConfigHandler(c) } type availableFieldsQuery struct { Instance string `form:"instance" binding:"required"` InstanceType string `form:"instance_type" binding:"required"` DataSource string `form:"data_source"` } func (q *availableFieldsQuery) Validate() error { if q.Instance == "" { return fmt.Errorf("instance not found") } if q.InstanceType == "" { return fmt.Errorf("instance_type not found") } if q.InstanceType != "tidb" && q.InstanceType != "tikv" { return fmt.Errorf("invalid instance_type: %s, must be one of: tidb, tikv", q.InstanceType) } return nil } // @Summary List available TopSQL order_by fields // @Tags topsql // @Description Return order_by field names supported by the given instance's TopSQL Delta Lake schema. Optional dimensions (network/logical/rocksdb) are omitted when their storage columns are absent. // @Produce json // @Param X-Provider header string true "Cloud provider for the query: aws or gcp" // @Param X-Region header string true "Cloud region where TopSQL data is stored" // @Param X-Org-Id header string true "Organization ID (x-org-id) used to locate TopSQL data" // @Param X-Project-Id header string false "Project ID in TiDB Cloud (optional; defaults to empty)" // @Param X-Cluster-Id header string true "Cluster ID (x-cluster-id) of the target TiDB deployment" // @Param X-Deploy-Type header string true "Deployment tier (used to choose storage adapter): dedicated or shared" // @Param instance query string true "Instance name" // @Param instance_type query string true "Instance type: tidb or tikv" // @Param data_source query string false "Must be deltalake for TopSQLv2 schema inspection" // @Success 200 {array} string // @Router /topsql/available_fields [get] func (s *Service) availableFieldsHandler(c *gin.Context) { var header ClusterHeader if err := c.BindHeader(&header); err != nil { log.Warn("failed to bind topsql available fields header", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to bind topsql available fields header", FullText: err.Error(), }) return } if err := header.Validate(); err != nil { log.Warn("invalid topsql available fields header", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid topsql available fields header", FullText: err.Error(), }) return } var query availableFieldsQuery if err := c.BindQuery(&query); err != nil { log.Warn("failed to bind topsql available fields query", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to bind topsql available fields query", FullText: err.Error(), }) return } if err := query.Validate(); err != nil { log.Warn("invalid topsql available fields query", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid topsql available fields query", FullText: err.Error(), }) return } if strings.ToLower(query.DataSource) != "deltalake" { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "data_source must be deltalake", FullText: "available_fields schema inspection is only supported for data_source=deltalake", }) return } env := "prod" if header.Env != nil && *header.Env != "" { env = *header.Env } if *header.DeployType == "byoc" { *header.BizType = "byoc" *header.DeployType = "nextgen" } ctx, cancel := context.WithTimeout(c.Request.Context(), DefaultTimeout) defer cancel() fields, err := s.listAvailableOrderByFields(ctx, &header, env, query.Instance, query.InstanceType) if err != nil { log.Warn("failed to list topsql available fields", zap.Error(err)) c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "failed to list topsql available fields", FullText: err.Error(), }) return } c.JSON(http.StatusOK, fields) } func (s *Service) listAvailableOrderByFields(ctx context.Context, header *ClusterHeader, env, instance, instanceType string) ([]string, error) { sharedTier := *header.DeployType == TierShared nextGen := *header.DeployType == TierNextGen || *header.DeployType == TierPremium bucket := storage.BucketName(env, *header.Region, sharedTier, nextGen) scheme, err := storageScheme(*header.Provider, bucket) if err != nil { return nil, err } tablePath := fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=topsql/component=%s/instance=%s", scheme, *header.TenantID, *header.ClusterID, instanceType, instance) alias := topsqlTableAlias(*header.ClusterID, instanceType, instance) columnNames, err := s.getTopSQLTableColumns(ctx, tablePath, alias) if err != nil { return nil, err } return buildAvailableOrderByFromSchema(columnNameSet(columnNames)), nil } // GetSummaryRequest represents the request parameters for TopSQL summary query type GetSummaryRequest struct { Instance string `json:"instance" form:"instance" binding:"required"` InstanceType string `json:"instance_type" form:"instance_type" binding:"required"` Start string `json:"start" form:"start" binding:"required"` End string `json:"end" form:"end" binding:"required"` Top string `json:"top" form:"top" binding:"required"` GroupBy string `json:"group_by" form:"group_by"` // query, table, db, region OrderBy string `json:"order_by" form:"order_by"` // cpu, network, network_in, network_out, logical_io, logical_read, logical_write, rocksdb_block_read Window string `json:"window" form:"window"` DataSource string `json:"data_source" form:"data_source"` // deltalake or empty (default to topsql) } var validSummaryOrderBy = map[string]struct{}{ "cpu": {}, "network": {}, "network_in": {}, "network_out": {}, "logical_io": {}, "logical_read": {}, "logical_write": {}, "rocksdb_block_read": {}, } func (q *GetSummaryRequest) Validate() error { if q.Start == "" { return fmt.Errorf("start not found") } if q.End == "" { return fmt.Errorf("end not found") } if q.Top == "" { return fmt.Errorf("top not found") } if q.Instance == "" { return fmt.Errorf("instance not found") } if q.InstanceType == "" { return fmt.Errorf("instance_type not found") } // Validate group_by if q.GroupBy != "" && q.GroupBy != "query" && q.GroupBy != "table" && q.GroupBy != "db" && q.GroupBy != "region" { return fmt.Errorf("invalid group_by: %s, must be one of: query, table, db, region", q.GroupBy) } // Validate order_by if q.OrderBy != "" { if _, ok := validSummaryOrderBy[q.OrderBy]; !ok { return fmt.Errorf("invalid order_by: %s, must be one of: cpu, network, network_in, network_out, logical_io, logical_read, logical_write, rocksdb_block_read", q.OrderBy) } } return nil } // SummaryResponse represents the response structure for TopSQL summary type SummaryResponse struct { Data []SummaryItem `json:"data"` DataBy []SummaryByItem `json:"data_by"` } // SummaryItem represents a summary item grouped by query (sql_digest, plan_digest) type SummaryItem struct { SQLDigest string `json:"sql_digest"` SQLText string `json:"sql_text"` IsOther bool `json:"is_other"` // Keyspace is set for premium TiKV summaries so callers can map a query to its TiDB cluster. Keyspace string `json:"keyspace,omitempty"` CPUTimeMs uint64 `json:"cpu_time_ms"` ExecCountPerSec float64 `json:"exec_count_per_sec"` DurationPerExecMs float64 `json:"duration_per_exec_ms"` ScanRecordsPerSec float64 `json:"scan_records_per_sec"` ScanIndexesPerSec float64 `json:"scan_indexes_per_sec"` NetworkBytes uint64 `json:"network_bytes"` NetworkInBytes uint64 `json:"network_in_bytes"` NetworkOutBytes uint64 `json:"network_out_bytes"` LogicalIoBytes uint64 `json:"logical_io_bytes"` LogicalReadBytes uint64 `json:"logical_read_bytes"` LogicalWriteBytes uint64 `json:"logical_write_bytes"` RocksdbBlockReadCount uint64 `json:"rocksdb_block_read_count"` Plans []SummaryPlanItem `json:"plans"` } // SummaryByItem represents a summary item grouped by table/db/region type SummaryByItem struct { Text string `json:"text"` TimestampSec []uint64 `json:"timestamp_sec"` CPUTimeMs []uint64 `json:"cpu_time_ms,omitempty"` CPUTimeMsSum uint64 `json:"cpu_time_ms_sum"` NetworkBytes []uint64 `json:"network_bytes,omitempty"` NetworkBytesSum uint64 `json:"network_bytes_sum"` NetworkInBytes []uint64 `json:"network_in_bytes,omitempty"` NetworkInBytesSum uint64 `json:"network_in_bytes_sum"` NetworkOutBytes []uint64 `json:"network_out_bytes,omitempty"` NetworkOutBytesSum uint64 `json:"network_out_bytes_sum"` LogicalIoBytes []uint64 `json:"logical_io_bytes,omitempty"` LogicalIoBytesSum uint64 `json:"logical_io_bytes_sum"` LogicalReadBytes []uint64 `json:"logical_read_bytes,omitempty"` LogicalReadBytesSum uint64 `json:"logical_read_bytes_sum"` LogicalWriteBytes []uint64 `json:"logical_write_bytes,omitempty"` LogicalWriteBytesSum uint64 `json:"logical_write_bytes_sum"` RocksdbBlockReadCount []uint64 `json:"rocksdb_block_read_count,omitempty"` RocksdbBlockReadCountSum uint64 `json:"rocksdb_block_read_count_sum"` } // SummaryPlanItem represents plan-level summary data type SummaryPlanItem struct { PlanDigest string `json:"plan_digest"` PlanText string `json:"plan_text"` TimestampSec []uint64 `json:"timestamp_sec"` CPUTimeMs []uint64 `json:"cpu_time_ms,omitempty"` ExecCountPerSec float64 `json:"exec_count_per_sec"` DurationPerExecMs float64 `json:"duration_per_exec_ms"` ScanRecordsPerSec float64 `json:"scan_records_per_sec"` ScanIndexesPerSec float64 `json:"scan_indexes_per_sec"` NetworkBytes []uint64 `json:"network_bytes"` NetworkInBytes []uint64 `json:"network_in_bytes"` NetworkOutBytes []uint64 `json:"network_out_bytes"` LogicalIoBytes []uint64 `json:"logical_io_bytes"` LogicalReadBytes []uint64 `json:"logical_read_bytes"` LogicalWriteBytes []uint64 `json:"logical_write_bytes"` RocksdbBlockReadCount []uint64 `json:"rocksdb_block_read_count"` } // InstanceItem represents an instance item type InstanceItem struct { Instance string `json:"instance"` InstanceType string `json:"instance_type"` } // queryResult represents a single row from the SQL query type queryResult struct { SQLDigest *string `db:"sql_digest"` PlanDigest *string `db:"plan_digest"` Keyspace *string `db:"keyspace"` // Premium TiKV: maps to TiDB cluster id GroupKey *string `db:"group_key"` // For non-query groups Text *string `db:"text"` // For SummaryByItem TimeWindow *time.Time `db:"time_window"` TotalCPUMs float64 `db:"total_cpu_ms"` TotalNetworkBytes *float64 `db:"total_network_bytes"` TotalNetworkInBytes *float64 `db:"total_network_in_bytes"` TotalNetworkOutBytes *float64 `db:"total_network_out_bytes"` TotalLogicalIoBytes *float64 `db:"total_logical_io_bytes"` TotalLogicalReadBytes *float64 `db:"total_logical_read_bytes"` TotalLogicalWriteBytes *float64 `db:"total_logical_write_bytes"` TotalRocksdbBlockReadCount *float64 `db:"total_rocksdb_block_read_count"` TotalStmtExecCount *float64 `db:"total_stmt_exec_count"` TotalStmtDurationSumNs *float64 `db:"total_stmt_duration_sum_ns"` TotalStmtDurationCount *float64 `db:"total_stmt_duration_count"` } // sqlMetaResult is a row from topsql_sql_meta. type sqlMetaResult struct { SQLDigest string `db:"sql_digest"` NormalizedSQL string `db:"normalized_sql"` } // planMetaResult is a row from topsql_plan_meta. type planMetaResult struct { PlanDigest string `db:"plan_digest"` NormalizedPlan string `db:"normalized_plan"` EncodedNormalizedPlan string `db:"encoded_normalized_plan"` } // keyspaceResult is used to discover TiKV keyspaces that map to TiDB cluster IDs. type keyspaceResult struct { Keyspace *string `db:"keyspace"` } // duckDBFetcher is an interface that defines the methods needed to query DuckDB // //nolint:unused // Kept for potential future use type duckDBFetcher interface { SelectSlowlogInDL(ctx context.Context, dest interface{}, query string) error SelectSlowlogInDLWithoutFilterPushdown(ctx context.Context, dest interface{}, tableAlias, table, query string) error } // @Summary Get TopSQL summary // @Tags topsql // @Description Return aggregated TopSQL summary for the given time window and instance filter. // @Produce json // @Param X-Provider header string true "Cloud provider for the query: aws or gcp" // @Param X-Region header string true "Cloud region where TopSQL data is stored" // @Param X-Org-Id header string true "Organization ID (x-org-id) used to locate TopSQL data" // @Param X-Project-Id header string false "Project ID in TiDB Cloud (optional; defaults to empty)" // @Param X-Cluster-Id header string true "Cluster ID (x-cluster-id) of the target TiDB deployment" // @Param X-Deploy-Type header string true "Deployment tier (used to choose storage adapter): dedicated or shared" // @Param start query string true "Start time in Unix seconds (inclusive)" // @Param end query string true "End time in Unix seconds (inclusive)" // @Param window query string true "Aggregation window duration (e.g. 5m, 1h)" // @Param top query int true "Top N results to return" // @Param instance query string true "Instance name filter" // @Param instance_type query string true "Instance type filter (e.g. tidb, tikv, pd)" // @Success 200 {object} summaryEnvelope // @Router /topsql/summary [get] func (s *Service) summaryHandler(c *gin.Context) { var header ClusterHeader if err := c.BindHeader(&header); err != nil { log.Warn("failed to bind topsql summary header", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to bind topsql summary header", FullText: err.Error(), }) return } if err := header.Validate(); err != nil { log.Warn("invalid statements list header", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "Bad Request", FullText: err.Error(), }) return } var query GetSummaryRequest if err := c.BindQuery(&query); err != nil { log.Warn("failed to bind topsql summary query", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to bind topsql summary query", FullText: err.Error(), }) return } if err := query.Validate(); err != nil { log.Warn("invalid topsql summary query", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid topsql summary query", FullText: err.Error(), }) return } // Check data_source field: if "deltalake", use topsqlv2 logic, otherwise use topsql if strings.ToLower(query.DataSource) == "deltalake" { // Use topsqlv2 logic (current implementation) if *header.DeployType == "byoc" { *header.BizType = "byoc" *header.DeployType = "nextgen" } ctx, cancel := context.WithTimeout(c.Request.Context(), DefaultTimeout) defer cancel() // Get summary data using DuckDB adapter summaryResponse, err := s.queryTopSQLSummary(ctx, &header, &query) if err != nil { log.Warn("failed to query topsql summary", zap.Error(err)) c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "Internal Server Error", FullText: err.Error(), }) return } // Determine which data to return based on group_by groupBy := query.GroupBy if groupBy == "" { groupBy = "query" } if groupBy == "query" { // Debug: Check plan_text before JSON serialization if len(summaryResponse.Data) > 0 && len(summaryResponse.Data[0].Plans) > 0 { planText := summaryResponse.Data[0].Plans[0].PlanText // Check what's actually in the string sampleLen := 100 if len(planText) < sampleLen { sampleLen = len(planText) } log.Info("Plan text before JSON serialization", zap.String("sample", planText[:sampleLen]), ) } c.JSON(http.StatusOK, summaryEnvelope{ Status: "ok", Data: summaryResponse.Data, }) } else { c.JSON(http.StatusOK, summaryEnvelope{ Status: "ok", DataBy: summaryResponse.DataBy, }) } return } // Use topsql service (legacy implementation) if s.topsqlService == nil { log.Warn("topsql service is not available") c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "Internal Server Error", FullText: "topsql service is not available", }) return } // Update query parameters in the request for topsql service // Note: We need to rebuild the query string since URL.Query() returns a copy queryParams := c.Request.URL.Query() queryParams.Set("start", query.Start) queryParams.Set("end", query.End) queryParams.Set("top", query.Top) queryParams.Set("instance", query.Instance) queryParams.Set("instance_type", query.InstanceType) if query.Window != "" { queryParams.Set("window", query.Window) } c.Request.URL.RawQuery = queryParams.Encode() // Call topsql service handler s.topsqlService.SummaryHandler(c) } type InstanceQuery struct { StartSecs *float64 `form:"start"` EndSecs *float64 `form:"end"` DataSource string `form:"data_source"` // deltalake or empty (default to topsql) } func (q *InstanceQuery) Validate() error { if q.StartSecs == nil { return fmt.Errorf("start not found") } if q.EndSecs == nil { return fmt.Errorf("end not found") } return nil } // @Summary Get available TopSQL instances // @Tags topsql // @Description List TopSQL instances that have data within the given time window. // @Produce json // @Param X-Provider header string true "Cloud provider for the query: aws or gcp" // @Param X-Region header string true "Cloud region where TopSQL data is stored" // @Param X-Org-Id header string true "Organization ID (x-org-id) used to locate TopSQL data" // @Param X-Project-Id header string false "Project ID in TiDB Cloud (optional; defaults to empty)" // @Param X-Cluster-Id header string true "Cluster ID (x-cluster-id) of the target TiDB deployment" // @Param X-Deploy-Type header string true "Deployment tier (used to choose storage adapter): dedicated or shared" // @Param start query string true "Start time in Unix seconds (inclusive)" // @Param end query string true "End time in Unix seconds (inclusive)" // @Success 200 {object} instancesEnvelope // @Router /topsql/instances [get] func (s *Service) instancesHandler(c *gin.Context) { var header ClusterHeader if err := c.BindHeader(&header); err != nil { log.Warn("failed to bind topsql instances header", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to bind topsql instances header", FullText: err.Error(), }) return } if err := header.Validate(); err != nil { log.Warn("invalid topsql instances header", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid topsql instances header", FullText: err.Error(), }) return } var query InstanceQuery if err := c.BindQuery(&query); err != nil { log.Warn("failed to bind topsql instances query", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to bind topsql instances query", FullText: err.Error(), }) return } if err := query.Validate(); err != nil { log.Warn("invalid topsql instances query", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid topsql instances query", FullText: err.Error(), }) return } // Check data_source field: if "deltalake", use topsqlv2 logic, otherwise use topsql if strings.ToLower(query.DataSource) == "deltalake" { env := "prod" if header.Env != nil && *header.Env != "" { env = *header.Env } if *header.DeployType == "byoc" { *header.BizType = "byoc" *header.DeployType = "nextgen" } ctx, cancel := context.WithTimeout(c.Request.Context(), DefaultTimeout) defer cancel() startSecs := int64(*query.StartSecs) endSecs := int64(*query.EndSecs) instances, err := s.listTopSQLInstances(ctx, &header, env, startSecs, endSecs) if err != nil { log.Warn("failed to list topsql instances", zap.Error(err)) c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "failed to list topsql instances", FullText: err.Error(), }) return } c.JSON(http.StatusOK, instancesEnvelope{ Status: "ok", Data: instances, }) return } // Use topsql service (legacy implementation) if s.topsqlService == nil { log.Warn("topsql service is not available") c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "Internal Server Error", FullText: "topsql service is not available", }) return } // Update query parameters in the request for topsql service queryParams := c.Request.URL.Query() if query.StartSecs != nil { queryParams.Set("start", fmt.Sprintf("%.0f", *query.StartSecs)) } if query.EndSecs != nil { queryParams.Set("end", fmt.Sprintf("%.0f", *query.EndSecs)) } c.Request.URL.RawQuery = queryParams.Encode() // Call topsql service handler s.topsqlService.InstancesHandler(c) } type instanceCatalogRow struct { Component string `db:"component"` InstanceID string `db:"instance_id"` } // listTopSQLInstances prefers the drhouse instance catalog table // (type=topsql/component=instance). Falls back to listing object-store // prefixes when the table is missing or has no rows in the time range. func (s *Service) listTopSQLInstances(ctx context.Context, header *ClusterHeader, env string, startSecs, endSecs int64) ([]InstanceItem, error) { sharedTier := *header.DeployType == TierShared nextGen := *header.DeployType == TierNextGen || *header.DeployType == TierPremium bucket := storage.BucketName(env, *header.Region, sharedTier, nextGen) scheme, err := storageScheme(*header.Provider, bucket) if err != nil { return nil, err } instances, err := s.queryInstancesFromCatalog(ctx, scheme, *header.TenantID, *header.ClusterID, startSecs, endSecs) if err != nil { log.Warn("topsql instance catalog unavailable; falling back to object listing", zap.Error(err), zap.String("tenantID", *header.TenantID), zap.String("clusterID", *header.ClusterID)) } else if len(instances) > 0 { return instances, nil } else { log.Info("topsql instance catalog empty; falling back to object listing", zap.String("tenantID", *header.TenantID), zap.String("clusterID", *header.ClusterID)) } return s.listInstancesFromObjectStore(ctx, bucket, *header.TenantID, *header.ClusterID) } func (s *Service) queryInstancesFromCatalog(ctx context.Context, scheme, tenantID, clusterID string, startSecs, endSecs int64) ([]InstanceItem, error) { tablePath := fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=topsql/component=instance", scheme, tenantID, clusterID) startDate, endDate := toUTCDateRange(startSecs, endSecs) sqlQuery := fmt.Sprintf(` SELECT DISTINCT component, instance_id FROM delta_scan('%s') WHERE date >= '%s' AND date <= '%s' AND window_begin BETWEEN %d AND %d AND component IN ('tidb', 'tikv') AND instance_id IS NOT NULL AND instance_id <> '' `, tablePath, startDate, endDate, startSecs, endSecs) var rows []instanceCatalogRow if err := s.SQLsDuckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &rows, "", tablePath, sqlQuery); err != nil { return nil, err } instanceSet := make(map[InstanceItem]struct{}, len(rows)) for _, row := range rows { component := strings.TrimSpace(row.Component) instanceID := strings.TrimSpace(row.InstanceID) if (component != "tidb" && component != "tikv") || instanceID == "" { continue } instanceSet[InstanceItem{ Instance: instanceID, InstanceType: component, }] = struct{}{} } instances := make([]InstanceItem, 0, len(instanceSet)) for item := range instanceSet { instances = append(instances, item) } sort.Slice(instances, func(i, j int) bool { if instances[i].InstanceType != instances[j].InstanceType { return instances[i].InstanceType < instances[j].InstanceType } return instances[i].Instance < instances[j].Instance }) return instances, nil } func (s *Service) listInstancesFromObjectStore(ctx context.Context, bucket, tenantID, clusterID string) ([]InstanceItem, error) { // Path: deltalake/org={tenant}/cluster={cluster}/type=topsql/component={tidb|tikv}/instance=... prefix := fmt.Sprintf("deltalake/org=%s/cluster=%s/type=topsql/component=", tenantID, clusterID) typeDirs, err := s.store.ObjectsDelimiter(ctx, bucket, prefix) if err != nil { return nil, fmt.Errorf("failed to list topsql directories: %w", err) } instanceSet := make(map[InstanceItem]struct{}) for _, dir := range typeDirs { if dir == "" { continue } dirName := strings.TrimPrefix(dir, prefix) dirName = strings.TrimSuffix(dirName, "/") if dirName == "" { continue } instanceType := dirName if instanceType != "tidb" && instanceType != "tikv" { continue } instancePrefix := fmt.Sprintf("deltalake/org=%s/cluster=%s/type=topsql/component=%s/", tenantID, clusterID, instanceType) instanceDirs, err := s.store.ObjectsDelimiter(ctx, bucket, instancePrefix) if err != nil { log.Warn("failed to list topsql instance directories", zap.String("type", instanceType), zap.Error(err)) continue } for _, instanceDir := range instanceDirs { if instanceDir == "" { continue } instanceName := strings.TrimPrefix(instanceDir, instancePrefix) instanceName = strings.TrimSuffix(instanceName, "/") if !strings.HasPrefix(instanceName, "instance=") { continue } instance := strings.TrimPrefix(instanceName, "instance=") if instance == "" { continue } instanceSet[InstanceItem{ Instance: instance, InstanceType: instanceType, }] = struct{}{} } } instances := make([]InstanceItem, 0, len(instanceSet)) for item := range instanceSet { instances = append(instances, item) } sort.Slice(instances, func(i, j int) bool { if instances[i].InstanceType != instances[j].InstanceType { return instances[i].InstanceType < instances[j].InstanceType } return instances[i].Instance < instances[j].Instance }) return instances, nil } func storageScheme(provider, bucket string) (string, error) { switch strings.ToLower(provider) { case "aws": return fmt.Sprintf("s3://%s", bucket), nil case "gcp": return fmt.Sprintf("gs://%s", bucket), nil case "azure": account := os.Getenv("AZURE_STORAGE_ACCOUNT") return fmt.Sprintf("abfss://%s@%s.dfs.core.windows.net", bucket, account), nil case "alicloud": return fmt.Sprintf("s3://%s", bucket), nil case "local": return "./", nil default: return "", fmt.Errorf("unsupported cloud provider: %s", provider) } } // queryTopSQLSummary queries TopSQL summary data from Delta Lake using SQL aggregation with time buckets func (s *Service) queryTopSQLSummary(ctx context.Context, header *ClusterHeader, query *GetSummaryRequest) (*SummaryResponse, error) { // Path: deltalake/org={tenant}/cluster={cluster}/type=topsql_tidb|type=topsql_tikv/instance={instance} env := "prod" if header.Env != nil && *header.Env != "" { env = *header.Env } sharedTier := *header.DeployType == TierShared nextGen := *header.DeployType == TierNextGen || *header.DeployType == TierPremium // Parse start and end times startSecs, err := strconv.ParseInt(query.Start, 10, 64) if err != nil { return nil, fmt.Errorf("invalid start time: %w", err) } endSecs, err := strconv.ParseInt(query.End, 10, 64) if err != nil { return nil, fmt.Errorf("invalid end time: %w", err) } timeRangeSecs := endSecs - startSecs // Parse top top, err := strconv.Atoi(query.Top) if err != nil { return nil, fmt.Errorf("invalid top: %w", err) } // Build table path bucket := storage.BucketName(env, *header.Region, sharedTier, nextGen) var scheme string switch strings.ToLower(*header.Provider) { case "aws": scheme = fmt.Sprintf("s3://%s", bucket) case "gcp": scheme = fmt.Sprintf("gs://%s", bucket) case "azure": account := os.Getenv("AZURE_STORAGE_ACCOUNT") scheme = fmt.Sprintf("abfss://%s@%s.dfs.core.windows.net", bucket, account) case "alicloud": scheme = fmt.Sprintf("s3://%s", bucket) case "local": scheme = "./" default: return nil, fmt.Errorf("unsupported cloud provider: %s", *header.Provider) } // For TopSQL, the path is deltalake/org={tenant}/cluster={cluster}/type=topsql/component={instanceType}/instance={instance} tablePath := fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=topsql/component=%s/instance=%s", scheme, *header.TenantID, *header.ClusterID, query.InstanceType, query.Instance) alias := topsqlTableAlias(*header.ClusterID, query.InstanceType, query.Instance) columnNames, err := s.getTopSQLTableColumns(ctx, tablePath, alias) if err != nil { return nil, err } columns := columnNameSet(columnNames) metricFrags := buildTopSQLMetricSQLFragments(columns) // Calculate time bucket size based on time range // Target: maximum 150 points, minimum 1 minute granularity maxPoints := 150 minBucketSeconds := 60 // 1 minute minimum bucketSizeSeconds := timeRangeSecs / int64(maxPoints) if bucketSizeSeconds < int64(minBucketSeconds) { bucketSizeSeconds = int64(minBucketSeconds) } // Convert to minutes (round up to ensure we don't exceed 150 points) bucketSizeMinutes := (bucketSizeSeconds + 59) / 60 // Round up to minutes if bucketSizeMinutes < 1 { bucketSizeMinutes = 1 } bucketSize := fmt.Sprintf("%d minutes", bucketSizeMinutes) // Determine order field based on order_by parameter. // Combined totals (network / logical_io) are kept; split dimensions are also queryable. // Availability depends on Delta Lake schema (see /topsql/available_fields). orderBy := query.OrderBy if orderBy == "" { orderBy = "cpu" // default } if required, ok := orderByRequiredColumns(orderBy); !ok { return nil, fmt.Errorf("invalid order_by: %s", orderBy) } else if !hasColumns(columns, required...) { return nil, fmt.Errorf("order_by %s is not available in table schema", orderBy) } var orderField string var orderFieldWhere string // For WHERE clause switch orderBy { case "cpu": orderField = "topsql_cpu_time_ms" orderFieldWhere = "topsql_cpu_time_ms" case "network": orderField = "network_bytes" orderFieldWhere = "network_bytes" case "network_in": orderField = "network_in_bytes" orderFieldWhere = "network_in_bytes" case "network_out": orderField = "network_out_bytes" orderFieldWhere = "network_out_bytes" case "logical_io": orderField = "logical_io_bytes" orderFieldWhere = "logical_io_bytes" case "logical_read": orderField = "logical_read_bytes" orderFieldWhere = "logical_read_bytes" case "logical_write": orderField = "logical_write_bytes" orderFieldWhere = "logical_write_bytes" case "rocksdb_block_read": orderField = "rocksdb_block_read_count" orderFieldWhere = "rocksdb_block_read_count" default: return nil, fmt.Errorf("invalid order_by: %s", orderBy) } // Determine group field, default to "query" groupBy := query.GroupBy if groupBy == "" { groupBy = "query" } isQueryGroup := groupBy == "query" // Premium TiKV shares one pool across many TiDB clusters; keyspace identifies the cluster. includeKeyspace := isQueryGroup && strings.EqualFold(query.InstanceType, "tikv") && strings.EqualFold(*header.DeployType, TierPremium) // Build group field selection and grouping clause var groupFieldSelect, groupByClause, groupFieldAlias, unionNullFields, selectFields, regionIdIsNull string if isQueryGroup { // Default: group by sql_digest, plan_digest (+ keyspace for premium TiKV) if includeKeyspace { groupFieldSelect = "sql_digest, plan_digest, keyspace, " groupByClause = "sql_digest, plan_digest, keyspace" selectFields = "sql_digest, plan_digest, keyspace, " unionNullFields = "NULL AS sql_digest, NULL AS plan_digest, NULL AS keyspace, " } else { groupFieldSelect = "sql_digest, plan_digest, " groupByClause = "sql_digest, plan_digest" selectFields = "sql_digest, plan_digest, " unionNullFields = "NULL AS sql_digest, NULL AS plan_digest, " } groupFieldAlias = "sql_digest" regionIdIsNull = "" } else { // Group by table, db, or region var dbFieldName string switch groupBy { case "table": dbFieldName = "\"table\"" case "db": dbFieldName = "db" case "region": dbFieldName = "region_id" regionIdIsNull = "NOT" default: return nil, fmt.Errorf("invalid group_by: %s", groupBy) } // For non-query groups, select the group field as group_key internally groupFieldSelect = fmt.Sprintf("%s AS group_key, ", dbFieldName) groupByClause = "group_key" groupFieldAlias = "group_key" // In final SELECT, return group_key as text for SummaryByItem selectFields = "group_key AS text, " // For SummaryByItem, we use 'text' field unionNullFields = "'other' AS text, " } // Project only metrics present in the instance Delta Lake schema. rawDataSelect := groupFieldSelect + "timestamps, " + strings.Join(metricFrags.RawSelectExprs, ", ") // Determine order by field for final ORDER BY clause orderByField := "sql_digest" if !isQueryGroup { orderByField = "text" } // Prepare template data tmplData := map[string]interface{}{ "RawDataSelect": rawDataSelect, "TablePath": tablePath, "RegionIdIsNull": regionIdIsNull, "OrderFieldWhere": orderFieldWhere, "StartSecs": startSecs, "EndSecs": endSecs, "GroupFieldAlias": groupFieldAlias, "OrderField": orderField, "Top": top, "GroupByClause": groupByClause, "BucketSize": bucketSize, "SelectFields": selectFields, "UnionNullFields": unionNullFields, "OrderByField": orderByField, "BucketSumExprs": strings.Join(metricFrags.BucketSumExprs, ",\n\t\t"), "MetricSumSelect": metricFrags.MetricSumSelect, "MetricSumAgg": metricFrags.MetricSumAgg, } // SQL query template sqlTemplate := ` WITH raw_data AS ( SELECT {{.RawDataSelect}} FROM delta_scan('{{.TablePath}}') WHERE region_id IS {{.RegionIdIsNull}} NULL AND {{.OrderFieldWhere}} > 0 AND timestamps BETWEEN {{.StartSecs}} AND {{.EndSecs}} ), top_groups AS ( SELECT {{.GroupFieldAlias}} FROM ( SELECT {{.GroupFieldAlias}}, SUM({{.OrderField}}) total_order_value FROM raw_data WHERE {{.GroupFieldAlias}} IS NOT NULL GROUP BY {{.GroupFieldAlias}} ORDER BY total_order_value DESC LIMIT {{.Top}} ) subquery1 ), time_bucket_data AS ( SELECT {{.GroupByClause}}, time_bucket('{{.BucketSize}}', to_timestamp(timestamps)) AS time_window, {{.BucketSumExprs}} FROM raw_data GROUP BY {{.GroupByClause}}, time_window ) SELECT * FROM ( SELECT {{.SelectFields}} time_window, {{.MetricSumSelect}}, total_stmt_exec_count, total_stmt_duration_sum_ns, total_stmt_duration_count FROM time_bucket_data WHERE {{.GroupFieldAlias}} IN (SELECT {{.GroupFieldAlias}} FROM top_groups) UNION ALL SELECT {{.UnionNullFields}} time_window, {{.MetricSumAgg}}, SUM(total_stmt_exec_count) AS total_stmt_exec_count, SUM(total_stmt_duration_sum_ns) AS total_stmt_duration_sum_ns, SUM(total_stmt_duration_count) AS total_stmt_duration_count FROM time_bucket_data WHERE {{.GroupFieldAlias}} NOT IN (SELECT {{.GroupFieldAlias}} FROM top_groups) GROUP BY time_window ) ORDER BY {{.OrderByField}}, time_window ASC ` // Parse and execute template tmpl, err := template.New("sqlQuery").Parse(sqlTemplate) if err != nil { return nil, fmt.Errorf("failed to parse SQL template: %w", err) } var sqlQueryBuf bytes.Buffer if err := tmpl.Execute(&sqlQueryBuf, tmplData); err != nil { return nil, fmt.Errorf("failed to execute SQL template: %w", err) } sqlQuery := sqlQueryBuf.String() var queryResults []queryResult err = s.SQLsDuckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &queryResults, "", tablePath, sqlQuery) if err != nil { return nil, fmt.Errorf("failed to query topsql summary: %w", err) } log.Info("Query results received", zap.Int("rowCount", len(queryResults))) if len(queryResults) == 0 { return &SummaryResponse{ Data: []SummaryItem{}, DataBy: []SummaryByItem{}, }, nil } // Collect unique sql_digest and plan_digest for metadata queries (only if group="query") sqlDigestSet := make(map[string]struct{}) planDigestSet := make(map[string]struct{}) if isQueryGroup { for _, row := range queryResults { if row.SQLDigest != nil && *row.SQLDigest != "" { sqlDigestSet[*row.SQLDigest] = struct{}{} } if row.PlanDigest != nil && *row.PlanDigest != "" { planDigestSet[*row.PlanDigest] = struct{}{} } } } sqlMetaMap := make(map[string]string) planMetaMap := make(map[string]string) if isQueryGroup && (len(sqlDigestSet) > 0 || len(planDigestSet) > 0) { metaClusterIDs := s.resolveMetaClusterIDs(ctx, tablePath, query.InstanceType, *header.DeployType, *header.ClusterID, startSecs, endSecs) sqlMetaMap = s.querySQLMetaFromClusters(ctx, scheme, *header.TenantID, metaClusterIDs, startSecs, endSecs, sqlDigestSet) planMetaMap = s.queryPlanMetaFromClusters(ctx, scheme, *header.TenantID, metaClusterIDs, startSecs, endSecs, planDigestSet) } // Handle different return types based on group_by if isQueryGroup { return s.processQueryGroupResults(queryResults, sqlMetaMap, planMetaMap, timeRangeSecs) } else { return s.processByGroupResults(queryResults, timeRangeSecs) } } // resolveMetaClusterIDs returns cluster IDs that hold topsql_sql_meta / topsql_plan_meta. // Only premium TiKV needs keyspace remapping: meta lives under each keyspace's TiDB // cluster directory (cluster=), not under the shared-pool cluster that stores // TiKV metrics. TiDB and dedicated TiKV keep using the request cluster ID. func (s *Service) resolveMetaClusterIDs( ctx context.Context, tablePath string, instanceType string, deployType string, headerClusterID string, startSecs, endSecs int64, ) []string { if !strings.EqualFold(instanceType, "tikv") || !strings.EqualFold(deployType, TierPremium) { return []string{headerClusterID} } keyspaces := s.listTiKVKeyspaces(ctx, tablePath, startSecs, endSecs) if len(keyspaces) == 0 { log.Warn("no keyspaces found in premium TiKV topsql data; sql/plan meta may be empty", zap.String("tablePath", tablePath)) return nil } return keyspaces } func (s *Service) listTiKVKeyspaces(ctx context.Context, tablePath string, startSecs, endSecs int64) []string { keyspaceQuery := fmt.Sprintf(` SELECT DISTINCT keyspace FROM delta_scan('%s') WHERE timestamps BETWEEN %d AND %d AND keyspace IS NOT NULL AND keyspace <> '' `, tablePath, startSecs, endSecs) var rows []keyspaceResult if err := s.SQLsDuckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &rows, "", tablePath, keyspaceQuery); err != nil { log.Warn("Failed to list TiKV keyspaces for topsql meta lookup", zap.Error(err), zap.String("tablePath", tablePath)) return nil } seen := make(map[string]struct{}, len(rows)) keyspaces := make([]string, 0, len(rows)) for _, row := range rows { if row.Keyspace == nil { continue } ks := strings.TrimSpace(*row.Keyspace) if ks == "" { continue } if _, ok := seen[ks]; ok { continue } seen[ks] = struct{}{} keyspaces = append(keyspaces, ks) } sort.Strings(keyspaces) return keyspaces } func (s *Service) querySQLMetaFromClusters( ctx context.Context, scheme string, tenantID string, clusterIDs []string, startSecs, endSecs int64, sqlDigestSet map[string]struct{}, ) map[string]string { sqlMetaMap := make(map[string]string) if len(sqlDigestSet) == 0 || len(clusterIDs) == 0 { return sqlMetaMap } startDate, endDate := toUTCDateRange(startSecs, endSecs) remaining := copyStringSet(sqlDigestSet) for _, clusterID := range clusterIDs { if len(remaining) == 0 { break } sqlMetaPath := fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=topsql/component=topsql_sql_meta", scheme, tenantID, clusterID) sqlDigestInClause := joinQuotedDigests(remaining) sqlMetaQuery := fmt.Sprintf(` SELECT sql_digest, normalized_sql FROM delta_scan('%s') WHERE date >= '%s' AND date <= '%s' AND sql_digest IN (%s) `, sqlMetaPath, startDate, endDate, sqlDigestInClause) var sqlMetaResults []sqlMetaResult if err := s.SQLsDuckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &sqlMetaResults, "", sqlMetaPath, sqlMetaQuery); err != nil { log.Warn("Failed to query SQL metadata", zap.Error(err), zap.String("clusterID", clusterID), zap.String("sqlMetaPath", sqlMetaPath)) continue } for _, meta := range sqlMetaResults { if meta.SQLDigest == "" { continue } if _, exists := sqlMetaMap[meta.SQLDigest]; exists { continue } sqlMetaMap[meta.SQLDigest] = meta.NormalizedSQL delete(remaining, meta.SQLDigest) } } return sqlMetaMap } func (s *Service) queryPlanMetaFromClusters( ctx context.Context, scheme string, tenantID string, clusterIDs []string, startSecs, endSecs int64, planDigestSet map[string]struct{}, ) map[string]string { planMetaMap := make(map[string]string) if len(planDigestSet) == 0 || len(clusterIDs) == 0 { return planMetaMap } startDate, endDate := toUTCDateRange(startSecs, endSecs) remaining := copyStringSet(planDigestSet) for _, clusterID := range clusterIDs { if len(remaining) == 0 { break } planMetaPath := fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=topsql/component=topsql_plan_meta", scheme, tenantID, clusterID) planDigestInClause := joinQuotedDigests(remaining) planMetaQuery := fmt.Sprintf(` SELECT plan_digest, normalized_plan, encoded_normalized_plan FROM delta_scan('%s') WHERE date >= '%s' AND date <= '%s' AND plan_digest IN (%s) `, planMetaPath, startDate, endDate, planDigestInClause) var planMetaResults []planMetaResult if err := s.SQLsDuckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &planMetaResults, "", planMetaPath, planMetaQuery); err != nil { log.Warn("Failed to query plan metadata", zap.Error(err), zap.String("clusterID", clusterID), zap.String("planMetaPath", planMetaPath)) continue } for _, meta := range planMetaResults { if meta.PlanDigest == "" { continue } if _, exists := planMetaMap[meta.PlanDigest]; exists { continue } text, err := planmeta.ResolvePlanText(meta.NormalizedPlan, meta.EncodedNormalizedPlan) if err != nil { log.Warn("Failed to resolve plan text", zap.String("plan_digest", meta.PlanDigest), zap.String("clusterID", clusterID), zap.Error(err)) continue } if text == "" { continue } if len(planMetaMap) == 0 { sampleLen := 100 if len(text) < sampleLen { sampleLen = len(text) } log.Info("Plan text analysis", zap.String("plan_digest", meta.PlanDigest), zap.String("sample", text[:sampleLen]), zap.Bool("from_encoded", strings.TrimSpace(meta.NormalizedPlan) == ""), ) } planMetaMap[meta.PlanDigest] = text delete(remaining, meta.PlanDigest) } } return planMetaMap } func toUTCDateRange(startSecs, endSecs int64) (string, string) { return time.Unix(startSecs, 0).UTC().Format("2006-01-02"), time.Unix(endSecs, 0).UTC().Format("2006-01-02") } func copyStringSet(in map[string]struct{}) map[string]struct{} { out := make(map[string]struct{}, len(in)) for k := range in { out[k] = struct{}{} } return out } func joinQuotedDigests(digests map[string]struct{}) string { list := make([]string, 0, len(digests)) for digest := range digests { escaped := strings.ReplaceAll(digest, "'", "''") list = append(list, fmt.Sprintf("'%s'", escaped)) } sort.Strings(list) return strings.Join(list, ", ") } // processQueryGroupResults processes results when group_by=query func (s *Service) processQueryGroupResults(queryResults []queryResult, sqlMetaMap map[string]string, planMetaMap map[string]string, timeRangeSecs int64) (*SummaryResponse, error) { // Group by sql_digest (+ keyspace when present for premium TiKV) sqlMap := make(map[string]*SummaryItem) // sqlKey -> SQL summary planMap := make(map[string]map[string]*SummaryPlanItem) // sqlKey -> plan_digest -> plan item sqlDurationSumNs := make(map[string]float64) // sqlKey -> total duration sum in ns sqlDurationCount := make(map[string]float64) // sqlKey -> total duration count sqlExecCount := make(map[string]float64) // sqlKey -> total exec count for _, row := range queryResults { // Handle grouping: for query group use sql_digest/plan_digest, for others use group_key var sqlDigest, planDigest, keyspace string // For query group, use sql_digest and plan_digest from query result if row.SQLDigest != nil { sqlDigest = *row.SQLDigest } // For "other" (sql_digest is empty), plan_digest should also be empty string // For normal SQLs, use the plan_digest from query result if sqlDigest == "" { planDigest = "" } else if row.PlanDigest != nil { planDigest = *row.PlanDigest } if row.Keyspace != nil { keyspace = strings.TrimSpace(*row.Keyspace) } sqlKey := summarySQLKey(sqlDigest, keyspace) // Get time window timestamp var timestampSec uint64 if row.TimeWindow != nil { timestampSec = uint64(row.TimeWindow.Unix()) } totalCPUMs := row.TotalCPUMs var totalExecCount float64 if row.TotalStmtExecCount != nil { totalExecCount = *row.TotalStmtExecCount } var stmtDurationSumNs float64 if row.TotalStmtDurationSumNs != nil { stmtDurationSumNs = *row.TotalStmtDurationSumNs } var stmtDurationCount float64 if row.TotalStmtDurationCount != nil { stmtDurationCount = *row.TotalStmtDurationCount } // Get network / logical_io (combined + split) and rocksdb_block_read metrics totalNetworkBytes := uint64FromFloatPtr(row.TotalNetworkBytes) totalNetworkInBytes := uint64FromFloatPtr(row.TotalNetworkInBytes) totalNetworkOutBytes := uint64FromFloatPtr(row.TotalNetworkOutBytes) totalLogicalIoBytes := uint64FromFloatPtr(row.TotalLogicalIoBytes) totalLogicalReadBytes := uint64FromFloatPtr(row.TotalLogicalReadBytes) totalLogicalWriteBytes := uint64FromFloatPtr(row.TotalLogicalWriteBytes) totalRocksdbBlockReadCount := uint64FromFloatPtr(row.TotalRocksdbBlockReadCount) // Initialize SQL summary if not exists if _, ok := sqlMap[sqlKey]; !ok { sqlText := "" if sqlDigest != "" { sqlText = sqlMetaMap[sqlDigest] // Get from metadata map } returnSQLDigest := sqlDigest isOther := sqlDigest == "" sqlMap[sqlKey] = &SummaryItem{ SQLDigest: returnSQLDigest, SQLText: sqlText, IsOther: isOther, Keyspace: keyspace, CPUTimeMs: 0, ExecCountPerSec: 0, DurationPerExecMs: 0, ScanRecordsPerSec: 0, ScanIndexesPerSec: 0, NetworkBytes: 0, NetworkInBytes: 0, NetworkOutBytes: 0, LogicalIoBytes: 0, LogicalReadBytes: 0, LogicalWriteBytes: 0, RocksdbBlockReadCount: 0, Plans: []SummaryPlanItem{}, } planMap[sqlKey] = make(map[string]*SummaryPlanItem) } // Aggregate SQL-level metrics sqlItem := sqlMap[sqlKey] sqlItem.CPUTimeMs += uint64(totalCPUMs) sqlItem.NetworkBytes += totalNetworkBytes sqlItem.NetworkInBytes += totalNetworkInBytes sqlItem.NetworkOutBytes += totalNetworkOutBytes sqlItem.LogicalIoBytes += totalLogicalIoBytes sqlItem.LogicalReadBytes += totalLogicalReadBytes sqlItem.LogicalWriteBytes += totalLogicalWriteBytes sqlItem.RocksdbBlockReadCount += totalRocksdbBlockReadCount sqlExecCount[sqlKey] += totalExecCount // Track duration for SQL-level calculation if stmtDurationCount > 0 && stmtDurationSumNs > 0 { sqlDurationSumNs[sqlKey] += stmtDurationSumNs sqlDurationCount[sqlKey] += stmtDurationCount } // Handle plan-level data (including empty plan_digest for "other") // Initialize plan item if not exists if _, ok := planMap[sqlKey][planDigest]; !ok { planText := "" if planDigest != "" { planText = planMetaMap[planDigest] // Get from metadata map log.Info("Plan text", zap.String("plan_digest", planDigest), zap.String("plan_text", planText)) } planMap[sqlKey][planDigest] = &SummaryPlanItem{ PlanDigest: planDigest, PlanText: planText, TimestampSec: []uint64{}, CPUTimeMs: []uint64{}, ExecCountPerSec: 0, DurationPerExecMs: 0, ScanRecordsPerSec: 0, ScanIndexesPerSec: 0, NetworkBytes: []uint64{}, NetworkInBytes: []uint64{}, NetworkOutBytes: []uint64{}, LogicalIoBytes: []uint64{}, LogicalReadBytes: []uint64{}, LogicalWriteBytes: []uint64{}, RocksdbBlockReadCount: []uint64{}, } } planItem := planMap[sqlKey][planDigest] if timestampSec > 0 { planItem.TimestampSec = append(planItem.TimestampSec, timestampSec) planItem.CPUTimeMs = append(planItem.CPUTimeMs, uint64(totalCPUMs)) planItem.NetworkBytes = append(planItem.NetworkBytes, totalNetworkBytes) planItem.NetworkInBytes = append(planItem.NetworkInBytes, totalNetworkInBytes) planItem.NetworkOutBytes = append(planItem.NetworkOutBytes, totalNetworkOutBytes) planItem.LogicalIoBytes = append(planItem.LogicalIoBytes, totalLogicalIoBytes) planItem.LogicalReadBytes = append(planItem.LogicalReadBytes, totalLogicalReadBytes) planItem.LogicalWriteBytes = append(planItem.LogicalWriteBytes, totalLogicalWriteBytes) planItem.RocksdbBlockReadCount = append(planItem.RocksdbBlockReadCount, totalRocksdbBlockReadCount) } planItem.ExecCountPerSec += totalExecCount / float64(timeRangeSecs) // Calculate duration per exec for this plan if stmtDurationCount > 0 && stmtDurationSumNs > 0 { planItem.DurationPerExecMs = (stmtDurationSumNs / stmtDurationCount) / 1000000.0 } } // Calculate SQL-level metrics for sqlKey, sqlItem := range sqlMap { // Calculate exec_count_per_sec if timeRangeSecs > 0 { sqlItem.ExecCountPerSec = sqlExecCount[sqlKey] / float64(timeRangeSecs) } // Calculate duration_per_exec_ms if sqlDurationCount[sqlKey] > 0 && sqlDurationSumNs[sqlKey] > 0 { sqlItem.DurationPerExecMs = (sqlDurationSumNs[sqlKey] / sqlDurationCount[sqlKey]) / 1000000.0 } // Add plans to SQL item for _, planItem := range planMap[sqlKey] { sqlItem.Plans = append(sqlItem.Plans, *planItem) } } // Convert map to slice and sort by CPU time result := make([]SummaryItem, 0, len(sqlMap)) for _, sqlItem := range sqlMap { result = append(result, *sqlItem) } // Sort by CPU time descending sort.Slice(result, func(i, j int) bool { return result[i].CPUTimeMs > result[j].CPUTimeMs }) return &SummaryResponse{ Data: result, DataBy: []SummaryByItem{}, }, nil } // summarySQLKey separates the same sql_digest across premium TiKV keyspaces (TiDB clusters). func summarySQLKey(sqlDigest, keyspace string) string { if keyspace == "" { return sqlDigest } return sqlDigest + "\x00" + keyspace } // processByGroupResults processes results when group_by=table/db/region func (s *Service) processByGroupResults(queryResults []queryResult, timeRangeSecs int64) (*SummaryResponse, error) { // Group results by text (table/db/region name) byMap := make(map[string]*SummaryByItem) // text -> SummaryByItem for _, row := range queryResults { var text string if row.Text != nil { text = *row.Text } else if row.GroupKey != nil { text = *row.GroupKey } else { text = "other" } // Initialize if not exists if _, ok := byMap[text]; !ok { byMap[text] = &SummaryByItem{ Text: text, TimestampSec: []uint64{}, CPUTimeMs: []uint64{}, CPUTimeMsSum: 0, NetworkBytes: []uint64{}, NetworkBytesSum: 0, NetworkInBytes: []uint64{}, NetworkInBytesSum: 0, NetworkOutBytes: []uint64{}, NetworkOutBytesSum: 0, LogicalIoBytes: []uint64{}, LogicalIoBytesSum: 0, LogicalReadBytes: []uint64{}, LogicalReadBytesSum: 0, LogicalWriteBytes: []uint64{}, LogicalWriteBytesSum: 0, RocksdbBlockReadCount: []uint64{}, RocksdbBlockReadCountSum: 0, } } item := byMap[text] // Get time window timestamp var timestampSec uint64 if row.TimeWindow != nil { timestampSec = uint64(row.TimeWindow.Unix()) } if timestampSec > 0 { item.TimestampSec = append(item.TimestampSec, timestampSec) item.CPUTimeMs = append(item.CPUTimeMs, uint64(row.TotalCPUMs)) } item.CPUTimeMsSum += uint64(row.TotalCPUMs) appendMetricSeries := func(series *[]uint64, sum *uint64, v uint64) { if timestampSec > 0 { *series = append(*series, v) } *sum += v } appendMetricSeries(&item.NetworkBytes, &item.NetworkBytesSum, uint64FromFloatPtr(row.TotalNetworkBytes)) appendMetricSeries(&item.NetworkInBytes, &item.NetworkInBytesSum, uint64FromFloatPtr(row.TotalNetworkInBytes)) appendMetricSeries(&item.NetworkOutBytes, &item.NetworkOutBytesSum, uint64FromFloatPtr(row.TotalNetworkOutBytes)) appendMetricSeries(&item.LogicalIoBytes, &item.LogicalIoBytesSum, uint64FromFloatPtr(row.TotalLogicalIoBytes)) appendMetricSeries(&item.LogicalReadBytes, &item.LogicalReadBytesSum, uint64FromFloatPtr(row.TotalLogicalReadBytes)) appendMetricSeries(&item.LogicalWriteBytes, &item.LogicalWriteBytesSum, uint64FromFloatPtr(row.TotalLogicalWriteBytes)) appendMetricSeries(&item.RocksdbBlockReadCount, &item.RocksdbBlockReadCountSum, uint64FromFloatPtr(row.TotalRocksdbBlockReadCount)) } // Convert map to slice and sort by CPU time sum descending result := make([]SummaryByItem, 0, len(byMap)) for _, item := range byMap { result = append(result, *item) } sort.Slice(result, func(i, j int) bool { return result[i].CPUTimeMsSum > result[j].CPUTimeMsSum }) return &SummaryResponse{ Data: []SummaryItem{}, DataBy: result, }, nil } func uint64FromFloatPtr(v *float64) uint64 { if v == nil { return 0 } return uint64(*v) } // Helper functions to extract values from map[string]interface{} func getStringValue(m map[string]interface{}, key string) string { if v, ok := m[key]; ok && v != nil { if s, ok := v.(string); ok { return s } return fmt.Sprintf("%v", v) } return "" } func getFloat64Value(m map[string]interface{}, key string) float64 { if v, ok := m[key]; ok && v != nil { switch val := v.(type) { case float64: return val case float32: return float64(val) case int64: return float64(val) case int32: return float64(val) case int16: return float64(val) case int8: return float64(val) case int: return float64(val) case uint64: return float64(val) case uint32: return float64(val) case uint16: return float64(val) case uint8: return float64(val) case uint: return float64(val) case string: // Try to parse string as number if f, err := strconv.ParseFloat(val, 64); err == nil { return f } default: // Try to convert via fmt.Sprintf and parse str := fmt.Sprintf("%v", val) if f, err := strconv.ParseFloat(str, 64); err == nil { return f } } } return 0 } // isValidFieldName validates that a field name is safe to use in SQL (prevents SQL injection) func isValidFieldName(fieldName string) bool { // Only allow alphanumeric characters, underscores, and dots // Field names should not contain spaces, special characters, or SQL keywords for _, r := range fieldName { if !((r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' || r == '.') { return false } } return len(fieldName) > 0 } func getInt64Value(m map[string]interface{}, key string) int64 { if v, ok := m[key]; ok && v != nil { switch val := v.(type) { case int64: return val case int: return int64(val) case float64: return int64(val) case uint64: return int64(val) } } return 0 }