package sqls

import (
	"context"
	"fmt"
	"net/http"
	"strconv"
	"strings"

	"github.com/gin-gonic/gin"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb/pkg/util/plancodec"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
	"go.uber.org/zap"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	perClusterQPSLimit          = 10.0
	perClusterLimiterBucketSize = 10
	maxListQueryTimeRange       = 3*60*60*24 + 5*60  // 3d (+5min)
	maxDuckdbListQueryTimeRange = 30*60*60*24 + 5*60 // 3d (+5min)
	perClusterQpsLimitDuckDB    = 10.0
)

type DuckDBConfig struct {
	Vendor string
	Region string
}

func InitSQLsApi(g *gin.RouterGroup, s *Service) {
	g.GET("/SQLs/", s.handleTableList)
	g.GET("/SQLs/:name/fields", s.handleTableFields)
	g.GET("/SQLs/:name/data", s.handleTableData)
	g.GET("/SQLs/:name/data/:id", s.handleTableDataByID)
	// g.GET("/SQLs/:name/topN", s.handleTableTopN)
	// g.GET("/SQLs/:name/timeseries", s.handleTableStatsMetric)
	// g.GET("/SQLs/query", s.handleCustomQuery)
	g.GET("/statements/list", s.StatementsListHandler)
	g.GET("/statements/plan/detail", s.StatementsPlanDetailHandler)
	g.GET("/statements/plans", s.StatementsPlansHandler)
	g.GET("/statements/stmt_types", s.StatementsStmtTypesHandler)
	g.GET("/statements/available_fields", s.StatementsAvailableFieldsHandler)
	g.GET("/statements/time_ranges", s.StatementsTimeRangesHandler)
	g.GET("/statements/summary", s.StatementsSummaryHandler)
	g.GET("/statements/topstats", s.StatementsTopStatsHandler)
	g.GET("/statements/field_values", s.StatementsFieldValuesHandler)
	g.GET("/statements/datasource", s.StatementsDataSourceHandler)

}

func NewSQLsService(cacheDir string, store storage.Storage, client client.Client, duckDBConfig DuckDBConfig) *Service {
	// Initialize metrics
	InitMetrics()

	return &Service{
		SQLsDuckDBAdapter: newRateLimitedQuery(NewQueryDuckDBAdapter(duckDBConfig.Vendor, duckDBConfig.Region, store, client), perClusterQpsLimitDuckDB, perClusterLimiterBucketSize),
		SQLsQuery:         newRateLimitedQuery(NewSQLsQuery(store), perClusterQPSLimit, perClusterLimiterBucketSize),
	}
}

type Service struct {
	SQLsDuckDBAdapter duckDBAdapter
	SQLsQuery         duckDBAdapter
}

// 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 clusterHeader struct {
	Provider   string `header:"x-provider" binding:"required"`
	Region     string `header:"x-region" binding:"required"`
	TenantID   string `header:"x-org-id"`
	ProjectID  string `header:"x-project-id"`
	ClusterID  string `header:"x-cluster-id" binding:"required"`
	DeployType string `header:"x-deploy-type"` // dedicated/shared/nextgen/premium
	Env        string `header:"x-env"`         // Use default env if emprty
	BizType    string `header:"x-biz-type"`    // byoc/other
}

type listRequest struct {
	Type        string `json:"type" form:"type"`
	DataSource  string `json:"dataSource" form:"dataSource"`
	DataSource2 string `json:"data_source" form:"data_source"`
}

type dataListRequest struct {
	StartTime   string   `form:"StartTime"`
	EndTime     string   `form:"EndTime"`
	Fields      []string `form:"Fields" binding:"required"`
	Filter      string   `form:"Filter"`
	GroupBy     string   `form:"GroupBy"`
	Sort        string   `form:"Sort"` // Sort field and order, e.g., "Query_time:desc" or "Time:asc"
	Limit       int      `form:"Limit"`
	Offset      int      `form:"Offset"`
	DataSource  string   `form:"dataSource"`
	DataSource2 string   `form:"data_source"`
}

// 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
}

// handleTableList handles the GET /SQLs/ endpoint to list all available tables
func (s *Service) handleTableList(c *gin.Context) {
	userAgent := c.Request.Header.Get("User-Agent")
	log.Debug("Received request with User-Agent", zap.String("User-Agent", userAgent))

	f := recordHttpMetric(c, "/tablelist")
	defer f()

	var header clusterHeader
	if err := c.ShouldBindHeader(&header); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid header",
			FullText: err.Error(),
		})
		return
	}
	var req listRequest
	if err := c.ShouldBindQuery(&req); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid params",
			FullText: err.Error(),
		})
		return
	}

	env := "prod"
	if header.Env != "" {
		env = header.Env
	}
	if header.DeployType == "byoc" {
		header.BizType = "byoc"
		header.DeployType = "nextgen"
	}
	sharedTier := strings.ToLower(header.DeployType) == "shared"
	nextGen := (strings.ToLower(header.DeployType) == "nextgen") || (strings.ToLower(header.DeployType) == "premium")
	params := &ListTableParams{
		Env:        env,
		Provider:   header.Provider,
		Region:     header.Region,
		TenantID:   header.TenantID,
		ProjectID:  header.ProjectID,
		ClusterID:  header.ClusterID,
		SharedTier: sharedTier,
		NextGen:    nextGen,
		Byoc:       header.BizType == "byoc",
		Type:       req.Type,
		DataSource: normalizeDataSource(req.DataSource, req.DataSource2),
	}

	var err error
	var tables []string

	tables, err = s.SQLsDuckDBAdapter.ListTables(c.Request.Context(), params)

	if err != nil {
		log.Error("failed to list sqls tables", zap.Error(err))
		return
	}
	rs := make([]string, 0, len(tables))
	for _, sl := range tables {
		rs = append(rs, sl)
	}
	c.JSON(http.StatusOK, rs)
}

// TableField represents a field in a table
type TableField struct {
	Name        string `json:"name"`
	Type        string `json:"type"`
	Description string `json:"description,omitempty"`
}

// TableFieldsResponse represents the response for listing table fields
type TableFieldsResponse struct {
	Fields []TableField `json:"fields"`
}

// DataListResponse represents the response for listing data
type DataListResponse struct {
	Data  []map[string]interface{} `json:"data"`
	Total int64                    `json:"total"`
}

// handleTableFields handles the GET /SQLs/:name/fields endpoint to list all fields of a table
func (s *Service) handleTableFields(c *gin.Context) {
	userAgent := c.Request.Header.Get("User-Agent")
	log.Debug("Received request with User-Agent", zap.String("User-Agent", userAgent))

	f := recordHttpMetric(c, "/tablefields")
	defer f()

	// Get table name from URL parameter
	tableName := c.Param("name")
	if tableName == "" {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:   true,
			Message: "table name is required",
		})
		return
	}

	// Parse cluster headers
	var header clusterHeader
	if err := c.ShouldBindHeader(&header); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid header",
			FullText: err.Error(),
		})
		return
	}

	env := "prod"
	if header.Env != "" {
		env = header.Env
	}
	if header.DeployType == "byoc" {
		header.BizType = "byoc"
		header.DeployType = "nextgen"
	}
	sharedTier := strings.ToLower(header.DeployType) == "shared"
	nextGen := (strings.ToLower(header.DeployType) == "nextgen") || (strings.ToLower(header.DeployType) == "premium")

	// Prepare the ListTableFieldsParams
	dataSource := normalizeDataSource(c.Query("dataSource"), c.Query("data_source"))
	params := &ListTableFieldsParams{
		Env:        env,
		Provider:   header.Provider,
		Region:     header.Region,
		TenantID:   header.TenantID,
		ProjectID:  header.ProjectID,
		ClusterID:  header.ClusterID,
		SharedTier: sharedTier,
		NextGen:    nextGen,
		Byoc:       header.BizType == "byoc",
		TableName:  tableName,
		DataSource: dataSource,
	}

	// Call ListTableFields to get the field information
	fieldInfos, err := s.SQLsDuckDBAdapter.ListTableFields(c.Request.Context(), params)
	if err != nil {
		log.Error("failed to list table fields",
			zap.String("table", tableName),
			zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "failed to get table fields",
			FullText: err.Error(),
		})
		return
	}

	// Convert field information to TableField objects
	fields := make([]TableField, 0, len(fieldInfos))
	for _, fieldInfo := range fieldInfos {
		fields = append(fields, TableField{
			Name: fieldInfo.Name,
			Type: fieldInfo.Type,
		})
	}

	// Return the fields
	c.JSON(http.StatusOK, TableFieldsResponse{
		Fields: fields,
	})
}

// handleTableData handles the GET /SQLs/:name/data endpoint to query raw data
func (s *Service) handleTableData(c *gin.Context) {
	userAgent := c.Request.Header.Get("User-Agent")
	log.Debug("Received request with User-Agent", zap.String("User-Agent", userAgent))

	f := recordHttpMetric(c, "/tabledata")
	defer f()

	// Get table name from URL parameter
	tableName := c.Param("name")
	if tableName == "" {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:   true,
			Message: "table name is required",
		})
		return
	}

	// Parse cluster headers
	var header clusterHeader
	if err := c.ShouldBindHeader(&header); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid header",
			FullText: err.Error(),
		})
		return
	}

	// Parse query parameters
	var req dataListRequest
	if err := c.ShouldBindQuery(&req); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid params",
			FullText: err.Error(),
		})
		return
	}

	// Validate required parameters
	if (req.StartTime == "" || req.EndTime == "") && req.Filter == "" {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:   true,
			Message: "startTime and endTime are required while filter is empty",
		})
		return
	}

	if len(req.Fields) == 0 {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:   true,
			Message: "fields parameter is required",
		})
		return
	}

	env := "prod"
	if header.Env != "" {
		env = header.Env
	}
	if header.DeployType == "byoc" {
		header.BizType = "byoc"
		header.DeployType = "nextgen"
	}
	sharedTier := strings.ToLower(header.DeployType) == "shared"
	nextGen := (strings.ToLower(header.DeployType) == "nextgen") || (strings.ToLower(header.DeployType) == "premium")

	// Prepare the ListTableDataParams
	params := &ListTableDataParams{
		Env:        env,
		Provider:   header.Provider,
		Region:     header.Region,
		TenantID:   header.TenantID,
		ProjectID:  header.ProjectID,
		ClusterID:  header.ClusterID,
		SharedTier: sharedTier,
		NextGen:    nextGen,
		Byoc:       header.BizType == "byoc",
		TableName:  tableName,
		StartTime:  req.StartTime,
		EndTime:    req.EndTime,
		Fields:     req.Fields,
		GroupBy:    req.GroupBy,
		Filter:     req.Filter,
		Sort:       req.Sort,
		Limit:      req.Limit,
		Offset:     req.Offset,
		DataSource: normalizeDataSource(req.DataSource, req.DataSource2),
	}

	// Call ListTableData to get the data
	response, err := s.SQLsDuckDBAdapter.ListTableData(c.Request.Context(), params)
	if err != nil {
		log.Error("failed to list table data",
			zap.String("table", tableName),
			zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "failed to get table data",
			FullText: err.Error(),
		})
		return
	}

	// Return the data
	c.JSON(http.StatusOK, response)
}

// handleTableDataByID handles the GET /SQLs/:name/data/:id endpoint to get a single record by ID
func (s *Service) handleTableDataByID(c *gin.Context) {
	userAgent := c.Request.Header.Get("User-Agent")
	log.Debug("Received request with User-Agent", zap.String("User-Agent", userAgent))

	f := recordHttpMetric(c, "/tabledataid")
	defer f()

	// Get table name and ID from URL parameters
	tableName := c.Param("name")
	if tableName == "" {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:   true,
			Message: "table name is required",
		})
		return
	}

	idStr := c.Param("id")
	id, err := strconv.ParseInt(idStr, 10, 64)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:   true,
			Message: "invalid ID parameter",
		})
		return
	}

	// Parse cluster headers
	var header clusterHeader
	if err := c.ShouldBindHeader(&header); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid header",
			FullText: err.Error(),
		})
		return
	}

	env := "prod"
	if header.Env != "" {
		env = header.Env
	}
	if header.DeployType == "byoc" {
		header.BizType = "byoc"
		header.DeployType = "nextgen"
	}
	sharedTier := strings.ToLower(header.DeployType) == "shared"
	nextGen := (strings.ToLower(header.DeployType) == "nextgen") || (strings.ToLower(header.DeployType) == "premium")

	// Prepare the GetTableDataByIDParams
	dataSource := normalizeDataSource(c.Query("dataSource"), c.Query("data_source"))
	params := &GetTableDataByIDParams{
		Env:        env,
		Provider:   header.Provider,
		Region:     header.Region,
		TenantID:   header.TenantID,
		ProjectID:  header.ProjectID,
		ClusterID:  header.ClusterID,
		SharedTier: sharedTier,
		NextGen:    nextGen,
		Byoc:       header.BizType == "byoc",
		TableName:  tableName,
		ID:         id,
		DataSource: dataSource,
	}

	// Call GetTableDataByID to get the record
	record, err := s.SQLsDuckDBAdapter.GetTableDataByID(c.Request.Context(), params)
	if err != nil {
		log.Error("failed to get table data by ID",
			zap.String("table", tableName),
			zap.Int64("id", id),
			zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "failed to get table data",
			FullText: err.Error(),
		})
		return
	}

	// Return the record
	c.JSON(http.StatusOK, record)
}

func parsePlan(plan, binaryPlan string) string {
	var r string
	var err error
	if binaryPlan != "" && strings.HasPrefix(binaryPlan, "tidb_decode_binary_plan('") {
		binaryPlan = strings.TrimPrefix(binaryPlan, "tidb_decode_binary_plan('")
		binaryPlan = strings.TrimSuffix(binaryPlan, "')")
		r, err = plancodec.DecodeBinaryPlan(binaryPlan)
	} else if plan != "" && strings.HasPrefix(plan, "tidb_decode_plan('") {
		plan = strings.TrimPrefix(plan, "tidb_decode_plan('")
		plan = strings.TrimSuffix(plan, "')")
		r, err = plancodec.DecodePlan(plan)
	} else {
		r = plan
	}
	if err != nil {
		log.Error("decode plan in slow log failed", zap.String("plan", plan), zap.String("binaryPlan", binaryPlan), zap.Error(err))
	}
	return r
}

var STATEMENTS_TABLE_NAME = "persisted_statements_summary"

type statementsAvailableFieldsQuery struct {
	// usage=list (default): all columns for /statements/list fields param.
	// usage=filter: fields valid for /statements/field_values and summary/topstats field+values filters.
	Usage string `form:"usage"`
}

// @Summary		List selectable fields for statements
// @Tags			statements
// @Description	Returns field names for statements APIs. usage=list (default): columns for /statements/list fields param; usage=filter: fields for /statements/field_values and field+values filters.
// @Produce		json
// @Param			X-Provider		header		string		true	"Cloud provider"
// @Param			X-Region		header		string		true	"Cloud region"
// @Param			X-Org-Id		header		string		true	"Organization ID"
// @Param			X-Project-Id	header		string		false	"Project ID"
// @Param			X-Cluster-Id	header		string		true	"Cluster ID"
// @Param			X-Deploy-Type	header		string		true	"Deployment tier"
// @Param			usage			query		string		false	"list or filter (default: list)"
// @Success		200	{array}		string
// @Router			/statements/available_fields [get]
func (s *Service) StatementsAvailableFieldsHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind statements available fields header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Invalid Request",
			FullText: err.Error(),
		})
		return
	}
	if err := header.Validate(); err != nil {
		log.Warn("invalid statements available fields header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}

	var q statementsAvailableFieldsQuery
	_ = c.BindQuery(&q)

	var source []string
	switch strings.ToLower(strings.TrimSpace(q.Usage)) {
	case "filter":
		source = StatementsFilterableFields
	default:
		source = StatementAvailableFields
	}

	fields := make([]string, len(source))
	copy(fields, source)

	c.JSON(http.StatusOK, fields)
}

// appendUniqueFields appends fields to a slice if they don't already exist
func appendUniqueFields(slice []string, elems []string) []string {
	for _, elem := range elems {
		found := false
		for _, s := range slice {
			if s == elem {
				found = true
				break
			}
		}
		if !found {
			slice = append(slice, elem)
		}
	}
	return slice
}

func convertStatementFields(fields []string, dataSource string) []string {
	if len(fields) == 0 {
		return nil
	}

	convertedFields := make([]string, 0, len(fields))
	for _, field := range fields {
		if sqlExpr, exists := sqlsFieldMap(dataSource)[field]; exists {
			convertedFields = append(convertedFields, sqlExpr)
		} else {
			// If no mapping exists, use the field name directly
			convertedFields = append(convertedFields, field)
		}
	}
	return convertedFields
}

// @Summary		Get statements time ranges
// @Tags			statements
// @Description	Placeholder endpoint for statements time range metadata. Currently returns an empty response.
// @Produce		plain
// @Param			X-Provider		header		string	true	"Cloud provider"
// @Param			X-Region		header		string	true	"Cloud region"
// @Param			X-Org-Id		header		string	true	"Organization ID"
// @Param			X-Project-Id	header		string	false	"Project ID"
// @Param			X-Cluster-Id	header		string	true	"Cluster ID"
// @Param			X-Deploy-Type	header		string	true	"Deployment tier"
// @Success		200	{string}	string
// @Router			/statements/time_ranges [get]
func (s *Service) StatementsTimeRangesHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind statements time ranges header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Invalid Request",
			FullText: err.Error(),
		})
		return
	}
	if err := header.Validate(); err != nil {
		log.Warn("invalid statements time ranges header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	c.String(http.StatusOK, "")
}

// @Summary		List statement types
// @Tags			statements
// @Description	Returns supported SQL statement types for statements filtering.
// @Produce		json
// @Param			X-Provider		header		string	true	"Cloud provider"
// @Param			X-Region		header		string	true	"Cloud region"
// @Param			X-Org-Id		header		string	true	"Organization ID"
// @Param			X-Project-Id	header		string	false	"Project ID"
// @Param			X-Cluster-Id	header		string	true	"Cluster ID"
// @Param			X-Deploy-Type	header		string	true	"Deployment tier"
// @Success		200	{array}		string
// @Router			/statements/stmt_types [get]
func (s *Service) StatementsStmtTypesHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind statements stmt types header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	if err := header.Validate(); err != nil {
		log.Warn("invalid statements stmt types header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}

	c.JSON(http.StatusOK, StatementsStmtTypes)
}

type StatementsListQuery struct {
	BeginTime   *int     `form:"begin_time"`
	EndTime     *int     `form:"end_time"`
	Schemas     []string `form:"schemas"`
	StmtTypes   []string `form:"stmt_types"`
	Text        string   `form:"text"`
	SampleUser  string   `form:"sample_user"`
	SampleUsers []string `form:"sample_users"`
	Tables      []string `form:"tables"`
	Field       string   `form:"field"`
	Values      []string `form:"values"`
	Filters     string   `form:"filters"`
	Fields      string   `form:"fields"`
	DataSource  string   `form:"dataSource"`
	DataSource2 string   `form:"data_source"`
}

// @Summary		List statement overview
// @Tags			statements
// @Description	Return an overview of SQL statements within a time window. The returned columns depend on the `fields` parameter.
// @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 statement data is stored"
// @Param			X-Org-Id		header		string		true	"Organization ID (x-org-id)"
// @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: dedicated or shared"
// @Param			begin_time		query		int			true	"Start time in Unix seconds (inclusive)"
// @Param			end_time		query		int			true	"End time in Unix seconds (inclusive)"
// @Param			schemas			query		[]string	false	"filter by schemas"
// @Param			stmt_types		query		[]string	false	"filter by stmt_types"
// @Param			text			query		string		false	"Substring filter on SQL text or digest"
// @Param			sample_user		query		string		false	"Filter rows by user (v2: key in auth_users; v1: exact sample_user match); use sample_users for multi-select"
// @Param			sample_users	query		[]string	false	"Filter rows by any of the users (supports sample_users[] / users[])"
// @Param			tables			query		[]string	false	"Filter rows by table_names (supports tables[] / table_names[])"
// @Param			field			query		string		false	"Filter rows by field; see available_fields?usage=filter"
// @Param			values			query		[]string	false	"Values for field (supports values[])"
// @Param			filters			query		string		false	"Additional field filters as JSON array"
// @Param			fields			query		string		false	"Comma-separated list of fields to return. Default includes summary fields."
// @Success		200	{object}	object	"list of statement rows (schema varies by fields param)"
// @Router			/statements/list [get]
func (s *Service) StatementsListHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind statements list header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			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 StatementsListQuery
	if err := c.BindQuery(&query); err != nil {
		log.Warn("failed to bind statements list query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	bindSchemasAndStmtTypes(c, &query.Schemas, &query.StmtTypes)
	bindStatementsUsers(c, &query.SampleUsers)
	bindStatementsTables(c, &query.Tables)
	bindStatementsValues(c, &query.Values)
	extraFilters, err := parseStatementsFiltersJSON(query.Filters)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	filterOpts := statementsFilterOptions{
		beginTime:    *query.BeginTime,
		endTime:      *query.EndTime,
		schemas:      query.Schemas,
		stmtTypes:    query.StmtTypes,
		text:         query.Text,
		sampleUser:   query.SampleUser,
		sampleUsers:  query.SampleUsers,
		tables:       query.Tables,
		field:        query.Field,
		values:       query.Values,
		extraFilters: extraFilters,
	}
	if err := validateStatementsQueryFilters(filterOpts); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	var fields []string
	// Get the base fields from query
	if query.Fields != "" {
		fields = strings.Split(query.Fields, ",")
	}

	// Add required summary time fields if not already present
	requiredFields := []string{"summary_begin_time", "summary_end_time", "schema_name", "sum_latency", "digest"}
	fields = appendUniqueFields(fields, requiredFields)
	// Convert statement fields to SQL expressions
	dataSource := normalizeDataSource(query.DataSource, query.DataSource2)
	fields = convertStatementFields(fields, dataSource)

	filterOpts.dataSource = dataSource
	filter := buildStatementsFilter(filterOpts)

	if *header.DeployType == "byoc" {
		*header.BizType = "byoc"
		*header.DeployType = "nextgen"
	}
	params := &ListTableDataParams{
		Env:        *header.Env,
		Provider:   *header.Provider,
		Region:     *header.Region,
		TenantID:   *header.TenantID,
		ProjectID:  *header.ProjectID,
		ClusterID:  *header.ClusterID,
		SharedTier: *header.DeployType == TierShared,
		NextGen:    *header.DeployType == TierNextGen || *header.DeployType == TierPremium,
		Byoc:       *header.BizType == "byoc",
		TableName:  STATEMENTS_TABLE_NAME,
		StartTime:  "",
		EndTime:    "",
		Fields:     fields,
		GroupBy:    "digest",
		Filter:     filter,
		Sort:       "",
		Limit:      100000,
		Offset:     0,
		DataSource: dataSource,
	}

	// Call ListTableData to get the data
	res, err := s.SQLsDuckDBAdapter.ListTableData(c.Request.Context(), params)
	if err != nil {
		log.Warn("failed to query statements list", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "Internal Server Error",
			FullText: err.Error(),
		})
		return
	}

	formatStatementsListRowsForAPI(res.Data, dataSource)
	c.JSON(http.StatusOK, res.Data)
}

type StatementsPlansQuery struct {
	BeginTime   *int   `form:"begin_time"`
	EndTime     *int   `form:"end_time"`
	SchemaName  string `form:"schema_name"`
	Digest      string `form:"digest"`
	DataSource  string `form:"dataSource"`
	DataSource2 string `form:"data_source"`
}

// @Summary		List execution plans for a statement
// @Tags			statements
// @Description	Returns plan-level aggregates for a SQL digest within a time window. One row per plan_digest.
// @Produce		json
// @Param			X-Provider		header		string		true	"Cloud provider"
// @Param			X-Region		header		string		true	"Cloud region"
// @Param			X-Org-Id		header		string		true	"Organization ID"
// @Param			X-Project-Id	header		string		false	"Project ID"
// @Param			X-Cluster-Id	header		string		true	"Cluster ID"
// @Param			X-Deploy-Type	header		string		true	"Deployment tier"
// @Param			begin_time		query		int			true	"Start time in Unix seconds (inclusive)"
// @Param			end_time		query		int			true	"End time in Unix seconds (inclusive)"
// @Param			schema_name		query		string		false	"Filter by schema name (substring match)"
// @Param			digest			query		string		true	"SQL digest"
// @Param			dataSource		query		string		false	"Data source identifier"
// @Param			data_source		query		string		false	"Data source identifier (alias)"
// @Success		200	{array}		object	"plan rows grouped by digest and plan_digest"
// @Router			/statements/plans [get]
func (s *Service) StatementsPlansHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind statements plans header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	if err := header.Validate(); err != nil {
		log.Warn("invalid statements plans header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}

	var query StatementsPlansQuery
	if err := c.BindQuery(&query); err != nil {
		log.Warn("failed to bind statements plans query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}

	dataSource := normalizeDataSource(query.DataSource, query.DataSource2)

	if *header.DeployType == "byoc" {
		*header.BizType = "byoc"
		*header.DeployType = "nextgen"
	}

	fieldInfos, err := s.SQLsDuckDBAdapter.ListTableFields(c.Request.Context(), &ListTableFieldsParams{
		Env:        *header.Env,
		Provider:   *header.Provider,
		Region:     *header.Region,
		TenantID:   *header.TenantID,
		ProjectID:  *header.ProjectID,
		ClusterID:  *header.ClusterID,
		SharedTier: *header.DeployType == TierShared,
		NextGen:    *header.DeployType == TierNextGen || *header.DeployType == TierPremium,
		Byoc:       *header.BizType == "byoc",
		TableName:  STATEMENTS_TABLE_NAME,
		DataSource: dataSource,
	})
	if err != nil {
		log.Warn("failed to list statements table fields for plans", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "Internal Server Error",
			FullText: err.Error(),
		})
		return
	}

	requestedFields := []string{
		"summary_begin_time", "summary_end_time", "digest_text", "exec_count", "stmt_type",
		"sum_latency", "max_latency", "min_latency", "avg_latency",
		"avg_mem", "max_mem", "last_seen", "schema_name", "plan_digest", "plan_hint", "plan_in_binding",
		"sum_ru", "avg_ru", "sum_rru", "sum_wru", "max_ru",
		"sum_ru_v2", "avg_ru_v2", "max_ru_v2",
		"sum_tidb_cpu", "avg_tidb_cpu", "sum_tikv_cpu", "avg_tikv_cpu",
		"plan_cache_hits", "plan_in_cache",
		"plan_cache_unqualified_count", "plan_cache_unqualified_last_reason",
		"sum_ia_read_segment_count", "avg_ia_read_segment_count", "max_ia_read_segment_count",
		"sum_ia_remote_read_segment_size", "avg_ia_remote_read_segment_size", "max_ia_remote_read_segment_size",
		"sum_ia_remote_read_segment_wait_time", "avg_ia_remote_read_segment_wait_time", "max_ia_remote_read_segment_wait_time",
	}
	fields := filterStatementsRequestedFieldNames(requestedFields, dataSource, columnNameSet(fieldInfos))
	fields = convertStatementFields(fields, dataSource)
	fields = replaceStatementsPlanDigestSelectExpr(fields)
	fields = append(fields, "any_value(digest) as digest")

	filter := fmt.Sprintf("digest = '%s' and %s", query.Digest, buildStatementsSummaryBaseFilter(*query.BeginTime, *query.EndTime, nil, nil, "", dataSource))
	filter = appendStatementsSchemaNameQueryFilter(filter, query.SchemaName, dataSource)
	// Group by digest and plan_digest so that each plan is a separate row (multiple plans -> multiple rows in response)
	params := &ListTableDataParams{
		Env:        *header.Env,
		Provider:   *header.Provider,
		Region:     *header.Region,
		TenantID:   *header.TenantID,
		ProjectID:  *header.ProjectID,
		ClusterID:  *header.ClusterID,
		SharedTier: *header.DeployType == TierShared,
		NextGen:    *header.DeployType == TierNextGen || *header.DeployType == TierPremium,
		Byoc:       *header.BizType == "byoc",
		TableName:  STATEMENTS_TABLE_NAME,
		StartTime:  "",
		EndTime:    "",
		Fields:     fields,
		GroupBy:    statementsPlanDigestGroupByClause(),
		Filter:     filter,
		Sort:       "",
		Limit:      100000,
		Offset:     0,
		DataSource: dataSource,
	}

	// Call ListTableData to get the data
	res, err := s.SQLsDuckDBAdapter.ListTableData(c.Request.Context(), params)
	if err != nil {
		log.Warn("failed to query statements plans", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "Internal Server Error",
			FullText: err.Error(),
		})
		return
	}

	formatStatementsListRowsForAPI(res.Data, dataSource)
	c.JSON(http.StatusOK, res.Data)
}

type StatementsPlanDetailQuery struct {
	BeginTime   *int     `form:"begin_time"`
	EndTime     *int     `form:"end_time"`
	SchemaName  string   `form:"schema_name"`
	Digest      string   `form:"digest"`
	Plans       []string `form:"plans"`
	DataSource  string   `form:"dataSource"`
	DataSource2 string   `form:"data_source"`
}

// @Summary		Get statement plan detail
// @Tags			statements
// @Description	Returns aggregated detail for a SQL digest, optionally narrowed to specific plan_digest values.
// @Produce		json
// @Param			X-Provider		header		string		true	"Cloud provider"
// @Param			X-Region		header		string		true	"Cloud region"
// @Param			X-Org-Id		header		string		true	"Organization ID"
// @Param			X-Project-Id	header		string		false	"Project ID"
// @Param			X-Cluster-Id	header		string		true	"Cluster ID"
// @Param			X-Deploy-Type	header		string		true	"Deployment tier"
// @Param			begin_time		query		int			true	"Start time in Unix seconds (inclusive)"
// @Param			end_time		query		int			true	"End time in Unix seconds (inclusive)"
// @Param			schema_name		query		string		false	"Filter by schema name (substring match)"
// @Param			digest			query		string		true	"SQL digest"
// @Param			plans			query		[]string	false	"Filter by plan_digest values (supports plans[])"
// @Param			dataSource		query		string		false	"Data source identifier"
// @Param			data_source		query		string		false	"Data source identifier (alias)"
// @Success		200	{object}	object	"aggregated statement detail row"
// @Router			/statements/plan/detail [get]
func (s *Service) StatementsPlanDetailHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind statements plan detail header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	if err := header.Validate(); err != nil {
		log.Warn("invalid statements plan detail header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}

	var query StatementsPlanDetailQuery
	if err := c.BindQuery(&query); err != nil {
		log.Warn("failed to bind statements plan detail query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	bindStatementsPlans(c, &query.Plans)

	dataSource := normalizeDataSource(query.DataSource, query.DataSource2)

	if *header.DeployType == "byoc" {
		*header.BizType = "byoc"
		*header.DeployType = "nextgen"
	}

	fieldInfos, err := s.SQLsDuckDBAdapter.ListTableFields(c.Request.Context(), &ListTableFieldsParams{
		Env:        *header.Env,
		Provider:   *header.Provider,
		Region:     *header.Region,
		TenantID:   *header.TenantID,
		ProjectID:  *header.ProjectID,
		ClusterID:  *header.ClusterID,
		SharedTier: *header.DeployType == TierShared,
		NextGen:    *header.DeployType == TierNextGen || *header.DeployType == TierPremium,
		Byoc:       *header.BizType == "byoc",
		TableName:  STATEMENTS_TABLE_NAME,
		DataSource: dataSource,
	})
	if err != nil {
		log.Warn("failed to list statements table fields for plan detail", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "Internal Server Error",
			FullText: err.Error(),
		})
		return
	}
	fields := filterStatementsSupportedFieldNames(dataSource, columnNameSet(fieldInfos))

	filter := fmt.Sprintf("digest = '%s' and %s", query.Digest, buildStatementsSummaryBaseFilter(*query.BeginTime, *query.EndTime, nil, nil, "", dataSource))

	plans := normalizeStatementsFilterValues(query.Plans)
	filter = appendStatementsPlanDigestFilter(filter, plans)
	filter = appendStatementsSchemaNameQueryFilter(filter, query.SchemaName, dataSource)

	params := &ListTableDataParams{
		Env:        *header.Env,
		Provider:   *header.Provider,
		Region:     *header.Region,
		TenantID:   *header.TenantID,
		ProjectID:  *header.ProjectID,
		ClusterID:  *header.ClusterID,
		SharedTier: *header.DeployType == TierShared,
		NextGen:    *header.DeployType == TierNextGen || *header.DeployType == TierPremium,
		Byoc:       *header.BizType == "byoc",
		TableName:  STATEMENTS_TABLE_NAME,
		StartTime:  "",
		EndTime:    "",
		Fields:     fields,
		GroupBy:    "digest",
		Filter:     filter,
		Sort:       "",
		Limit:      100000,
		Offset:     0,
		DataSource: dataSource,
	}

	// Call ListTableData to get the data
	res, err := s.SQLsDuckDBAdapter.ListTableData(c.Request.Context(), params)
	if err != nil {
		log.Warn("failed to query statements list", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "Internal Server Error",
			FullText: err.Error(),
		})
		return
	}

	if len(res.Data) == 0 {
		c.JSON(http.StatusOK, gin.H{})
		return
	}

	formatStatementsListRowsForAPI(res.Data, dataSource)
	c.JSON(http.StatusOK, res.Data[0])
}