package engine

import (
	"encoding/json"
	"testing"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dialect"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

const maxComputeTestProject = "clinic_proxy_test_ap_southeast_1"

func maxComputeTestDataSource(vendor, scheme, region string) *dto.DataSourceInfo {
	return &dto.DataSourceInfo{Vendor: vendor, Scheme: scheme, Region: region}
}

func TestMaxComputeSlowQueryParquetHintsScope(t *testing.T) {
	aliOSS := maxComputeTestDataSource("alicloud", "oss", "ap-southeast-1")
	rolloutRegions := []string{"ap-southeast-1"}
	physicalSlowQuery := "SELECT `time` FROM `" + maxComputeTestProject + "`.`slow_query_logs_1234567890123456789` WHERE `date` = '20260826'"
	tests := []struct {
		name           string
		enabled        bool
		allowedRegions []string
		project        string
		sql            string
		dataSource     *dto.DataSourceInfo
		wantHints      bool
	}{
		{name: "disabled", sql: physicalSlowQuery, project: maxComputeTestProject, dataSource: aliOSS},
		{name: "nil data source", enabled: true, sql: physicalSlowQuery, project: maxComputeTestProject},
		{name: "wrong vendor", enabled: true, sql: physicalSlowQuery, project: maxComputeTestProject, dataSource: maxComputeTestDataSource("aws", "oss", "ap-southeast-1")},
		{name: "wrong scheme", enabled: true, sql: physicalSlowQuery, project: maxComputeTestProject, dataSource: maxComputeTestDataSource("alicloud", "s3", "ap-southeast-1")},
		{name: "excluded region", enabled: true, allowedRegions: []string{"ap-northeast-1"}, sql: physicalSlowQuery, project: maxComputeTestProject, dataSource: aliOSS},
		{name: "included region case insensitive", enabled: true, allowedRegions: []string{" AP-SOUTHEAST-1 "}, sql: physicalSlowQuery, project: maxComputeTestProject, dataSource: aliOSS, wantHints: true},
		{name: "empty region allowlist disables rollout", enabled: true, sql: physicalSlowQuery, project: maxComputeTestProject, dataSource: aliOSS},
		{name: "logical slow query table", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT time FROM slow_query_logs WHERE date = '20260826'", project: maxComputeTestProject, dataSource: aliOSS, wantHints: true},
		{name: "physical slow query table", enabled: true, allowedRegions: rolloutRegions, sql: physicalSlowQuery, project: maxComputeTestProject, dataSource: aliOSS, wantHints: true},
		{name: "ordered slow query list", enabled: true, allowedRegions: rolloutRegions, sql: physicalSlowQuery + " AND `time` >= 1787714580 AND `time` <= 1787725380 ORDER BY `time` DESC LIMIT 31", project: maxComputeTestProject, dataSource: aliOSS, wantHints: true},
		{name: "slow query CTE", enabled: true, allowedRegions: rolloutRegions, sql: "WITH recent AS (" + physicalSlowQuery + ") SELECT time FROM recent", project: maxComputeTestProject, dataSource: aliOSS, wantHints: true},
		{name: "slow query self join", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT a.time FROM `" + maxComputeTestProject + "`.`slow_query_logs_1` a JOIN `" + maxComputeTestProject + "`.`slow_query_logs_2` b ON a.digest = b.digest WHERE a.date = '20260826' AND b.date = '20260826'", project: maxComputeTestProject, dataSource: aliOSS, wantHints: true},
		{name: "all slow query union", enabled: true, allowedRegions: rolloutRegions, sql: physicalSlowQuery + " UNION ALL " + physicalSlowQuery, project: maxComputeTestProject, dataSource: aliOSS, wantHints: true},
		{name: "tableless select", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT 1", project: maxComputeTestProject, dataSource: aliOSS},
		{name: "malformed SQL", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT FROM", project: maxComputeTestProject, dataSource: aliOSS},
		{name: "insert select", enabled: true, allowedRegions: rolloutRegions, sql: "INSERT INTO target " + physicalSlowQuery, project: maxComputeTestProject, dataSource: aliOSS},
		{name: "topsql table", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT * FROM `" + maxComputeTestProject + "`.`topsql_1234567890123456789` WHERE date = '2026-08-26'", project: maxComputeTestProject, dataSource: aliOSS},
		{name: "mixed slow query and topsql", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT s.time FROM `" + maxComputeTestProject + "`.`slow_query_logs_1` s JOIN `" + maxComputeTestProject + "`.`topsql_1` t ON s.digest = t.sql_digest WHERE s.date = '20260826' AND t.date = '2026-08-26'", project: maxComputeTestProject, dataSource: aliOSS},
		{name: "view-like table", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT * FROM slow_query_logs_view_1 WHERE date = '20260826'", project: maxComputeTestProject, dataSource: aliOSS},
		{name: "non-numeric physical suffix", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT * FROM slow_query_logs_backup WHERE date = '20260826'", project: maxComputeTestProject, dataSource: aliOSS},
		{name: "cross project", enabled: true, allowedRegions: rolloutRegions, sql: "SELECT * FROM `other_project`.`slow_query_logs_1` WHERE date = '20260826'", project: maxComputeTestProject, dataSource: aliOSS},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			hints := maxComputeSlowQueryParquetHints(
				tt.enabled,
				tt.allowedRegions,
				tt.project,
				tt.sql,
				tt.dataSource,
			)
			if !tt.wantHints {
				if hints != nil {
					t.Fatalf("hints = %#v, want nil", hints)
				}
				return
			}
			want := map[string]string{
				maxComputeParquetNativeHint:            "true",
				maxComputeParquetPredicatePushdownHint: "true",
			}
			if len(hints) != len(want) {
				t.Fatalf("hints = %#v, want %#v", hints, want)
			}
			for key, value := range want {
				if hints[key] != value {
					t.Fatalf("hints[%q] = %q, want %q", key, hints[key], value)
				}
			}
		})
	}
}

func TestMaxComputeSlowQueryParquetHintsAfterExplorerSQLRewrite(t *testing.T) {
	const clusterID = "1234567890123456789"
	logicalSQL := "SELECT time, query_time, plan_digest, total_keys, request_count, process_keys, db, tidb_cpu_time, tikv_cpu_time " +
		"FROM clinic_data_proxy.slow_query_logs " +
		"WHERE date IN ('20260825') AND time >= 1787628180 AND time <= 1787714580 " +
		"ORDER BY time DESC LIMIT 31"
	rewrittenSQL, err := dialect.RewriteForEngine(logicalSQL, "maxcompute", &dialect.RewriteOptions{
		TableNameMap: map[string]string{
			"slow_query_logs": "slow_query_logs_" + clusterID,
		},
		DatabaseNameMap: map[string]string{
			"clinic_data_proxy": maxComputeTestProject,
		},
		PartitionColumns: []string{dto.DefaultPartitionColumn},
	})
	if err != nil {
		t.Fatalf("rewrite representative Slow Log Explorer SQL: %v", err)
	}

	ds := maxComputeTestDataSource("alicloud", "oss", "ap-southeast-1")
	hints := maxComputeSlowQueryParquetHints(
		true,
		[]string{"ap-southeast-1"},
		maxComputeTestProject,
		rewrittenSQL,
		ds,
	)
	if hints[maxComputeParquetNativeHint] != "true" || hints[maxComputeParquetPredicatePushdownHint] != "true" {
		t.Fatalf("hints after Explorer SQL rewrite = %#v, want native Parquet and predicate pushdown", hints)
	}
}

func TestMaxComputeSlowQueryParquetHintsReturnsFreshMap(t *testing.T) {
	ds := maxComputeTestDataSource("alicloud", "oss", "ap-southeast-1")
	sql := "SELECT time FROM slow_query_logs WHERE date = '20260826'"
	regions := []string{"ap-southeast-1"}
	hints := maxComputeSlowQueryParquetHints(true, regions, maxComputeTestProject, sql, ds)
	hints[maxComputeParquetNativeHint] = "false"
	if got := maxComputeSlowQueryParquetHints(true, regions, maxComputeTestProject, sql, ds)[maxComputeParquetNativeHint]; got != "true" {
		t.Fatalf("fresh native Parquet hint = %q, want true", got)
	}
}

func TestNewMaxComputeQueryTaskSerializesParquetHints(t *testing.T) {
	hints := map[string]string{
		maxComputeParquetNativeHint:            "true",
		maxComputeParquetPredicatePushdownHint: "true",
	}
	task := newMaxComputeQueryTask("SELECT 1", hints)
	settings := ""
	for _, property := range task.Config {
		if property.Name == "settings" {
			settings = property.Value
			break
		}
	}
	if settings == "" {
		t.Fatal("MaxCompute SQL task is missing settings")
	}

	var got map[string]string
	if err := json.Unmarshal([]byte(settings), &got); err != nil {
		t.Fatalf("unmarshal MaxCompute task settings: %v", err)
	}
	if len(got) != len(hints) {
		t.Fatalf("task settings = %#v, want %#v", got, hints)
	}
	for key, value := range hints {
		if got[key] != value {
			t.Fatalf("task settings[%q] = %q, want %q", key, got[key], value)
		}
	}
	if task.Query != "SELECT 1;" {
		t.Fatalf("task query = %q, want SELECT 1;", task.Query)
	}
}

func TestNewMaxComputeQueryTaskOmitsSettingsWhenIneligible(t *testing.T) {
	task := newMaxComputeQueryTask("SELECT 1", nil)
	for _, property := range task.Config {
		if property.Name == "settings" {
			t.Fatalf("ineligible MaxCompute SQL task has settings %q", property.Value)
		}
	}
}