package slowquery

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"time"
	"unicode"

	"github.com/pingcap/log"
	o11yv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/slowquery/slowstats"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/utils/duckdb"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/utils/metacache"
	"go.uber.org/zap"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type DuckDBAdapter struct {
	vendor    string
	region    string
	mataCache metacache.MetaCache
	duckdb    duckdb.QueryDuckDBFetcherInterface
	kubeCli   client.Client
}

func NewSlowQueryDuckDBAdapter(vendor, region string, store storage.Storage, kubeCli client.Client) *DuckDBAdapter {
	adapter := &DuckDBAdapter{
		vendor:    vendor,
		region:    region,
		mataCache: metacache.NewMetaCache(kubeCli, store),
		kubeCli:   kubeCli,
	}

	newDuckDB, err := duckdb.NewQueryDuckDBFetcherWithPool(context.Background(), vendor, region)
	if err != nil {
		log.Error("Failed to initialize DuckDB", zap.Error(err))
		panic(err)
	}
	adapter.duckdb = newDuckDB

	// Start the warmup process in a goroutine
	// no need to warmup for pooled connection
	// go func() {
	// 	err := adapter.warmup(context.Background(), newDuckDB)
	// 	if err != nil {
	// 		log.Error("Failed to warmup DuckDB", zap.Error(err))
	// 	}
	// 	adapter.startWarmupRefreshLoop(context.Background())
	// }()
	return adapter
}

// List lists slow logs which match the conditions described in ListParams.
func (s *DuckDBAdapter) List(ctx context.Context, params *ListParams) ([]*SlowLog, error) {
	log.Info("SlowQueryDuckDBAdapter List called",
		zap.String("provider", params.Provider),
		zap.String("region", params.Region),
		zap.String("tenant", params.TenantID),
		zap.String("project", params.ProjectID),
		zap.String("cluster", params.ClusterID),
		zap.Bool("sharedTier", params.SharedTier),
		zap.Int("beginTime", params.BeginTime),
		zap.Int("endTime", params.EndTime),
		zap.Int("limit", params.Limit),
		zap.String("orderBy", params.OrderBy),
		zap.String("text", params.Text),
		zap.String("digest", params.Digest),
		zap.Strings("db", params.DB),
		zap.Strings("plans", params.Plans),
		zap.Bool("desc", params.IsDesc),
		zap.Bool("showInternal", params.ShowInternal),
		zap.String("fields", params.Fields),
	)

	metaInfo, err := s.mataCache.GetMetaInfo(ctx, params.Env, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.NextGen, params.Byoc)
	if err != nil {
		log.Error("GetMetaInfo failed", zap.Error(err))
		return nil, err
	}
	if metaInfo == nil {
		return nil, fmt.Errorf("meta info not found for tenant %s and cluster %s", params.TenantID, params.ClusterID)
	}
	slowlogTable := s.getDuckDBSlowLogBaseURL(params.TenantID, metaInfo.O11yID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen)
	fullQuery, alias, err := buildListSQL(params, slowlogTable)
	if err != nil {
		return nil, err
	}

	log.Debug("DuckDB List Query", zap.String("query", fullQuery))

	var dlResults []*SlowlogInDL
	err = s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &dlResults, alias, slowlogTable, fullQuery)
	if err != nil {
		log.Error("Failed to select slowlog from DuckDB", zap.Error(err),
			zap.String("query", fullQuery),
			zap.String("clusterID", params.ClusterID))
		return nil, fmt.Errorf("failed to select slowlog from DuckDB: %w", err)
	}

	slowLogs := make([]*SlowLog, len(dlResults))
	for i, dl := range dlResults {
		slowLogs[i] = mapSlowlogInDLToSlowLog(dl)
	}

	log.Info("SlowQueryDuckDBAdapter List finished", zap.Int("count", len(slowLogs)))
	return slowLogs, nil
}

// Detail looks for a single slow log specified in DetailParams.
func (s *DuckDBAdapter) Detail(ctx context.Context, params *DetailParams) (*SlowLog, error) {
	log.Info("SlowQueryDuckDBAdapter Detail called.",
		zap.String("provider", params.Provider),
		zap.String("region", params.Region),
		zap.String("tenant", params.TenantID),
		zap.String("project", params.ProjectID),
		zap.String("cluster", params.ClusterID),
		zap.Bool("sharedTier", params.SharedTier),
		zap.Float64("timestamp", params.Timestamp),
		zap.Uint64("connID", params.ConnID),
		zap.String("digest", params.Digest),
	)

	metaInfo, err := s.mataCache.GetMetaInfo(ctx, params.Env, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.NextGen, params.Byoc)
	if metaInfo == nil {
		return nil, fmt.Errorf("meta info not found for tenant %s and cluster %s", params.TenantID, params.ClusterID)
	}
	if err != nil {
		log.Error("GetMetaInfo failed", zap.Error(err))
		return nil, err
	}

	slowlogTable := s.getDuckDBSlowLogBaseURL(params.TenantID, metaInfo.O11yID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen)
	alias := tableAlias(params.ClusterID)

	columnNames, err := s.getSlowlogTableColumns(ctx, slowlogTable, alias)
	if err != nil {
		return nil, err
	}

	fullQuery, alias, err := buildDetailSQL(params, slowlogTable, slowlogColumnNameSet(columnNames))
	if err != nil {
		return nil, err
	}

	log.Debug("DuckDB Detail Query (Delta Lake, joined)", zap.String("query", fullQuery))

	var dlResults []*SlowlogInDL
	err = s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &dlResults, alias, slowlogTable, fullQuery)
	if err != nil {
		log.Error("Failed to select slowlog from DuckDB", zap.Error(err),
			zap.String("query", fullQuery),
			zap.String("clusterID", params.ClusterID))
		return nil, fmt.Errorf("failed to select slowlog detail from DuckDB Delta Lake (joined): %w", err)
	}

	if len(dlResults) == 0 {
		return nil, errors.New("slow log for detail not found")
	}
	// In case multiple results match (highly unlikely for ConnID + Timestamp + Digest),
	// we return the first one.
	slowLog := mapSlowlogInDLToSlowLog(dlResults[0])

	log.Info("SlowQueryDuckDBAdapter Detail finished (Delta Lake, joined)")
	return slowLog, nil
}

//nolint:unused // Kept for potential future use
func (s *DuckDBAdapter) startWarmupRefreshLoop(ctx context.Context) {
	refreshInterval := 2 * time.Hour
	ticker := time.NewTicker(refreshInterval)
	defer ticker.Stop()

	for range ticker.C {
		log.Debug("Refreshing DuckDB")
		err := s.RenewAndWarmupDuckDB(ctx)
		if err != nil {
			log.Error("Failed to re-initialize DuckDB", zap.Error(err))
			continue
		}
		log.Info("DuckDB re-initialized")
	}
}

func (s *DuckDBAdapter) RenewAndWarmupDuckDB(ctx context.Context) error {
	log.Debug("Initializing DuckDB for slow query adapter")
	defer log.Debug("DuckDB initialization completed")

	newDuckDB, err := duckdb.NewQueryDuckDBFetcherWithPool(context.Background(), s.vendor, s.region)
	if err != nil {
		log.Error("Failed to reinitialize DuckDB", zap.Error(err))
		return err
	}

	err = s.warmup(ctx, newDuckDB)
	if err != nil {
		log.Error("Failed to warmup DuckDB", zap.Error(err))
		return err
	}

	originalDuckDB := s.duckdb
	s.duckdb = newDuckDB
	if originalDuckDB != nil {
		go func() {
			time.Sleep(time.Minute * 5)
			if err := originalDuckDB.Close(context.Background()); err != nil {
				log.Error("Failed to close original DuckDB", zap.Error(err))
			} else {
				log.Info("Original DuckDB closed successfully")
			}
		}()
	}
	return nil
}

const warmupQuery = "SELECT count(*) as count FROM %s WHERE date >= '%s';"

func (s *DuckDBAdapter) warmup(ctx context.Context, db duckdb.QueryDuckDBFetcherInterface) error {
	log.Debug("Starting warmup for DuckDB slow query adapter")
	defer log.Debug("DuckDB warmup completed")

	date := time.Now().Format("2006-01-02")

	clusterTables, err := s.getClusterTables(ctx)
	if err != nil {
		log.Error("Failed to get cluster tables", zap.Error(err))
	}

	type warmupResult struct {
		Count int64 `db:"count"`
	}
	// Execute the query using a custom query function that handles the custom result type
	var warmupResults []warmupResult

	for alias, table := range clusterTables {
		err := db.SelectSlowlogInDLWithoutFilterPushdown(ctx, &warmupResults, alias, table, fmt.Sprintf(warmupQuery, alias, date))
		if err != nil {
			log.Error("warmup failed", zap.String("tableAlias", alias), zap.Error(err))
			continue
		}
	}

	// Clean up cache files older than 30 days
	cacheDir := "/cache"
	if _, err := os.Stat(cacheDir); !os.IsNotExist(err) {
		cmd := exec.Command("find", cacheDir, "-type", "f", "-mtime", "+30", "-delete")
		output, err := cmd.CombinedOutput()
		if err != nil {
			log.Error("Failed to clean up old cache files",
				zap.String("cacheDir", cacheDir),
				zap.String("output", string(output)),
				zap.Error(err))
		} else {
			log.Info("Successfully cleaned up old cache files",
				zap.String("cacheDir", cacheDir),
				zap.String("output", string(output)))
		}
	}

	return nil
}

func tableAlias(clusterID string) string {
	tableName := strings.ReplaceAll(clusterID, "-", "_")
	return fmt.Sprintf("slowlog_%s", tableName)
}

// buildListSQL constructs the SQL query for the List operation.
func buildListSQL(params *ListParams, slowlogTablePath string) (string, string, error) {
	//fromClause := fmt.Sprintf("FROM delta_scan('%s')", slowlogTablePath)
	// alias := tableAlias(params.ClusterID)
	fromClause := fmt.Sprintf("FROM delta_scan('%s')", slowlogTablePath)

	var conditions []string
	beginTimeFloat := float64(params.BeginTime)
	endTimeFloat := float64(params.EndTime)
	conditions = append(conditions, fmt.Sprintf("time >= %.6f AND time <= %.6f", beginTimeFloat, endTimeFloat))

	beginDate := time.Unix(int64(params.BeginTime), 0).UTC().Format("20060102")
	endDate := time.Unix(int64(params.EndTime), 0).UTC().Format("20060102")
	conditions = append(conditions, fmt.Sprintf("date >= '%s' AND date <= '%s'", beginDate, endDate))

	if params.Text != "" {
		textLower := strings.ToLower(params.Text)
		conditions = append(conditions, fmt.Sprintf(
			"(lower(query) LIKE '%%%s%%' OR lower(digest) LIKE '%%%s%%')",
			textLower, textLower,
		))
	}

	// Digest filter
	if params.Digest != "" {
		conditions = append(conditions, fmt.Sprintf("digest = '%s'", strings.ToLower(params.Digest)))
	}

	// DB filter
	if len(params.DB) > 0 {
		dbs := make([]string, len(params.DB))
		for i, db := range params.DB {
			dbs[i] = fmt.Sprintf("'%s'", db)
		}
		conditions = append(conditions, fmt.Sprintf("db IN (%s)", strings.Join(dbs, ", ")))
	}

	// Plans (PlanDigest) filter
	if len(params.Plans) > 0 {
		plans := make([]string, len(params.Plans))
		for i, plan := range params.Plans {
			plans[i] = fmt.Sprintf("'%s'", plan)
		}
		// Assume 'plan_digest' field is directly in the slowlog table
		conditions = append(conditions, fmt.Sprintf("plan_digest IN (%s)", strings.Join(plans, ", ")))
	}

	// ShowInternal logic
	if !params.ShowInternal {
		// Assume 'is_internal' field is directly in the slowlog table
		conditions = append(conditions, "(is_internal = false OR is_internal IS NULL)")
	}

	whereClause := ""
	if len(conditions) > 0 {
		whereClause = "WHERE " + strings.Join(conditions, " AND ")
	}

	// Order By clause - field names remain the same but without 'sl.' prefix
	var orderByField string
	switch strings.ToLower(params.OrderBy) {
	case "", "timestamp", "time":
		orderByField = "time"
	default:
		orderByField = fieldAdapt(params.OrderBy)
	}

	if castType, ok := FieldCastTypeMap[orderByField]; ok {
		orderByField = fmt.Sprintf("CAST(%s AS %s)", orderByField, castType)
	}

	orderDirection := "ASC"
	if params.IsDesc {
		orderDirection = "DESC"
	}
	orderByClause := fmt.Sprintf("ORDER BY %s %s", orderByField, orderDirection)

	limit := params.Limit
	if limit == 0 {
		limit = defaultListLimit
	}
	limitClause := fmt.Sprintf("LIMIT %d", limit)

	listFields := convertDeltalakeFields(params.Fields)

	fullQuery := fmt.Sprintf("SELECT %s %s %s %s %s;", listFields, fromClause, whereClause, orderByClause, limitClause)
	return fullQuery, "", nil
}

var fieldsMap = map[string]string{
	"success":            "succ",
	"lock_keys_time":     "lockkeys_time",
	"plan":               "plan",
	"plan_from_biniding": "plan_from_binding",
	"prewrite_region":    "prewrite_region",
	"timestamp":          "time",
	"connection_id":      "conn_id",
	"memory_max":         "mem_max",
	"ru":                 "ru",
	"ru_v2":              "ru_v2",
}

var toRemoveFieldMap = map[string]bool{
	"connection_id": true,
	"digest":        true,
	"timestamp":     true,
	"query_time":    true,
	"memory_max":    true,
}

func convertDeltalakeFields(fields string) string {
	fieldList := strings.Split(fields, ",")
	selectFields := []string{
		"id",
		"conn_id",
		"digest",
		"time",
		"query_time",
		"mem_max",
	}

	for _, field := range fieldList {
		if _, ok := toRemoveFieldMap[field]; ok {
			continue
		}
		if field == "ru" {
			selectFields = append(selectFields, "(COALESCE(request_unit_read, 0) + COALESCE(request_unit_write, 0)) AS ru")
		} else if field == "ru_v2" {
			selectFields = append(selectFields, "(COALESCE(request_unit_v2, 0)) AS ru_v2")
		} else {
			selectFields = append(selectFields, fieldAdapt(field))
		}
	}
	return strings.TrimSuffix(strings.Join(selectFields, ", "), ", ")
}

func fieldAdapt(field string) string {
	if mappedField, ok := fieldsMap[field]; ok {
		return mappedField
	} else {
		return field
	}
}

// buildDetailSQL constructs the SQL query for the Detail operation.
func buildDetailSQL(params *DetailParams, slowlogTablePath string, columns map[string]struct{}) (string, string, error) {
	//fromClause := fmt.Sprintf("FROM delta_scan('%s')", slowlogTablePath)
	//alias := tableAlias(params.ClusterID)
	fromClause := fmt.Sprintf("FROM delta_scan('%s')", slowlogTablePath)

	timeLowerBound := params.Timestamp - 0.001
	timeUpperBound := params.Timestamp + 0.001

	targetDate := time.Unix(int64(params.Timestamp), 0).UTC().Format("20060102")

	var whereClause string
	if params.ConnID == 0 {
		whereClause = fmt.Sprintf("WHERE time >= %.6f AND time <= %.6f AND digest = '%s' AND date = '%s'",
			timeLowerBound, timeUpperBound, params.Digest, targetDate)
	} else {
		whereClause = fmt.Sprintf("WHERE time >= %.6f AND time <= %.6f AND conn_id = '%d' AND digest = '%s' AND date = '%s'",
			timeLowerBound, timeUpperBound, params.ConnID, params.Digest, targetDate)
	}

	selectColumns := buildDetailSelectColumnsFromSchema(columns)
	if selectColumns == "" {
		return "", "", errors.New("no supported slowlog detail columns found in table schema")
	}

	fullQuery := fmt.Sprintf("SELECT %s %s %s LIMIT 1;", selectColumns, fromClause, whereClause)
	return fullQuery, tableAlias(params.ClusterID), nil
}

// StatsMetric returns time-bucketed slow query counts.
// Implements the SlowQuery interface.
func (s *DuckDBAdapter) StatsMetric(
	ctx context.Context,
	cluster slowstats.ClusterInfo,
	beginHourTs int64,
	hours int,
	name string,
) ([][]int64, error) {
	log.Info("SlowQueryDuckDBAdapter StatsMetric called",
		zap.String("provider", cluster.Provider),
		zap.String("region", cluster.Region),
		zap.String("tenant", cluster.TenantID),
		zap.String("project", cluster.ProjectID),
		zap.String("cluster", cluster.ClusterID),
		zap.Bool("sharedTier", cluster.SharedTier),
		zap.Int64("beginTime", beginHourTs),
		zap.Int("hours", hours),
		zap.String("name", name),
	)

	metaInfo, err := s.mataCache.GetMetaInfo(ctx, cluster.Env, cluster.Region, cluster.TenantID, cluster.ProjectID, cluster.ClusterID, cluster.NextGen, cluster.Byoc)
	if err != nil {
		log.Error("GetMetaInfo failed", zap.Error(err))
		return nil, err
	}

	slowlogTable := s.getDuckDBSlowLogBaseURL(cluster.TenantID, metaInfo.O11yID, cluster.Env, cluster.Provider, cluster.Region, cluster.SharedTier, cluster.NextGen)
	// tableAlias := tableAlias(cluster.ClusterID)

	// Calculate time range and bucket size
	startTime := time.Unix(beginHourTs, 0)
	endTime := time.Unix(beginHourTs+int64(hours)*3600, 0)
	timeRange := endTime.Sub(startTime)

	// Calculate bucket size to get around 120 buckets
	totalMinutes := int(timeRange.Minutes())
	bucketMinutes := totalMinutes / 120
	if bucketMinutes < 1 {
		bucketMinutes = 1 // Minimum 1 minute bucket size
	}

	// Build the SQL query
	query := fmt.Sprintf(`
		SELECT 
			FLOOR(time / %d) * %d AS bucket_ts,
			COUNT(*) as count
		FROM delta_scan('%s') 
		WHERE time >= %d AND time <= %d AND (is_internal = false OR is_internal IS NULL)
`, bucketMinutes*60, bucketMinutes*60, slowlogTable, beginHourTs, beginHourTs+int64(hours)*3600)

	query += " GROUP BY bucket_ts ORDER BY bucket_ts ASC"

	log.Debug("DuckDB StatsMetric Query", zap.String("query", query))

	// Execute the query using SelectSlowlogInDLWithoutFilterPushdown
	type statsMetricResult struct {
		BucketTS float64 `db:"bucket_ts"`
		Count    int64   `db:"count"`
	}
	// Execute the query using a custom query function that handles the custom result type
	var dlResults []statsMetricResult
	err = s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &dlResults, "", slowlogTable, query)
	if err != nil {
		log.Error("Failed to execute stats metric query",
			zap.Error(err),
			zap.String("query", query),
			zap.String("clusterID", cluster.ClusterID))
		return nil, fmt.Errorf("failed to execute stats metric query: %w", err)
	}

	// Process results into the format [[ts1, count1], [ts2, count2], ...]
	result := make([][]int64, 0, len(dlResults))

	for _, dl := range dlResults {
		// Convert the timestamp to int64, handling any floating-point precision
		ts := int64(dl.BucketTS)
		result = append(result, []int64{ts, dl.Count})
	}

	return result, nil
}

// Stats returns aggregated statistics about slow queries.
func (s *DuckDBAdapter) Stats(
	ctx context.Context,
	cluster slowstats.ClusterInfo,
	beginHourTs int64,
	hours int,
	statementTypes []string,
	statementTypesNotIn bool,
	databases []string,
	internal bool,
	order string,
	limit int,
) ([]slowstats.SlowStats, error) {
	log.Info("SlowQueryDuckDBAdapter Stats called",
		zap.String("provider", cluster.Provider),
		zap.String("region", cluster.Region),
		zap.String("tenant", cluster.TenantID),
		zap.String("project", cluster.ProjectID),
		zap.String("cluster", cluster.ClusterID),
		zap.Bool("sharedTier", cluster.SharedTier),
		zap.Int64("beginTime", beginHourTs),
		zap.Int("hours", hours),
		zap.Int("limit", limit),
	)

	metaInfo, err := s.mataCache.GetMetaInfo(ctx, cluster.Env, cluster.Region, cluster.TenantID, cluster.ProjectID, cluster.ClusterID, cluster.NextGen, cluster.Byoc)
	if err != nil {
		log.Error("GetMetaInfo failed", zap.Error(err))
		return nil, err
	}
	if metaInfo == nil {
		err := fmt.Errorf("meta info not found for cluster %s", cluster.ClusterID)
		log.Error(err.Error())
		return nil, err
	}

	slowlogTable := s.getDuckDBSlowLogBaseURL(cluster.TenantID, metaInfo.O11yID, cluster.Env, cluster.Provider, cluster.Region, cluster.SharedTier, cluster.NextGen)
	//tableAlias := tableAlias(cluster.ClusterID)

	// Build the SQL query for stats
	query := fmt.Sprintf(`
		SELECT 
			avg(disk_max) as avg_disk, 
			avg(query_time) as avg_latency, 
			avg(mem_max) as avg_memory, 
			count(*) as count, 
			first(db) as db, 
			max(disk_max) as max_disk, 
			max(query_time) as max_latency, 
			max(mem_max) as max_memory, 
			digest as sql_digest, 
			first(query) as sql_text, 
			sum(query_time) as sum_latency, 
			sum(mem_max) as sum_memory, 
			sum(disk_max) as sum_disk,
			CASE
			    WHEN first(query) ILIKE 'select%%' THEN 'Select'
			    WHEN first(query) ILIKE 'insert%%' THEN 'Insert'
			    WHEN first(query) ILIKE 'update%%' THEN 'Update'
			    WHEN first(query) ILIKE 'delete%%' THEN 'Delete'
			    WHEN first(query) ILIKE 'begin%%' THEN 'Begin'
			    WHEN first(query) ILIKE 'change%%' THEN 'Change'
			    WHEN first(query) ILIKE 'commit%%' THEN 'Commit'
			    WHEN first(query) ILIKE 'analyze%%' THEN 'AnalyzeTable'
			    WHEN first(query) ILIKE 'alter table%%' THEN 'AlterTable'
			    WHEN first(query) ILIKE 'show%%' THEN 'Show'
			    WHEN first(query) ILIKE 'set%%' THEN 'Set'
			    ELSE 'Others'
			END as statement_type
		FROM delta_scan('%s') 
		WHERE time >= %d AND time <= %d AND (is_internal = false OR is_internal IS NULL)
	`, slowlogTable, beginHourTs, beginHourTs+int64(hours)*3600)

	// Add database filter if specified
	if len(databases) > 0 {
		query += fmt.Sprintf(" AND db IN ('%s')", strings.Join(databases, "','"))
	}

	// Group by digest and order by sum_latency
	query += " GROUP BY digest "
	// Add statement type
	if len(statementTypes) > 0 {
		query += fmt.Sprintf(" HAVING statement_type IN ('%s')", strings.Join(statementTypes, "','"))
	}
	query += fmt.Sprintf(" ORDER BY %s DESC", order)

	// Add limit if specified
	if limit > 0 {
		query += fmt.Sprintf(" LIMIT %d", limit)
	}

	log.Debug("DuckDB Stats Query", zap.String("query", query))

	// Execute the query using SelectSlowlogInDLWithoutFilterPushdown
	type statsResult struct {
		AvgDisk       sql.NullFloat64 `db:"avg_disk"`
		AvgLatency    sql.NullFloat64 `db:"avg_latency"`
		AvgMemory     sql.NullFloat64 `db:"avg_memory"`
		Count         int             `db:"count"`
		DB            sql.NullString  `db:"db"`
		MaxDisk       sql.NullFloat64 `db:"max_disk"`
		MaxLatency    sql.NullFloat64 `db:"max_latency"`
		MaxMemory     sql.NullFloat64 `db:"max_memory"`
		SQLDigest     string          `db:"sql_digest"`
		SQLText       string          `db:"sql_text"`
		SumLatency    sql.NullFloat64 `db:"sum_latency"`
		SumMemory     sql.NullFloat64 `db:"sum_memory"`
		SumDisk       sql.NullFloat64 `db:"sum_disk"`
		StatementType string          `db:"statement_type"`
	}
	// Execute the query using a custom query function that handles the custom result type
	var dlResults []statsResult
	err = s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &dlResults, "", slowlogTable, query)
	if err != nil {
		log.Error("Failed to execute stats query",
			zap.Error(err),
			zap.String("query", query),
			zap.String("clusterID", cluster.ClusterID))
		return nil, fmt.Errorf("failed to execute stats query: %w", err)
	}

	// Process results
	results := make([]slowstats.SlowStats, 0, len(dlResults))

	// Helper function to get float64 value from sql.NullFloat64 field
	getFloat64 := func(nf sql.NullFloat64) float64 {
		if nf.Valid {
			return nf.Float64
		}
		return 0
	}

	for _, dl := range dlResults {
		// Create a new SlowStats object and populate it with the query results
		result := slowstats.SlowStats{
			AvgDisk:    getFloat64(dl.AvgDisk),
			AvgLatency: getFloat64(dl.AvgLatency),
			AvgMemory:  getFloat64(dl.AvgMemory),
			Count:      dl.Count,
			DB:         dl.DB.String,
			MaxDisk:    getFloat64(dl.MaxDisk),
			MaxLatency: getFloat64(dl.MaxLatency),
			MaxMemory:  getFloat64(dl.MaxMemory),
			SQLDigest:  dl.SQLDigest,
			SQLText:    dl.SQLText,
			SumLatency: getFloat64(dl.SumLatency),
			SumMemory:  getFloat64(dl.SumMemory),
			SumDisk:    getFloat64(dl.SumDisk),
		}

		results = append(results, result)
	}

	return results, nil
}

// Analyze is not implemented for DuckDBAdapter.
func (s *DuckDBAdapter) Analyze(ctx context.Context, params *AnalyzeParams) error {
	//TODO implement me
	panic("Analyze not implemented for SlowQueryDuckDBAdapter")
}

// AvailableFields returns API field names present in the cluster's slowlog Delta Lake schema.
func (s *DuckDBAdapter) AvailableFields(ctx context.Context, params *AvailableFieldsParams) ([]string, error) {
	log.Info("SlowQueryDuckDBAdapter AvailableFields called.",
		zap.String("provider", params.Provider),
		zap.String("region", params.Region),
		zap.String("tenant", params.TenantID),
		zap.String("project", params.ProjectID),
		zap.String("cluster", params.ClusterID),
		zap.Bool("sharedTier", params.SharedTier),
	)

	metaInfo, err := s.mataCache.GetMetaInfo(ctx, params.Env, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.NextGen, params.Byoc)
	if metaInfo == nil {
		return nil, fmt.Errorf("meta info not found for tenant %s and cluster %s", params.TenantID, params.ClusterID)
	}
	if err != nil {
		log.Error("GetMetaInfo failed", zap.Error(err))
		return nil, err
	}

	slowlogTable := s.getDuckDBSlowLogBaseURL(params.TenantID, metaInfo.O11yID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen)
	alias := tableAlias(params.ClusterID)

	columnNames, err := s.getSlowlogTableColumns(ctx, slowlogTable, alias)
	if err != nil {
		return nil, err
	}

	fields := buildAvailableFieldsFromSchema(slowlogColumnNameSet(columnNames))
	log.Info("SlowQueryDuckDBAdapter AvailableFields finished",
		zap.String("clusterID", params.ClusterID),
		zap.Int("fieldCount", len(fields)),
	)
	return fields, nil
}

// Download is not implemented for DuckDBAdapter.
func (s *DuckDBAdapter) Download(ctx context.Context, params *DownloadParams) (*DownloadData, error) {
	//TODO implement me
	panic("Download not implemented for SlowQueryDuckDBAdapter")
}

// Databases returns a list of distinct database names from slow logs within the specified time range.
func (s *DuckDBAdapter) Databases(ctx context.Context, cluster slowstats.ClusterInfo, beginHourTs int64, hours int, internal bool) ([]string, error) {
	log.Info("SlowQueryDuckDBAdapter Databases called",
		zap.String("provider", cluster.Provider),
		zap.String("region", cluster.Region),
		zap.String("tenant", cluster.TenantID),
		zap.String("project", cluster.ProjectID),
		zap.String("cluster", cluster.ClusterID),
		zap.String("env", cluster.Env),
		zap.Bool("sharedTier", cluster.SharedTier),
		zap.Int64("beginTime", beginHourTs),
		zap.Int("hours", hours),
	)

	metaInfo, err := s.mataCache.GetMetaInfo(ctx, cluster.Env, cluster.Region, cluster.TenantID, cluster.ProjectID, cluster.ClusterID, cluster.NextGen, cluster.Byoc)
	if err != nil {
		log.Error("GetMetaInfo failed", zap.Error(err))
		return nil, err
	}
	slowlogTable := s.getDuckDBSlowLogBaseURL(cluster.TenantID, metaInfo.O11yID, cluster.Env, cluster.Provider, cluster.Region, cluster.SharedTier, cluster.NextGen)
	//alias := tableAlias(cluster.ClusterID)

	// Build the SQL query to get distinct databases
	if hours < 1 {
		hours = 1
	}
	beginTimeFloat := float64(beginHourTs)
	endTimeFloat := float64(beginHourTs + int64(hours)*3600)
	var internalCondition string
	if !internal {
		internalCondition = "AND (is_internal = false OR is_internal IS NULL)"
	}
	query := fmt.Sprintf(`
		SELECT DISTINCT db 
		FROM delta_scan('%s')
		WHERE time >= %.6f AND time <= %.6f
		  AND date >= '%s' AND date <= '%s'
		  AND db IS NOT NULL
		  AND db != ''
		  %s
		ORDER BY db
	`,
		slowlogTable,
		beginTimeFloat,
		endTimeFloat,
		time.Unix(beginHourTs, 0).UTC().Format("20060102"),
		time.Unix(beginHourTs+int64(hours)*3600, 0).UTC().Format("20060102"),
		internalCondition)

	log.Debug("DuckDB Databases Query", zap.String("query", query))

	// Execute the query
	var dlResults []*SlowlogInDL
	err = s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &dlResults, "", slowlogTable, query)
	if err != nil {
		log.Error("Failed to query databases from DuckDB",
			zap.Error(err),
			zap.String("query", query),
			zap.String("clusterID", cluster.ClusterID))
		return nil, fmt.Errorf("failed to query databases from DuckDB: %w", err)
	}

	// Extract database names from results
	databases := make([]string, 0, len(dlResults))
	for _, dl := range dlResults {
		if dl.DB.Valid && dl.DB.String != "" {
			databases = append(databases, dl.DB.String)
		}
	}

	log.Info("SlowQueryDuckDBAdapter Databases finished", zap.Int("count", len(databases)))
	return databases, nil
}

// getDuckDBSlowLogBaseURL constructs the base path for DuckDB to read the slow log data table
// based on the cloud provider.
func (s *DuckDBAdapter) getDuckDBSlowLogBaseURL(tenantID, o11yClusterID, env, provider, region string, sharedTier, nextGen bool) string {
	var scheme string

	// The bucket name is determined by the `bucketName` function, which you confirmed
	// works for all providers with your current naming convention.
	bucket := storage.BucketName(env, region, sharedTier, nextGen)

	// Determine the URI scheme based on the provider
	switch strings.ToLower(provider) {
	case "aws":
		scheme = fmt.Sprintf("s3://%s", bucket)
	case "gcp":
		scheme = fmt.Sprintf("gs://%s", bucket) // Google Cloud Storage uses gs:// prefix
	case "azure":
		account := os.Getenv("AZURE_STORAGE_ACCOUNT")
		scheme = fmt.Sprintf("abfss://%s@%s.dfs.core.windows.net", bucket, account) // Azure Blob Storage uses abfss:// prefix
	case "alicloud":
		scheme = fmt.Sprintf("s3://%s", bucket)
	default:
		// Fallback or error if provider is not recognized.
		log.Error("Unsupported cloud provider for slow log base URL", zap.String("provider", provider))
		return "" // Return empty string to indicate an error, which should be handled by the caller
	}

	// Construct the full path
	return fmt.Sprintf("%s/deltalake/%s/%s/slowlogs", scheme, tenantID, o11yClusterID)
}

var (
	labelBizType      = "observability.pingcap.com/bizType"
	labelBizTypeValue = "tidbcloud"

	labelTenantID      = "observability.pingcap.com/tenantID"
	labelApplicationID = "observability.pingcap.com/applicationID"
)

func (s *DuckDBAdapter) getClusterTables(ctx context.Context) (map[string]string, error) {
	tablesMap := make(map[string]string)

	clusters := &o11yv1.ClusterList{}
	if err := s.kubeCli.List(ctx, clusters, client.MatchingLabels{
		labelBizType: labelBizTypeValue,
	}); err != nil {
		return nil, err
	}

	for _, cluster := range clusters.Items {
		if cluster.Labels[labelBizType] != labelBizTypeValue {
			continue
		}
		applicationID := cluster.Labels[labelApplicationID]
		if applicationID == "dataapp" ||
			applicationID == "k8s-infra" ||
			applicationID == "seed" ||
			applicationID == "base" {
			continue
		}
		if isPureDigits(applicationID) {
			tableUrl := s.getDuckDBSlowLogBaseURL(cluster.Labels[labelTenantID],
				cluster.Name,
				"prod",
				s.vendor,
				s.region,
				false,
				false,
			)
			tablesMap[tableAlias(applicationID)] = tableUrl
		}
	}
	return tablesMap, nil
}

func isPureDigits(s string) bool {
	if s == "" {
		return false
	}
	for _, r := range s {
		if !unicode.IsDigit(r) {
			return false
		}
	}
	return true
}

// mapSlowlogInDLToSlowLog converts a duckdb.SlowlogInDL struct to a SlowLog struct.
func mapSlowlogInDLToSlowLog(dl *SlowlogInDL) *SlowLog {
	sl := &SlowLog{
		Time:                        timeFromFloat(dl.Time),
		Digest:                      dl.Digest,
		Instance:                    dl.Instance.String,
		TxnStartTs:                  parseStringToUint64(dl.TxnStartTS.String),
		User:                        dl.User.String,
		Host:                        dl.Host.String,
		ConnID:                      parseStringToUint64(dl.ConnID.String),
		SessionAlias:                dl.SessionAlias.String,
		ExecRetryCount:              parseStringToUint64(dl.ExecRetryCount.String),
		ExecRetryTime:               dl.ExecRetryTime.Float64,
		QueryTime:                   dl.QueryTime.Float64,
		ParseTime:                   dl.ParseTime.Float64,
		CompileTime:                 dl.CompileTime.Float64,
		RewriteTime:                 dl.RewriteTime.Float64,
		PreprocSubqueries:           parseStringToUint64(dl.PreprocSubqueries.String),
		PreprocSubqueriesTime:       dl.PreprocSubqueriesTime.Float64,
		OptimizeTime:                dl.OptimizeTime.Float64,
		WaitTs:                      dl.WaitTS.Float64,
		PrewriteTime:                dl.PrewriteTime.Float64,
		WaitPrewriteBinlogTime:      dl.WaitPrewriteBinlogTime.Float64,
		CommitTime:                  dl.CommitTime.Float64,
		GetCommitTsTime:             dl.GetCommitTSTime.Float64,
		CommitBackoffTime:           dl.CommitBackoffTime.Float64,
		BackoffTypes:                dl.BackoffTypes.String,
		ResolveLockTime:             dl.ResolveLockTime.Float64,
		LocalLatchWaitTime:          dl.LocalLatchWaitTime.Float64,
		WriteKeys:                   parseInt64ToUint64(dl.WriteKeys.Int64),
		WriteSize:                   parseInt64ToUint64(dl.WriteSize.Int64),
		PrewriteRegion:              parseInt64ToUint64(dl.PrewriteRegion.Int64),
		TxnRetry:                    parseInt64ToUint64(dl.TxnRetry.Int64),
		CopTime:                     dl.CopTime.Float64,
		ProcessTime:                 dl.ProcessTime.Float64,
		WaitTime:                    dl.WaitTime.Float64,
		BackoffTime:                 dl.BackoffTime.Float64,
		LockKeysTime:                dl.LockkeysTime.Float64,
		RequestCount:                parseStringToUint64(dl.RequestCount.String),
		TotalKeys:                   parseStringToUint64(dl.TotalKeys.String),
		ProcessKeys:                 parseStringToUint64(dl.ProcessKeys.String),
		RocksdbDeleteSkippedCount:   parseStringToUint64(dl.RocksdbDeleteSkippedCount.String),
		RocksdbKeySkippedCount:      parseStringToUint64(dl.RocksdbKeySkippedCount.String),
		RocksdbBlockCacheHitCount:   parseStringToUint64(dl.RocksdbBlockCacheHitCount.String),
		RocksdbBlockReadCount:       parseStringToUint64(dl.RocksdbBlockReadCount.String),
		RocksdbBlockReadByte:        parseStringToUint64(dl.RocksdbBlockReadByte.String),
		DB:                          dl.DB.String,
		IndexNames:                  dl.IndexNames.String,
		IsInternal:                  dl.IsInternal.Bool,
		Stats:                       dl.Stats.String,
		CopProcAvg:                  dl.CopProcAvg.Float64,
		CopProcP90:                  dl.CopProcP90.Float64,
		CopProcMax:                  dl.CopProcMax.Float64,
		CopProcAddr:                 dl.CopProcAddr.String,
		CopWaitAvg:                  dl.CopWaitAvg.Float64,
		CopWaitP90:                  dl.CopWaitP90.Float64,
		CopWaitMax:                  dl.CopWaitMax.Float64,
		CopWaitAddr:                 dl.CopWaitAddr.String,
		MemMax:                      parseInt64ToUint64(dl.MemMax.Int64),
		DiskMax:                     parseInt64ToUint64(dl.DiskMax.Int64),
		KVTotal:                     dl.KvTotal.Float64,
		PDTotal:                     dl.PdTotal.Float64,
		BackoffTotal:                dl.BackoffTotal.Float64,
		WriteSQLResponseTotal:       dl.WriteSQLResponseTotal.Float64,
		ResultRows:                  parseInt64ToUint64(dl.ResultRows.Int64),
		Warnings:                    dl.Warnings.String,
		BackoffDetail:               dl.BackoffDetail.String,
		Prepared:                    dl.Prepared.Bool,
		Succ:                        dl.Succ.Bool,
		IsExplicitTxn:               dl.IsExplicitTxn.Bool,
		IsWriteCacheTable:           dl.IsWriteCacheTable.Bool,
		PlanFromCache:               dl.PlanFromCache.Bool,
		PlanFromBinding:             dl.PlanFromBinding.Bool,
		HasMoreResults:              dl.HasMoreResults.Bool,
		ResourceGroup:               dl.ResourceGroup.String,
		RequestUnitRead:             dl.RequestUnitRead.Float64,
		RequestUnitWrite:            dl.RequestUnitWrite.Float64,
		RequestUnitV2:               dl.RequestUnitV2.Float64,
		RUv2:                        dl.RUv2.Float64,
		TimeQueuedByRc:              dl.TimeQueuedByRC.Float64,
		PlanDigest:                  dl.PlanDigest.String,
		BinaryPlan:                  dl.BinaryPlan.String,
		ReadPoolTaskDetails:         dl.ReadPoolTaskDetails.String,
		PrevStmt:                    dl.PrevStmt.String,
		Query:                       dl.Query.String,
		Plan:                        dl.Plan.String,
		ServerlessTenantID:          dl.ServerlessTenantID.String,
		ServerlessProjectID:         dl.ServerlessProjectID.String,
		ServerlessClusterID:         dl.ServerlessClusterID.String,
		IsTiflash:                   dl.IsTiflash.Bool,
		TidbCPUTime:                 dl.TidbCPUTime.Float64,
		TikvCPUTime:                 dl.TikvCPUTime.Float64,
		IARemoteReadSegmentCount:    parseStringToUint64(dl.IARemoteReadSegmentCount.String),
		IARemoteReadSegmentSize:     parseStringToUint64(dl.IARemoteReadSegmentSize.String),
		IARemoteReadSegmentWaitTime: dl.IARemoteReadSegmentWaitTime.Float64,
		RU:                          dl.RU.Float64,
	}
	return sl
}

// Helper function to parse string to uint64, returning 0 if invalid
func parseStringToUint64(s string) uint64 {
	val, err := strconv.ParseUint(s, 10, 64)
	if err != nil {
		return 0 // Return 0 or handle error as appropriate for your application
	}
	return val
}

// Helper function to convert int64 to uint64, returning 0 if negative
func parseInt64ToUint64(i int64) uint64 {
	if i < 0 {
		return 0 // Or handle as an error/specific default if negative values are not expected
	}
	return uint64(i)
}