package service

import (
	"context"
	"crypto/sha256"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dialect"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/engine"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/engine_router"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"

	// pingcapLog "github.com/pingcap/log"
	"go.uber.org/zap"
)

type athenaTableManager interface {
	EnsureTableExists(context.Context, string, string, string, string) error
	EnsureViewExists(context.Context, string, string, string, string, string, string) error
	IsViewTemplate(string) bool
	RepairDeltaSchemaMismatch(
		context.Context,
		[]engine.TableReference,
		string,
		string,
		string,
		time.Time,
	) ([]string, error)
	RepairInstanceProjection(
		context.Context,
		[]engine.TableReference,
		string,
		string,
		string,
	) ([]engine.InstanceProjectionRepair, error)
}

type bigQueryTableManager interface {
	EnsureTableExists(context.Context, string, string, string, string) error
}

const defaultDataProxyQueryTimeout = 2 * time.Minute

const maxGCPTopSQLDateRangeDays = 31

// DataProxyService Clinic Data Proxy 核心服务
type DataProxyService struct {
	router *engine_router.Router
	config *config.DataProxyConfig
	// logger                 *zap.Logger
	tableManager           athenaTableManager             // AWS Athena 表管理器
	bigQueryTableManager   bigQueryTableManager           // GCP BigQuery 表管理器
	maxComputeTableManager *engine.MaxComputeTableManager // AliCloud MaxCompute 表管理器
}

// NewDataProxyService 创建服务实例
func NewDataProxyService(cfg *config.DataProxyConfig) (*DataProxyService, error) {
	r, err := engine_router.NewRouter(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create router: %w", err)
	}

	svc := &DataProxyService{
		router: r,
		config: cfg,
		// logger: pingcapLog.L(),
	}

	// 初始化 AWS Athena 表管理器
	if cfg.AWS.Enabled && cfg.AWS.TableManager.Enabled {
		athenaEngine, err := r.Route("aws")
		if err == nil {
			if athenaEng, ok := athenaEngine.(*engine.AthenaEngine); ok {
				svc.tableManager = engine.NewTableManager(athenaEng, &cfg.AWS.TableManager)
			}
		}
	}

	// 初始化 GCP BigQuery 表管理器（支持 Parquet Schema 自动识别）
	if cfg.GCP.Enabled && cfg.GCP.TableManager.Enabled {
		bqEngine, err := r.Route("gcp")
		if err == nil {
			if bqEng, ok := bqEngine.(*engine.BigQueryEngine); ok {
				svc.bigQueryTableManager = engine.NewBigQueryTableManager(bqEng, &cfg.GCP.TableManager)
			}
		}
	}

	// 初始化 AliCloud MaxCompute 表管理器
	if cfg.AliCloud.Enabled && cfg.AliCloud.TableManager.Enabled {
		maxComputeEngine, err := r.Route("alicloud")
		if err == nil {
			if mcEng, ok := maxComputeEngine.(*engine.MaxComputeEngine); ok {
				svc.maxComputeTableManager = engine.NewMaxComputeTableManager(
					mcEng, &cfg.AliCloud.TableManager)
			}
		}
	}

	return svc, nil
}

// Query 执行查询（核心方法）
func (s *DataProxyService) Query(ctx context.Context, req *dto.DataProxyQueryRequest) (*dto.DataProxyQueryResult, error) {
	startTime := time.Now()
	isStatementsQuery := queryReferencesLogicalTable(req.SQL, req.ClusterId, "statements")
	validatedStatementsSQL := ""
	if isStatementsQuery {
		if err := engine.ValidateStatementsTableReferences(engine.ExtractTableReferences(req.SQL), req.ClusterId); err != nil {
			return nil, err
		}
		var err error
		validatedStatementsSQL, err = validateAndNormalizeStatementsSQL(req.SQL)
		if err != nil {
			return nil, err
		}
		req.Options.SkipSQLRewrite = false
		req.Options.SkipTableManagement = false
	}

	// 1. 解析数据源
	dsInfo, err := engine_router.ParseAndDetectDataSource(ctx, req.Options.DataSource)
	if err != nil {
		return nil, fmt.Errorf("parse data source failed: %w", err)
	}

	// 允许用户覆盖自动检测的 Vendor/Region
	if req.Options.Vendor != "" {
		dsInfo.Vendor = req.Options.Vendor
	}
	if req.Options.Region != "" {
		dsInfo.Region = req.Options.Region
	}
	if dsInfo.Vendor == "alicloud" && isStatementsQuery {
		return nil, fmt.Errorf("statements queries are not supported for provider alicloud yet")
	}

	gcpTopSQLScopeSuffix := ""
	if dsInfo.Vendor == "gcp" && queryReferencesCanonicalLogicalTable(req.SQL, "topsql") {
		if req.Options.SkipSQLRewrite {
			return nil, fmt.Errorf("GCP TopSQL validation failed: skipSQLRewrite is not supported for logical queries")
		}
		dateScope, err := dialect.ExtractLiteralDateScope(req.SQL, dto.DefaultPartitionColumn, maxGCPTopSQLDateRangeDays)
		if err != nil {
			return nil, fmt.Errorf("GCP TopSQL date scope validation failed: %w", err)
		}
		gcpTopSQLScopeSuffix = topSQLMonthScopeSuffix(dateScope)
	}

	log.Info(ctx, "query routing",
		zap.String("vendor", dsInfo.Vendor),
		zap.String("region", dsInfo.Region),
	)

	// 1.5. 验证 AWS region 是否有效（仅针对 AWS）
	if dsInfo.Vendor == "aws" {
		if err := engine_router.ValidateAWSRegion(dsInfo.Region); err != nil {
			return nil, fmt.Errorf("invalid region: %w", err)
		}
	}

	// 2. 验证 ClusterId（必传，表名替换依赖此参数）
	if req.ClusterId == "" {
		return nil, fmt.Errorf("clusterId is required and cannot be empty")
	}

	// 3. 选择查询引擎
	queryEngine, err := s.router.Route(dsInfo.Vendor)
	if err != nil {
		return nil, fmt.Errorf("route engine failed: %w", err)
	}
	log.Info(ctx, "engine selected",
		zap.String("engine", queryEngine.Name()),
		zap.String("vendor", dsInfo.Vendor),
	)

	// 4. 根据 vendor 和配置重写 SQL（含 dialect 转换、表名、数据库名替换）
	configDBName := queryEngine.GetDatabase(dsInfo.Region)
	if configDBName == "" {
		configDBName = "clinic_data_proxy"
	}

	var rewrittenSQL string
	if dsInfo.Vendor == "azure" {
		if req.Options.SkipSQLRewrite {
			return nil, fmt.Errorf("skipSQLRewrite is not supported for Azure Synapse queries")
		}
		azureSQL := req.SQL
		if isStatementsQuery {
			azureSQL = validatedStatementsSQL
		}
		rewrittenSQL, err = engine.BuildAzureSynapseSQL(azureSQL, req.ClusterId, dsInfo)
		if err != nil {
			return nil, err
		}
		log.Info(ctx, "Azure SQL planned",
			zap.String("vendor", dsInfo.Vendor),
		)
	} else {
		// 检查 SQL 是否已包含 date 分区过滤（针对 CTE/子查询场景）
		// 如果已有 date IN (...) 过滤，跳过分区验证（因为验证器无法识别子查询中的条件）
		skipPartitionValidation := gcpTopSQLScopeSuffix != "" || (!isStatementsQuery && (strings.Contains(req.SQL, `date IN (`) ||
			strings.Contains(req.SQL, `"date" IN (`) ||
			strings.Contains(req.SQL, "`date` IN (")))

		tableNameMap := physicalTableNameMapWithTopSQLScope(req.ClusterId, req.Options.DataSource, gcpTopSQLScopeSuffix)
		rewriteOpts := &dialect.RewriteOptions{
			TableNameMap: tableNameMap,
			DatabaseNameMap: map[string]string{
				"clinic_data_proxy": configDBName,
				"clinic-data-proxy": configDBName,
			},
		}

		// 只有在 SQL 未包含 date 过滤时才启用分区验证
		if !skipPartitionValidation {
			rewriteOpts.PartitionColumns = []string{dto.DefaultPartitionColumn}
		}

		if req.Options.SkipSQLRewrite {
			// 跳过 TiDB parser 方言重写，只做表名和数据库名的字符串替换
			rewrittenSQL = req.SQL
			for from, to := range rewriteOpts.TableNameMap {
				rewrittenSQL = strings.ReplaceAll(rewrittenSQL, from, to)
			}
			for from, to := range rewriteOpts.DatabaseNameMap {
				rewrittenSQL = strings.ReplaceAll(rewrittenSQL, from, to)
			}
			log.Info(ctx, "SQL rewrite skipped (skipSQLRewrite=true), applied table/db name substitution only",
				zap.String("vendor", dsInfo.Vendor),
			)
		} else {
			rewrittenSQL, err = dialect.RewriteForEngine(req.SQL, queryEngine.Name(), rewriteOpts)
			if err != nil {
				return nil, fmt.Errorf("SQL rewrite failed: %w", err)
			}
			log.Info(ctx, "SQL rewritten",
				zap.String("vendor", dsInfo.Vendor),
			)
		}
	}

	queryCtx, cancelQuery := dataProxyQueryContext(ctx, req.Timeout)
	defer cancelQuery()

	if isStatementsQuery {
		availabilityTimeout := req.Timeout
		if availabilityTimeout <= 0 || availabilityTimeout > 30*time.Second {
			availabilityTimeout = 30 * time.Second
		}
		availabilityCtx, cancel := context.WithTimeout(queryCtx, availabilityTimeout)
		available, err := statementsDataAvailable(availabilityCtx, queryEngine, dsInfo)
		cancel()
		if err != nil {
			return nil, fmt.Errorf("check statements data availability: %w", err)
		}
		if !available {
			return emptyStatementsResult(queryEngine, dsInfo, time.Since(startTime)), nil
		}
	}

	var tableRefs []engine.TableReference
	if dsInfo.Vendor != "azure" {
		tableRefs = engine.ExtractTableReferences(rewrittenSQL)
		for i := range tableRefs {
			if tableRefs[i].Database == "" {
				tableRefs[i].Database = configDBName
			}
		}
	}

	// 5. 表管理：检查并自动创建表（AWS Athena / GCP BigQuery / AliCloud MaxCompute）
	// 如果设置了 SkipTableManagement，跳过表管理以提升性能
	if dsInfo.Vendor != "azure" && !req.Options.SkipTableManagement {
		switch dsInfo.Vendor {
		case "aws":
			if err := s.ensureTablesExist(queryCtx, tableRefs, req.Options.DataSource, dsInfo.Region, req.ClusterId, req.Options.ClinicTenantId); err != nil {
				log.Warn(ctx, "table management encountered issues", zap.String("vendor", dsInfo.Vendor))
				return nil, err
			}
		case "gcp":
			if err := s.ensureBigQueryTablesExist(queryCtx, tableRefs, req.Options.DataSource, dsInfo.Region); err != nil {
				log.Warn(ctx, "BigQuery table management encountered issues", zap.String("vendor", dsInfo.Vendor))
				return nil, err
			}
		case "alicloud":
			if s.maxComputeTableManager != nil {
				if err := s.ensureMaxComputeTablesExist(queryCtx, tableRefs, req.Options.DataSource, dsInfo.Region, req.ClusterId); err != nil {
					log.Warn(ctx, "MaxCompute table management encountered issues", zap.String("vendor", dsInfo.Vendor))
					return nil, err
				}
			}
		}
	}

	// 7. 执行查询
	firstQueryStartedAt := time.Now()
	result, err := s.executeQueryWithAthenaSchemaRepair(
		queryCtx,
		queryEngine,
		rewrittenSQL,
		dsInfo,
		tableRefs,
		req,
		firstQueryStartedAt,
	)
	if err != nil {
		fields := []zap.Field{
			zap.String("engine", queryEngine.Name()),
		}
		if unknownErr, ok := engine.AsUnknownFieldError(err); ok {
			fields = append(fields,
				zap.String("query_error", "unknown_field"),
				zap.String("query_id", unknownErr.QueryID),
				zap.Int("field_count", len(unknownErr.Fields)),
				zap.Strings("fields", unknownErr.Fields),
			)
		}
		fields = append(fields, bigQueryFailureLogFields(err)...)
		fields = append(fields, athenaQueryFailureLogFields(err, req.ClusterId)...)
		log.Error(ctx, "query execution failed", fields...)
		return nil, fmt.Errorf("query execution failed: %w", err)
	}

	// 8. 填充元数据
	result.Metadata.ExecutionTime = dto.FormattedDuration(time.Since(startTime))
	if result.Metadata.Vendor == "" {
		result.Metadata.Vendor = dsInfo.Vendor
	}
	if result.Metadata.Region == "" {
		result.Metadata.Region = dsInfo.Region
	}
	if result.Metadata.Engine == "" {
		result.Metadata.Engine = queryEngine.Name()
	}

	log.Info(ctx, "query completed",
		zap.String("query_id", result.Metadata.QueryID),
		zap.Int64("row_count", result.Metadata.RowCount),
		zap.Duration("execution_time", time.Duration(result.Metadata.ExecutionTime)),
		zap.String("engine", result.Metadata.Engine),
	)

	return result, nil
}

func dataProxyQueryContext(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) {
	if timeout <= 0 {
		timeout = defaultDataProxyQueryTimeout
	}
	return context.WithTimeout(parent, timeout)
}

// BatchQueryFailure preserves the request index and error chain for one failed
// item in an all-or-nothing batch.
type BatchQueryFailure struct {
	Index int
	Err   error
}

// BatchQueryError contains deterministically ordered failures from a batch.
type BatchQueryError struct {
	Failures []BatchQueryFailure
}

func (e *BatchQueryError) Error() string {
	if e == nil {
		return "batch query failed"
	}
	return fmt.Sprintf("batch query failed with %d errors", len(e.Failures))
}

func (s *DataProxyService) executeQueryWithAthenaSchemaRepair(
	ctx context.Context,
	queryEngine engine.QueryEngine,
	sql string,
	dataSource *dto.DataSourceInfo,
	tableRefs []engine.TableReference,
	req *dto.DataProxyQueryRequest,
	firstQueryStartedAt time.Time,
) (*dto.DataProxyQueryResult, error) {
	result, err := queryEngine.Execute(ctx, sql, dataSource)
	if err == nil ||
		dataSource.Vendor != "aws" ||
		req.Options.SkipTableManagement ||
		s.tableManager == nil {
		return result, err
	}

	firstQueryID := engine.AthenaQueryID(err)
	repairName := ""
	repairedTables := []string(nil)
	var repairErr error
	switch {
	case engine.IsInvalidGlueSchema(err):
		repairName = "Delta schema"
		repairedTables, repairErr = s.tableManager.RepairDeltaSchemaMismatch(
			ctx,
			tableRefs,
			req.Options.DataSource,
			dataSource.Region,
			req.ClusterId,
			firstQueryStartedAt,
		)
	case engine.IsInvalidInstanceProjection(err):
		repairName = "instance projection"
		var repairs []engine.InstanceProjectionRepair
		repairs, repairErr = s.tableManager.RepairInstanceProjection(
			ctx,
			tableRefs,
			req.Options.DataSource,
			dataSource.Region,
			req.ClusterId,
		)
		for _, repair := range repairs {
			repairedTables = append(repairedTables, repair.Table)
			log.Info(ctx, "Athena instance projection reconciled",
				zap.String("table", repair.Table),
				zap.Int("instance_count", repair.InstanceCount),
				zap.Bool("updated", repair.Updated),
			)
		}
	default:
		return result, err
	}
	if repairErr != nil {
		log.Error(ctx, "Athena table repair failed",
			zap.String("query_id", firstQueryID),
			zap.String("repair", repairName),
			zap.String("region", dataSource.Region),
			zap.Error(repairErr),
		)
		return nil, fmt.Errorf(
			"query execution failed: %w; Athena %s repair failed: %v",
			err,
			repairName,
			repairErr,
		)
	}

	log.Info(ctx, "Athena table repaired; retrying query once",
		zap.String("query_id", firstQueryID),
		zap.String("repair", repairName),
		zap.String("region", dataSource.Region),
		zap.Strings("tables", repairedTables),
	)
	result, err = queryEngine.Execute(ctx, sql, dataSource)
	if err != nil {
		log.Error(ctx, "query retry after Athena table repair failed",
			zap.String("first_query_id", firstQueryID),
			zap.String("retry_query_id", engine.AthenaQueryID(err)),
			zap.String("repair", repairName),
			zap.String("region", dataSource.Region),
		)
		return nil, fmt.Errorf("query execution failed after Athena %s repair: %w", repairName, err)
	}
	return result, nil
}

func physicalTableNameMap(clusterID, statementsDataSource string) map[string]string {
	return physicalTableNameMapWithTopSQLScope(clusterID, statementsDataSource, "")
}

func physicalTableNameMapWithTopSQLScope(clusterID, statementsDataSource, topSQLScopeSuffix string) map[string]string {
	return map[string]string{
		"slow_query_logs":  fmt.Sprintf("slow_query_logs_%s", clusterID),
		"statements":       statementsPhysicalTableName(clusterID, statementsDataSource),
		"topsql":           fmt.Sprintf("topsql_%s%s", clusterID, topSQLScopeSuffix),
		"topsql_tidb":      fmt.Sprintf("topsql_tidb_%s", clusterID),
		"topsql_tikv":      fmt.Sprintf("topsql_tikv_%s", clusterID),
		"topsql_plan_meta": fmt.Sprintf("topsql_plan_meta_%s", clusterID),
		"topsql_sql_meta":  fmt.Sprintf("topsql_sql_meta_%s", clusterID),
		"topsql_topru":     fmt.Sprintf("topsql_topru_%s", clusterID),
		"topru":            fmt.Sprintf("topru_%s", clusterID),
		"log_stats":        fmt.Sprintf("log_stats_%s", clusterID),
	}
}

func statementsPhysicalTableName(clusterID, dataSource string) string {
	canonicalSource := strings.TrimRight(strings.TrimSpace(dataSource), "/")
	fingerprint := sha256.Sum256([]byte(canonicalSource))
	return fmt.Sprintf("statements_%s_%x", clusterID, fingerprint[:8])
}

func validateAndNormalizeStatementsSQL(sql string) (string, error) {
	normalized, err := dialect.RewriteForEngine(sql, "", &dialect.RewriteOptions{
		PartitionColumns: []string{dto.DefaultPartitionColumn},
	})
	if err != nil {
		return "", fmt.Errorf("SQL validation failed: %w", err)
	}
	return normalized, nil
}

func queryReferencesLogicalTable(sql, clusterID, logicalTable string) bool {
	for _, ref := range engine.ExtractTableReferences(sql) {
		if logicalTable == "statements" && engine.IsStatementsTableReference(ref.Table, clusterID) {
			return true
		}
		if engine.LogicalStandardTableName(ref.Table, clusterID) == logicalTable {
			return true
		}
	}
	return false
}

func queryReferencesCanonicalLogicalTable(sql, logicalTable string) bool {
	for _, ref := range engine.ExtractTableReferences(sql) {
		if strings.EqualFold(strings.TrimSpace(ref.Table), logicalTable) {
			return true
		}
	}
	return false
}

func topSQLMonthScopeSuffix(scope dialect.DateScope) string {
	lastIncluded := scope.End.AddDate(0, 0, -1)
	month := time.Date(scope.Start.Year(), scope.Start.Month(), 1, 0, 0, 0, 0, time.UTC)
	lastMonth := time.Date(lastIncluded.Year(), lastIncluded.Month(), 1, 0, 0, 0, 0, time.UTC)
	parts := make([]string, 0, 2)
	for !month.After(lastMonth) {
		parts = append(parts, month.Format("200601"))
		month = month.AddDate(0, 1, 0)
	}
	if len(parts) == 0 {
		return ""
	}
	return "_m" + strings.Join(parts, "_")
}

type deltaLogInspector interface {
	HasDeltaLog(context.Context, *dto.DataSourceInfo) (bool, error)
}

func statementsDataAvailable(ctx context.Context, queryEngine engine.QueryEngine, dataSource *dto.DataSourceInfo) (bool, error) {
	inspector, ok := queryEngine.(deltaLogInspector)
	if !ok {
		return false, fmt.Errorf("engine %s does not support Delta availability checks", queryEngine.Name())
	}
	return inspector.HasDeltaLog(ctx, dataSource)
}

func emptyStatementsResult(queryEngine engine.QueryEngine, dataSource *dto.DataSourceInfo, elapsed time.Duration) *dto.DataProxyQueryResult {
	return &dto.DataProxyQueryResult{
		Columns: make([]string, 0),
		Rows:    make([][]interface{}, 0),
		Metadata: dto.DataProxyResultMetadata{
			RowCount:      0,
			ExecutionTime: dto.FormattedDuration(elapsed),
			Engine:        queryEngine.Name(),
			Vendor:        dataSource.Vendor,
			Region:        dataSource.Region,
		},
	}
}

// BatchQuery 批量查询（跨 Region 聚合）
func (s *DataProxyService) BatchQuery(ctx context.Context, requests []*dto.DataProxyQueryRequest) ([]*dto.DataProxyQueryResult, error) {
	if len(requests) == 0 {
		return nil, fmt.Errorf("no queries to execute")
	}

	log.Info(ctx, "batch query started",
		zap.Int("query_count", len(requests)),
	)

	results := make([]*dto.DataProxyQueryResult, len(requests))
	errChan := make(chan BatchQueryFailure, len(requests))

	var wg sync.WaitGroup
	for i, req := range requests {
		wg.Add(1)
		go func(index int, request *dto.DataProxyQueryRequest) {
			defer wg.Done()

			result, err := s.Query(ctx, request)
			if err != nil {
				errChan <- BatchQueryFailure{Index: index, Err: err}
				return
			}
			results[index] = result
		}(i, req)
	}

	wg.Wait()
	close(errChan)

	// 收集错误
	var failures []BatchQueryFailure
	for failure := range errChan {
		failures = append(failures, failure)
	}

	if len(failures) > 0 {
		failures = sortedBatchQueryFailures(failures)
		log.Error(ctx, "batch query failed",
			zap.Int("error_count", len(failures)),
		)
		return nil, &BatchQueryError{Failures: failures}
	}

	log.Info(ctx, "batch query completed",
		zap.Int("query_count", len(requests)),
		zap.Int("success_count", len(results)),
	)

	return results, nil
}

func sortedBatchQueryFailures(failures []BatchQueryFailure) []BatchQueryFailure {
	result := append([]BatchQueryFailure(nil), failures...)
	sort.Slice(result, func(i, j int) bool { return result[i].Index < result[j].Index })
	return result
}

// Health 健康检查
func (s *DataProxyService) Health(ctx context.Context) error {
	engines := s.router.GetEngines()

	log.Info(ctx, "health check started",
		zap.Int("engine_count", len(engines)),
	)

	var errs []error
	for vendor, eng := range engines {
		if err := eng.Health(ctx); err != nil {
			errs = append(errs, fmt.Errorf("engine %s unhealthy: %w", vendor, err))
			log.Warn(ctx, "engine unhealthy",
				zap.String("vendor", vendor),
				zap.String("engine", eng.Name()),
				zap.Error(err),
			)
		}
	}

	if len(errs) > 0 {
		return fmt.Errorf("some engines are unhealthy: %v", errs)
	}

	log.Info(ctx, "health check passed",
		zap.Int("engine_count", len(engines)),
	)

	return nil
}

// GetSupportedVendors 获取支持的云厂商列表
func (s *DataProxyService) GetSupportedVendors() []string {
	engines := s.router.GetEngines()
	vendors := make([]string, 0, len(engines))
	for vendor := range engines {
		vendors = append(vendors, vendor)
	}
	return vendors
}

// ensureTablesExist 确保 AWS Athena 表/VIEW 存在（如不存在则自动创建）
func (s *DataProxyService) ensureTablesExist(ctx context.Context, tableRefs []engine.TableReference, dataSource string, region string, clusterId string, clinicTenantId string) error {
	log.Info(ctx, "ensure Athena tables exist",
		zap.Int("table_count", len(tableRefs)),
		zap.String("region", region),
	)
	if s.tableManager == nil {
		log.Debug(ctx, "Athena table manager not enabled, skip")
		return nil
	}

	for _, ref := range tableRefs {
		if s.tableManager.IsViewTemplate(ref.Table) {
			// VIEW 类型：先创建依赖表，再创建 VIEW
			if err := s.tableManager.EnsureViewExists(ctx, ref.Database, ref.Table, dataSource, region, clusterId, clinicTenantId); err != nil {
				log.Warn(ctx, "failed to ensure Athena view exists",
					zap.String("database", ref.Database),
					zap.String("view", ref.Table),
				)
				return err
			} else {
				log.Info(ctx, "Athena view ensured",
					zap.String("database", ref.Database),
					zap.String("view", ref.Table),
				)
			}
		} else {
			// 物理表类型：对 TopSQL 子表，根据表名生成对应 component 的 S3 路径
			tableDataSource := engine.ResolveTopSQLDataSource(dataSource, ref.Table)
			if err := s.tableManager.EnsureTableExists(ctx, ref.Database, ref.Table, tableDataSource, region); err != nil {
				log.Warn(ctx, "failed to ensure Athena table exists",
					zap.String("database", ref.Database),
					zap.String("table", ref.Table),
				)
				return err
			} else {
				log.Info(ctx, "Athena table ensured",
					zap.String("database", ref.Database),
					zap.String("table", ref.Table),
				)
			}
		}
	}

	return nil
}

// ensureBigQueryTablesExist 确保 GCP BigQuery 表存在（如不存在则自动创建，支持 Parquet Schema 自动识别）
func (s *DataProxyService) ensureBigQueryTablesExist(ctx context.Context, tableRefs []engine.TableReference, dataSource, region string) error {
	log.Info(ctx, "ensure BigQuery tables exist",
		zap.Int("table_count", len(tableRefs)),
	)
	if s.bigQueryTableManager == nil {
		log.Debug(ctx, "BigQuery table manager not enabled, skip")
		return nil
	}

	for _, ref := range tableRefs {
		// 对 TopSQL 子表，根据表名生成对应 component 的数据路径
		tableDataSource := engine.ResolveTopSQLDataSource(dataSource, ref.Table)
		if err := s.bigQueryTableManager.EnsureTableExists(ctx, ref.Database, ref.Table, tableDataSource, region); err != nil {
			fields := []zap.Field{
				zap.String("dataset", ref.Database),
				zap.String("table", ref.Table),
			}
			if setupFields := bigQueryTopSQLSetupFailureLogFields(err); len(setupFields) > 0 {
				fields = append(fields, setupFields...)
				log.Warn(ctx, "TOPSQL_BIGQUERY_SETUP_FAILED", fields...)
			} else {
				log.Warn(ctx, "failed to ensure BigQuery table exists", fields...)
			}
			return err
		} else {
			log.Info(ctx, "BigQuery table ensured",
				zap.String("dataset", ref.Database),
				zap.String("table", ref.Table),
			)
		}
	}

	return nil
}

// ensureMaxComputeTablesExist 确保 AliCloud MaxCompute 表/VIEW 存在（VIEW：先建依赖外部表再 CREATE VIEW；物理表：OSS Parquet + MSCK REPAIR）
func (s *DataProxyService) ensureMaxComputeTablesExist(ctx context.Context, tableRefs []engine.TableReference, dataSource, region, clusterId string) error {
	log.Info(ctx, "ensure MaxCompute tables exist",
		zap.Int("table_count", len(tableRefs)),
		zap.String("region", region),
	)
	if s.maxComputeTableManager == nil {
		log.Debug(ctx, "MaxCompute table manager not enabled, skip")
		return nil
	}

	queryEngine, _ := s.router.Route("alicloud")
	for _, ref := range tableRefs {
		project := ref.Database
		if project == "" && queryEngine != nil {
			project = queryEngine.GetDatabase(region)
		}
		if s.maxComputeTableManager.IsViewTemplate(ref.Table) {
			if err := s.maxComputeTableManager.EnsureViewExists(ctx, project, ref.Table, dataSource, region, clusterId); err != nil {
				log.Warn(ctx, "failed to ensure MaxCompute view exists",
					zap.String("project", project),
					zap.String("view", ref.Table),
				)
			} else {
				log.Info(ctx, "MaxCompute view ensured",
					zap.String("project", project),
					zap.String("view", ref.Table),
				)
			}
			continue
		}
		tableDataSource := engine.ResolveTopSQLDataSource(dataSource, ref.Table)
		if err := s.maxComputeTableManager.EnsureTableExists(ctx, project, ref.Table, tableDataSource, region); err != nil {
			log.Warn(ctx, "failed to ensure MaxCompute table exists",
				zap.String("project", project),
				zap.String("table", ref.Table),
			)
			return err
		}
		log.Info(ctx, "MaxCompute table ensured",
			zap.String("project", project),
			zap.String("table", ref.Table),
		)
	}
	return nil
}

// GetSchema 获取表结构信息
// GetSchema 已废弃，请使用 GetSchemaForTable
// Deprecated: Use GetSchemaForTable instead
func (s *DataProxyService) GetSchema(ctx context.Context, database string, table string, options *dto.DataProxyQueryOptions) (*dto.SchemaResponse, error) {
	return s.GetSchemaForTable(ctx, table, options)
}

// GetSchemaForTable 获取单个表的 schema（新版 API 使用）
func (s *DataProxyService) GetSchemaForTable(ctx context.Context, table string, options *dto.DataProxyQueryOptions) (*dto.SchemaResponse, error) {
	if engine.LogicalStandardTableName(table, options.ClusterId) == "statements" {
		return nil, fmt.Errorf("schema API for statements is not supported; use POST /data-proxy/query")
	}

	// 1. 解析数据源
	dsInfo, err := engine_router.ParseAndDetectDataSource(ctx, options.DataSource)
	if err != nil {
		return nil, fmt.Errorf("parse data source failed: %w", err)
	}

	// 允许用户覆盖自动检测的 Vendor/Region
	if options.Vendor != "" {
		dsInfo.Vendor = options.Vendor
	}
	if options.Region != "" {
		dsInfo.Region = options.Region
	}

	log.Info(ctx, "get schema request",
		zap.String("table", table),
		zap.String("vendor", dsInfo.Vendor),
		zap.String("region", dsInfo.Region),
	)

	// 2. 支持 AWS Athena、GCP BigQuery、AliCloud MaxCompute、Azure Synapse
	if dsInfo.Vendor != "aws" && dsInfo.Vendor != "gcp" && dsInfo.Vendor != "alicloud" && dsInfo.Vendor != "azure" {
		return nil, fmt.Errorf("schema API is only supported for AWS Athena, GCP BigQuery, AliCloud, and Azure, got vendor: %s", dsInfo.Vendor)
	}

	if dsInfo.Vendor == "aws" {
		if err := engine_router.ValidateAWSRegion(dsInfo.Region); err != nil {
			return nil, fmt.Errorf("invalid region: %w", err)
		}
	}

	// 3. 获取查询引擎
	queryEngine, err := s.router.Route(dsInfo.Vendor)
	if err != nil {
		return nil, fmt.Errorf("route engine failed: %w", err)
	}

	if dsInfo.Vendor == "azure" {
		database := queryEngine.GetDatabase(dsInfo.Region)
		if database == "" {
			// wlt @todo review
			database = "clinic_data_proxy"
		}
		logicalTable := engine.LogicalStandardTableName(table, options.ClusterId)
		schema, ok := engine.StandardTableSchema(logicalTable)
		if !ok {
			return nil, fmt.Errorf("%s", engine.UnsupportedStandardTableError(logicalTable))
		}
		schema.Database = database
		schema.Table = table
		schema.DataSource = options.DataSource
		schema.Location = options.DataSource
		log.Info(ctx, "get Azure schema completed from standard registry",
			zap.String("database", schema.Database),
			zap.String("table", schema.Table),
			zap.Int("column_count", len(schema.Columns)),
			zap.Int("partition_count", len(schema.Partitions)),
		)
		return schema, nil
	}

	// 4. 确保表存在（如不存在则自动创建）
	database := queryEngine.GetDatabase(dsInfo.Region)
	tableRef := engine.TableReference{Database: database, Table: table}
	if !options.SkipTableManagement {
		switch dsInfo.Vendor {
		case "aws":
			if err := s.ensureTablesExist(ctx, []engine.TableReference{tableRef}, options.DataSource, dsInfo.Region, options.ClusterId, options.ClinicTenantId); err != nil {
				log.Warn(ctx, "failed to ensure Athena table exists before getting schema",
					zap.String("database", database),
					zap.String("table", table),
				)
				return nil, err
			}
		case "gcp":
			if err := s.ensureBigQueryTablesExist(ctx, []engine.TableReference{tableRef}, options.DataSource, dsInfo.Region); err != nil {
				log.Warn(ctx, "failed to ensure BigQuery table exists before getting schema",
					zap.String("dataset", database),
					zap.String("table", table),
				)
				return nil, err
			}
		case "alicloud":
			if s.maxComputeTableManager != nil {
				if err := s.ensureMaxComputeTablesExist(ctx, []engine.TableReference{tableRef}, options.DataSource, dsInfo.Region, options.ClusterId); err != nil {
					log.Warn(ctx, "failed to ensure MaxCompute table exists before getting schema",
						zap.String("database", database),
						zap.String("table", table),
					)
					return nil, err
				}
			}
		}
	}

	logicalTable := engine.LogicalStandardTableName(table, options.ClusterId)
	if logicalTable == "log_stats" {
		schema, ok := engine.StandardTableSchema(logicalTable)
		if !ok {
			return nil, fmt.Errorf("%s", engine.UnsupportedStandardTableError(logicalTable))
		}
		schema.Database = database
		schema.Table = table
		schema.DataSource = options.DataSource
		schema.Location = options.DataSource
		return schema, nil
	}

	// 5. 调用 GetTableSchema
	schema, err := queryEngine.GetTableSchema(ctx, database, table, dsInfo.Region)
	if err != nil {
		log.Error(ctx, "get table schema failed",
			zap.String("database", database),
			zap.String("table", table),
		)
		return nil, fmt.Errorf("get table schema failed: %w", err)
	}

	// 5. 填充 dataSource 字段
	schema.DataSource = options.DataSource

	log.Info(ctx, "get schema completed",
		zap.String("database", schema.Database),
		zap.String("table", schema.Table),
		zap.Int("column_count", len(schema.Columns)),
		zap.Int("partition_count", len(schema.Partitions)),
	)

	return schema, nil
}