package sqls

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

	"github.com/gin-gonic/gin"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

// StatementsSummaryEnvelope matches TopRU summary / user_stats response shape.
type StatementsSummaryEnvelope struct {
	Status string                  `json:"status"`
	Data   []StatementsSummaryItem `json:"data"`
}

// StatementsSummaryQuery binds GET /statements/summary query parameters.
type StatementsSummaryQuery 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"`
	Top         string   `form:"top"`
	OrderBy     string   `form:"order_by"`  // sortable available-field metric (default sum_ru)
	Step        int      `form:"step"`      // bucket seconds; v1 default 60, v2 default/min 300 and multiple of 300
	Dimension   string   `form:"dimension"` // user | sqltypes | digest | db | table (default digest)
	DataSource  string   `form:"dataSource"`
	DataSource2 string   `form:"data_source"`
}

// StatementsSummaryItem follows TopRU summary row shape.
type StatementsSummaryItem struct {
	Dimension         string  `json:"dimension,omitempty"`
	SQLDigest         string  `json:"sql_digest"`
	PlanDigest        string  `json:"plan_digest"`
	SQLText           string  `json:"sql_text,omitempty"`
	User              string  `json:"user,omitempty"`
	StmtType          string  `json:"stmt_type,omitempty"`
	DB                string  `json:"db,omitempty"`
	Table             string  `json:"table,omitempty"`
	TotalRU           float64 `json:"total_ru"`
	ExecCount         int64   `json:"exec_count"`
	DurationNs        int64   `json:"duration_ns"`
	ExecCountPerSec   float64 `json:"exec_count_per_sec"`
	DurationPerExecMs float64 `json:"duration_per_exec_ms"`
	// OrderMetric is the digest-level value used for Top N ranking (same as top_digests.order_metric).
	OrderMetric  float64 `json:"order_metric"`
	IsOther      bool    `json:"is_other"`
	TimestampSec []int64 `json:"timestamp_sec"`
	// ValueByTime is one value per time_bucket, aggregated with the same expression as order_by (e.g. sum_ru, exec_count, sum_latency).
	ValueByTime []float64 `json:"value_by_time,omitempty"`
}

// StatementsTopStatsEnvelope matches TopRU list-style envelope.
type StatementsTopStatsEnvelope struct {
	Status string                   `json:"status"`
	Data   []StatementsTopStatsItem `json:"data"`
}

// StatementsTopStatsQuery binds GET /statements/topstats query parameters.
type StatementsTopStatsQuery 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"`
	Top         string   `form:"top"`
	Dimension   string   `form:"dimension"` // user | sqltypes | digest | db | table
	OrderBy     string   `form:"order_by"`  // sortable available-field metric
	DataSource  string   `form:"dataSource"`
	DataSource2 string   `form:"data_source"`
}

// StatementsTopStatsItem is one rank item for topstats.
type StatementsTopStatsItem struct {
	// Compatibility field; prefer dimension-specific fields below.
	Key string `json:"key,omitempty"`
	// Compatibility field; total_ru is only populated when order_by is sum_ru or sum_ru_v2.
	TotalRU float64 `json:"total_ru,omitempty"`

	Dimension  string  `json:"dimension"`
	OrderBy    string  `json:"order_by"`
	Value      float64 `json:"value"`
	ExecCount  int64   `json:"exec_count"`
	SumLatency int64   `json:"sum_latency"`

	User      string `json:"user,omitempty"`
	StmtType  string `json:"stmt_type,omitempty"`
	DB        string `json:"db,omitempty"`
	Table     string `json:"table,omitempty"`
	SQLDigest string `json:"sql_digest,omitempty"`

	DigestText      string `json:"digest_text,omitempty"`
	QuerySampleText string `json:"query_sample_text,omitempty"`
	PlanCount       int64  `json:"plan_count,omitempty"`
}

func escapeSQLLiteral(s string) string {
	return strings.ReplaceAll(s, "'", "''")
}

// queryMultiValue reads repeated query parameters for any of the given keys.
// Supports frontends that send array params as schemas[] / stmt_types[] instead of schemas.
func queryMultiValue(c *gin.Context, keys ...string) []string {
	if c == nil || c.Request == nil || c.Request.URL == nil {
		return nil
	}
	q := c.Request.URL.Query()
	seen := make(map[string]struct{})
	out := make([]string, 0)
	for _, key := range keys {
		for _, v := range q[key] {
			v = strings.TrimSpace(v)
			if v == "" {
				continue
			}
			if _, ok := seen[v]; ok {
				continue
			}
			seen[v] = struct{}{}
			out = append(out, v)
		}
	}
	return out
}

func mergeQueryStringSlices(slices ...[]string) []string {
	seen := make(map[string]struct{})
	out := make([]string, 0)
	for _, slice := range slices {
		for _, v := range slice {
			v = strings.TrimSpace(v)
			if v == "" {
				continue
			}
			if _, ok := seen[v]; ok {
				continue
			}
			seen[v] = struct{}{}
			out = append(out, v)
		}
	}
	return out
}

// bindSchemasAndStmtTypes merges BindQuery slices with schemas/schemas[] and stmt_types/stmt_types[].
func bindSchemasAndStmtTypes(c *gin.Context, schemas, stmtTypes *[]string) {
	*schemas = mergeQueryStringSlices(*schemas, queryMultiValue(c, "schemas", "schemas[]"))
	*stmtTypes = mergeQueryStringSlices(*stmtTypes, queryMultiValue(c, "stmt_types", "stmt_types[]"))
}

func buildStatementsSummaryBaseFilter(beginTime, endTime int, schemas, stmtTypes []string, text, dataSource string) string {
	filter := ""
	if isSQLsDataSourceV2(dataSource) {
		filter = fmt.Sprintf("\"begin\" <= %d and \"end\" >= %d", endTime, beginTime)
	} else {
		filter = fmt.Sprintf("SUMMARY_BEGIN_TIME <= to_timestamp('%d') and SUMMARY_END_TIME >= to_timestamp('%d')", endTime, beginTime)
	}
	if text != "" {
		escaped := escapeSQLLiteral(text)
		if isSQLsDataSourceV2(dataSource) {
			filter += fmt.Sprintf(" and (sample_sql like '%%%s%%' or digest like '%%%s%%')", escaped, escaped)
		} else {
			filter += fmt.Sprintf(" and (QUERY_SAMPLE_TEXT like '%%%s%%' or digest like '%%%s%%')", escaped, escaped)
		}
	}
	schemas = normalizeStatementsFilterValues(schemas)
	if len(schemas) > 0 {
		schemaCol := statementsDBNameColumn(dataSource)
		filter = appendStatementsCoalescedInFilter(filter, schemaCol, schemas)
	}
	stmtTypes = normalizeStatementsFilterValues(stmtTypes)
	if len(stmtTypes) > 0 {
		filter = appendStatementsCoalescedInFilter(filter, "stmt_type", stmtTypes)
	}
	return filter
}

// appendStatementsSampleUserFilter restricts rows to the given user.
// v2: match only the first auth_users key; v1: exact match on sample_user column.
func appendStatementsSampleUserFilter(filter, sampleUser, dataSource string) string {
	sampleUser = strings.TrimSpace(sampleUser)
	if sampleUser == "" {
		return filter
	}
	if isStatementsNAValue(sampleUser) {
		if isSQLsDataSourceV2(dataSource) {
			return filter + fmt.Sprintf(" and coalesce(%s, '') = ''", statementsV2FirstUserKeyColumn())
		}
		userCol := statementsSampleUserColumn(dataSource)
		return filter + fmt.Sprintf(" and coalesce(%s, '') = ''", userCol)
	}
	escaped := escapeSQLLiteral(sampleUser)
	if isSQLsDataSourceV2(dataSource) {
		return filter + fmt.Sprintf(" and %s = '%s'", statementsV2FirstUserKeyColumn(), escaped)
	}
	userCol := statementsSampleUserColumn(dataSource)
	return filter + fmt.Sprintf(" and %s = '%s'", userCol, escaped)
}

func parsePositiveInt(s string, def int) (int, error) {
	if s == "" {
		return def, nil
	}
	v, err := strconv.Atoi(s)
	if err != nil {
		return 0, fmt.Errorf("invalid integer: %w", err)
	}
	if v <= 0 {
		return 0, fmt.Errorf("value must be positive")
	}
	return v, nil
}

var statementsSortableOrderByFields = map[string]struct{}{
	"exec_count":   {},
	"sum_latency":  {},
	"max_latency":  {},
	"min_latency":  {},
	"avg_latency":  {},
	"sum_errors":   {},
	"sum_warnings": {},
	"plan_count":   {},
	"sum_ru":       {},
	"sum_ru_v2":    {},
	"sum_rru":      {},
	"sum_wru":      {},
	"avg_ru":       {},
	"avg_ru_v2":    {},
	"avg_rru":      {},
	"avg_wru":      {},
	"max_ru":       {},
	"max_ru_v2":    {},
	"max_rru":      {},
	"max_wru":      {},
}

func statementsSummaryOrderColumn(orderBy string) (string, error) {
	orderBy = strings.TrimSpace(strings.ToLower(orderBy))
	if orderBy == "" {
		return "sum_ru", nil
	}
	if orderBy == "total_ru" {
		return "sum_ru", nil
	}
	if _, ok := statementsSortableOrderByFields[orderBy]; ok {
		return orderBy, nil
	}
	return "", fmt.Errorf("invalid order_by: %s (allowed sortable fields include: sum_ru, sum_ru_v2, exec_count, sum_latency, avg_latency, max_latency, min_latency, ...)", orderBy)
}

func parseTopStatsDimension(d string) (string, error) {
	d = strings.ToLower(strings.TrimSpace(d))
	if d == "" {
		return "digest", nil
	}
	switch d {
	case "user", "sqltypes", "digest", "db", "table":
		return d, nil
	default:
		return "", fmt.Errorf("invalid dimension: %s (allowed: user, sqltypes, digest, db, table)", d)
	}
}

func normalizeDataSource(v1, v2 string) string {
	if strings.TrimSpace(v1) != "" {
		return strings.TrimSpace(v1)
	}
	return strings.TrimSpace(v2)
}

// statementsV2SummaryStepSeconds is the native aggregation window of statements v2
// (~5 minutes). Summary bucketing for v2 must align to this grid.
const statementsV2SummaryStepSeconds = 300

func alignUnixFloor(ts, step int) int {
	if step <= 0 {
		return ts
	}
	if ts >= 0 {
		return ts - ts%step
	}
	// Negative timestamps are unlikely for wall-clock unix, but keep floor semantics.
	r := ts % step
	if r == 0 {
		return ts
	}
	return ts - (step + r)
}

func alignUnixCeil(ts, step int) int {
	if step <= 0 {
		return ts
	}
	floor := alignUnixFloor(ts, step)
	if floor == ts {
		return ts
	}
	return floor + step
}

// normalizeStatementsV2SummaryWindow enforces v2 summary query stability:
// step defaults to 300; invalid step values are coerced to 300; begin/end are aligned
// down/up to 300-second boundaries so identical wall ranges produce identical queries.
func normalizeStatementsV2SummaryWindow(begin, end, step int) (int, int, int, error) {
	if step <= 0 || step < statementsV2SummaryStepSeconds || step%statementsV2SummaryStepSeconds != 0 {
		step = statementsV2SummaryStepSeconds
	}
	begin = alignUnixFloor(begin, statementsV2SummaryStepSeconds)
	end = alignUnixCeil(end, statementsV2SummaryStepSeconds)
	if end <= begin {
		return 0, 0, 0, fmt.Errorf(
			"end_time must be greater than begin_time after aligning to %d seconds",
			statementsV2SummaryStepSeconds,
		)
	}
	return begin, end, step, nil
}

// @Summary		Statements summary statistics
// @Tags			statements
// @Description	Returns top-N summary with time-bucketed series (TopRU-like response shape). Group by dimension (default digest). For data_source=v2, step defaults to 300 (invalid values coerced to 300); begin_time/end_time are aligned to 300s.
// @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); v2: floored to 300s"
// @Param			end_time		query		int			true	"End time in Unix seconds (inclusive); v2: ceiled to 300s"
// @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: auth_users key; v1: sample_user); 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, e.g. [{\"field\":\"digest\",\"values\":[\"abc\"]}]"
// @Param			top				query		string		false	"Top N for each ranking list (default 20)"
// @Param			order_by		query		string		false	"Sort metric from available fields (numeric subset), e.g. sum_ru|sum_ru_v2|exec_count|sum_latency"
// @Param			dimension		query		string		false	"Group dimension: user | sqltypes | digest | db | table (default: digest)"
// @Param			step			query		int			false	"Time bucket size in seconds (v1 default 60; v2 default 300, invalid values coerced to 300)"
// @Success		200	{object}	StatementsSummaryEnvelope
// @Router			/statements/summary [get]
func (s *Service) StatementsSummaryHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind statements summary 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 summary header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}

	var q StatementsSummaryQuery
	if err := c.BindQuery(&q); err != nil {
		log.Warn("failed to bind statements summary query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: err.Error(),
		})
		return
	}
	bindSchemasAndStmtTypes(c, &q.Schemas, &q.StmtTypes)
	bindStatementsUsers(c, &q.SampleUsers)
	bindStatementsTables(c, &q.Tables)
	bindStatementsValues(c, &q.Values)
	if q.BeginTime == nil || q.EndTime == nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "Bad Request",
			FullText: "begin_time and end_time are required",
		})
		return
	}
	extraFilters, err := parseStatementsFiltersJSON(q.Filters)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}

	beginTime := *q.BeginTime
	endTime := *q.EndTime
	step := q.Step
	dataSource := normalizeDataSource(q.DataSource, q.DataSource2)
	if isSQLsDataSourceV2(dataSource) {
		beginTime, endTime, step, err = normalizeStatementsV2SummaryWindow(beginTime, endTime, step)
		if err != nil {
			c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
			return
		}
	} else if step <= 0 {
		step = 60
	}

	filterOpts := statementsFilterOptions{
		beginTime:    beginTime,
		endTime:      endTime,
		schemas:      q.Schemas,
		stmtTypes:    q.StmtTypes,
		text:         q.Text,
		sampleUser:   q.SampleUser,
		sampleUsers:  q.SampleUsers,
		tables:       q.Tables,
		field:        q.Field,
		values:       q.Values,
		extraFilters: extraFilters,
		dataSource:   dataSource,
	}
	if err := validateStatementsQueryFilters(filterOpts); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}

	topN, err := parsePositiveInt(q.Top, 20)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}
	orderCol, err := statementsSummaryOrderColumn(q.OrderBy)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}
	dimension, err := parseTopStatsDimension(q.Dimension)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}

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

	filter := buildStatementsFilter(filterOpts)
	params := &StatementsSummaryParams{
		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",
		Filter:     filter,
		StartUnix:  int64(beginTime),
		EndUnix:    int64(endTime),
		Top:        topN,
		OrderBy:    orderCol,
		Step:       step,
		Dimension:  dimension,
		DataSource: dataSource,
	}

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

	c.JSON(http.StatusOK, StatementsSummaryEnvelope{Status: "ok", Data: data})
}

// @Summary		Statements top statistics
// @Tags			statements
// @Description	Returns ranking list grouped by one dimension: user, sqltypes, digest, db (database/schema name), or table.
// @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			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: auth_users key; v1: sample_user); 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, e.g. [{\"field\":\"digest\",\"values\":[\"abc\"]}]"
// @Param			top				query		string		false	"Top N rows (default 20)"
// @Param			dimension		query		string		false	"Group dimension: user | sqltypes | digest | db | table (default: digest)"
// @Param			order_by		query		string		false	"Sort metric from available fields (numeric subset), e.g. sum_ru|sum_ru_v2|exec_count|sum_latency"
// @Success		200	{object}	StatementsTopStatsEnvelope
// @Router			/statements/topstats [get]
func (s *Service) StatementsTopStatsHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind statements topstats 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 topstats header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}

	var q StatementsTopStatsQuery
	if err := c.BindQuery(&q); err != nil {
		log.Warn("failed to bind statements topstats query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}
	bindSchemasAndStmtTypes(c, &q.Schemas, &q.StmtTypes)
	bindStatementsUsers(c, &q.SampleUsers)
	bindStatementsTables(c, &q.Tables)
	bindStatementsValues(c, &q.Values)
	if q.BeginTime == nil || q.EndTime == nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: "begin_time and end_time are required"})
		return
	}
	extraFilters, err := parseStatementsFiltersJSON(q.Filters)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}
	filterOpts := statementsFilterOptions{
		beginTime:    *q.BeginTime,
		endTime:      *q.EndTime,
		schemas:      q.Schemas,
		stmtTypes:    q.StmtTypes,
		text:         q.Text,
		sampleUser:   q.SampleUser,
		sampleUsers:  q.SampleUsers,
		tables:       q.Tables,
		field:        q.Field,
		values:       q.Values,
		extraFilters: extraFilters,
	}
	if err := validateStatementsQueryFilters(filterOpts); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}

	topN, err := parsePositiveInt(q.Top, 20)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}
	dimension, err := parseTopStatsDimension(q.Dimension)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}
	orderCol, err := statementsSummaryOrderColumn(q.OrderBy)
	if err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}

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

	dataSource := normalizeDataSource(q.DataSource, q.DataSource2)
	filterOpts.dataSource = dataSource
	filter := buildStatementsFilter(filterOpts)
	params := &StatementsTopStatsParams{
		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",
		Filter:     filter,
		Top:        topN,
		Dimension:  dimension,
		OrderBy:    orderCol,
		DataSource: dataSource,
	}

	data, err := s.SQLsDuckDBAdapter.QueryStatementsTopStats(c.Request.Context(), params)
	if err != nil {
		log.Warn("failed to query statements topstats", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{Error: true, Message: "Internal Server Error", FullText: err.Error()})
		return
	}
	c.JSON(http.StatusOK, StatementsTopStatsEnvelope{Status: "ok", Data: data})
}