package engine

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"go.uber.org/zap"
	"golang.org/x/sync/singleflight"
)

type maxComputeTableManagerEngine interface {
	CheckTableExists(context.Context, string, string, string) (bool, error)
	ExecuteStatement(context.Context, string, string) error
}

// MaxComputeTableManager MaxCompute 表管理器
// 支持自动创建 OSS Parquet 外部表，分区通过 MSCK REPAIR TABLE ... ADD PARTITIONS 同步
// endpoint 固定格式由 engine 按 region 推导
type MaxComputeTableManager struct {
	engine           maxComputeTableManagerEngine
	config           *config.TableManagerConfig
	cache            sync.Map // key = "table:project.table@region", value = *TableMetadata
	partitionCache   sync.Map // key = "partition:project.table@region:repaired", value = *PartitionMetadata
	prefetches       sync.Map // key = table cache key, value = *maxComputeTablePrefetch
	prefetchAttempts sync.Map // key = table cache key, value = cache generation last admitted for prefetch
	activePrefetches sync.Map // key = table cache key, value = active background marker
	prefetchSlot     chan struct{}
	prefetchGroup    singleflight.Group
	maintenanceGroup singleflight.Group
	mu               sync.Mutex
}

// NewMaxComputeTableManager 创建 MaxCompute 表管理器
func NewMaxComputeTableManager(engine *MaxComputeEngine, cfg *config.TableManagerConfig) *MaxComputeTableManager {
	return &MaxComputeTableManager{
		engine:       engine,
		config:       cfg,
		prefetchSlot: make(chan struct{}, 1),
	}
}

// EnsureTableExists 确保表存在，不存在则自动创建（OSS Parquet 外部表）
func (tm *MaxComputeTableManager) EnsureTableExists(ctx context.Context, database, table, dataSource, region string) error {
	if !tm.config.Enabled || !tm.config.AutoCreate {
		return nil
	}
	if isPhysicalMaxComputeSlowQueryTable(table) {
		return tm.ensureSlowQueryTableExists(ctx, database, table, dataSource, region)
	}

	project := database
	cacheKey := fmt.Sprintf("table:%s.%s@%s", project, table, region)
	if metadata, ok := tm.getCachedMetadata(cacheKey); ok {
		if metadata.Exists {
			return nil
		}
	}

	// 1. 检查表是否存在
	exists, err := tm.engine.CheckTableExists(ctx, project, table, region)
	if err != nil {
		return fmt.Errorf("failed to check table existence: %w", err)
	}

	if exists {
		tm.setCachedMetadata(cacheKey, &TableMetadata{
			Exists:      true,
			LastChecked: time.Now(),
		})
		_ = tm.repairPartitions(ctx, project, table, region)
		return nil
	}

	tm.mu.Lock()
	defer tm.mu.Unlock()

	// 2. 创建表
	if err := tm.createTableIfNotExists(ctx, project, table, dataSource, region); err != nil {
		if strings.Contains(err.Error(), "already exists") ||
			strings.Contains(err.Error(), "Table already exists") ||
			strings.Contains(err.Error(), "already exist") {
			tm.setCachedMetadata(cacheKey, &TableMetadata{
				Exists:      true,
				LastChecked: time.Now(),
			})
			return nil
		}
		return fmt.Errorf("failed to create table: %w", err)
	}

	log.Info(ctx, "MaxCompute external table created successfully",
		zap.String("project", project),
		zap.String("table", table),
		zap.String("region", region),
	)

	// 3. 同步分区（MSCK REPAIR TABLE ... ADD PARTITIONS）
	_ = tm.repairPartitions(ctx, project, table, region)

	tm.setCachedMetadata(cacheKey, &TableMetadata{
		Exists:      true,
		LastChecked: time.Now(),
	})

	return nil
}

func isPhysicalMaxComputeSlowQueryTable(table string) bool {
	return maxComputeSlowQueryTablePattern.MatchString(table) &&
		!strings.EqualFold(table, "slow_query_logs")
}

type maxComputeTablePrefetch struct {
	cacheGeneration int64
	exists          bool
	checkDuration   time.Duration
	validUntil      time.Time
}

type maxComputeTableCheckError struct {
	cause        error
	fromPrefetch bool
}

var errMaxComputeTablePrefetchExpired = errors.New("MaxCompute table prefetch expired before completion")

func (e *maxComputeTableCheckError) Error() string {
	return e.cause.Error()
}

func (e *maxComputeTableCheckError) Unwrap() error {
	return e.cause
}

// ensureSlowQueryTableExists prefetches only the table-existence check shortly
// before the existing hard TTL. It deliberately leaves the main table cache
// and partition cache untouched: the hard-TTL request consumes the prefetched
// result, runs the existing partition-repair policy, and publishes freshness.
func (tm *MaxComputeTableManager) ensureSlowQueryTableExists(
	ctx context.Context,
	project, table, dataSource, region string,
) error {
	cacheKey := maxComputeTableCacheKey(project, table, region)
	metadata, ok := tm.getRawCachedMetadata(cacheKey)
	if ok && metadata.Exists {
		ttl := tm.cacheTTL()
		if ttl == 0 {
			return nil
		}

		age := time.Since(metadata.LastChecked)
		if age < ttl {
			if age >= slowQueryPrefetchThreshold(ttl) {
				tm.startSlowQueryTablePrefetch(
					project,
					table,
					region,
					metadata.LastChecked,
				)
			}
			return nil
		}
	}

	_, err, _ := tm.maintenanceGroup.Do(cacheKey, func() (interface{}, error) {
		// MaxComputeEngine's existing table-management operations do not observe
		// request cancellation: the ODPS SDK Run/WaitForSuccess calls have no
		// context-aware variant. Keep that legacy behavior inside the shared
		// maintenance flight so a canceled leader cannot fail healthy waiters.
		maintenanceCtx := context.WithoutCancel(ctx)
		// Another hard-TTL/cold caller may have completed maintenance after this
		// caller read the cache and before it entered the maintenance flight.
		if current, loaded := tm.getRawCachedMetadata(cacheKey); loaded && current.Exists && current != metadata {
			return nil, nil
		}
		hardPathStartedAt := time.Now()
		existence, err := tm.slowQueryTableExists(maintenanceCtx, project, table, region, metadata)
		if err != nil {
			return nil, fmt.Errorf("failed to check table existence: %w", err)
		}
		return nil, tm.completeSlowQueryTableMaintenance(
			maintenanceCtx, project, table, dataSource, region, existence, hardPathStartedAt,
		)
	})
	return err
}

func (tm *MaxComputeTableManager) startSlowQueryTablePrefetch(
	project, table, region string,
	cacheGeneration time.Time,
) {
	cacheKey := maxComputeTableCacheKey(project, table, region)
	select {
	case tm.prefetchSlot <- struct{}{}:
	default:
		// Do not queue detached work. A later request in the prefetch window may
		// retry once the slot is free; the hard-TTL fallback remains synchronous.
		return
	}
	if !tm.admitSlowQueryTablePrefetch(cacheKey, cacheGeneration) {
		<-tm.prefetchSlot
		return
	}
	if _, loaded := tm.activePrefetches.LoadOrStore(cacheKey, struct{}{}); loaded {
		<-tm.prefetchSlot
		return
	}

	go func() {
		defer tm.activePrefetches.Delete(cacheKey)
		defer func() { <-tm.prefetchSlot }()
		// Do not retain request-scoped values or cancellation after the request
		// returns. A hard-TTL caller can join the same singleflight operation.
		ctx := context.Background()
		startedAt := time.Now()
		hardDeadline := cacheGeneration.Add(tm.cacheTTL())
		result, err := tm.checkSlowQueryTableExists(
			ctx,
			cacheKey,
			project,
			table,
			region,
			cacheGeneration.UnixNano(),
			hardDeadline,
		)
		if errors.Is(err, errMaxComputeTablePrefetchExpired) {
			log.Debug(ctx, "background MaxCompute slow-query table prefetch expired",
				zap.String("project", project),
				zap.String("table", table),
				zap.String("region", region),
				zap.Duration("duration", time.Since(startedAt)),
			)
			return
		}
		if err != nil {
			log.Warn(ctx, "background MaxCompute slow-query table prefetch failed",
				zap.String("project", project),
				zap.String("table", table),
				zap.String("region", region),
				zap.Duration("duration", time.Since(startedAt)),
				zap.Error(err),
			)
			return
		}
		log.Debug(ctx, "background MaxCompute slow-query table prefetch completed",
			zap.String("project", project),
			zap.String("table", table),
			zap.String("region", region),
			zap.Bool("exists", result.exists),
			zap.Duration("duration", time.Since(startedAt)),
		)
	}()
}

func (tm *MaxComputeTableManager) admitSlowQueryTablePrefetch(cacheKey string, generation time.Time) bool {
	generationUnixNano := generation.UnixNano()
	for {
		previous, loaded := tm.prefetchAttempts.LoadOrStore(cacheKey, generationUnixNano)
		if !loaded {
			return true
		}
		if previous.(int64) == generationUnixNano {
			return false
		}
		if tm.prefetchAttempts.CompareAndSwap(cacheKey, previous, generationUnixNano) {
			return true
		}
	}
}

func (tm *MaxComputeTableManager) slowQueryTableExists(
	ctx context.Context,
	project, table, region string,
	metadata *TableMetadata,
) (*maxComputeTablePrefetch, error) {
	cacheKey := maxComputeTableCacheKey(project, table, region)
	generation := int64(0)
	if metadata != nil {
		generation = metadata.LastChecked.UnixNano()
	}
	if prefetched, ok := tm.loadSlowQueryTablePrefetch(cacheKey, generation); ok {
		return prefetched, nil
	}
	return tm.checkSlowQueryTableExists(ctx, cacheKey, project, table, region, generation, time.Time{})
}

func (tm *MaxComputeTableManager) checkSlowQueryTableExists(
	ctx context.Context,
	cacheKey, project, table, region string,
	cacheGeneration int64,
	prefetchHardDeadline time.Time,
) (*maxComputeTablePrefetch, error) {
	flightKey := fmt.Sprintf("%s#generation=%d", cacheKey, cacheGeneration)
	value, err, _ := tm.prefetchGroup.Do(flightKey, func() (interface{}, error) {
		if prefetched, ok := tm.loadSlowQueryTablePrefetch(cacheKey, cacheGeneration); ok {
			return prefetched, nil
		}
		checkStartedAt := time.Now()
		exists, err := tm.checkMaxComputeTableExistsSafely(ctx, project, table, region)
		if err != nil {
			return nil, &maxComputeTableCheckError{
				cause:        err,
				fromPrefetch: !prefetchHardDeadline.IsZero(),
			}
		}
		checkDuration := time.Since(checkStartedAt)
		validUntil := time.Time{}
		if !prefetchHardDeadline.IsZero() {
			// Match the time at which the old foreground path would have finished
			// the same DESC: hard deadline plus this observed DESC duration. This
			// both covers long checks and bounds reuse to the remaining lead time.
			validUntil = prefetchHardDeadline.Add(checkDuration)
		}
		prefetched := &maxComputeTablePrefetch{
			cacheGeneration: cacheGeneration,
			exists:          exists,
			checkDuration:   checkDuration,
			validUntil:      validUntil,
		}
		if exists && !validUntil.IsZero() {
			tm.prefetches.Store(cacheKey, prefetched)
		}
		return prefetched, nil
	})
	if err != nil {
		if checkErr, ok := err.(*maxComputeTableCheckError); ok &&
			prefetchHardDeadline.IsZero() && checkErr.fromPrefetch {
			// A hard-TTL caller joined a failed detached prefetch. The optimization
			// must not turn that background failure into a foreground regression;
			// retry once through a new synchronous flight.
			return tm.checkSlowQueryTableExists(
				ctx, cacheKey, project, table, region, cacheGeneration, time.Time{},
			)
		}
		return nil, err
	}
	prefetched := value.(*maxComputeTablePrefetch)
	if !prefetched.exists && !prefetched.validUntil.IsZero() && prefetchHardDeadline.IsZero() {
		// Negative prefetches are advisory only. A hard-TTL caller must confirm
		// synchronously before creating a table or repairing partitions.
		return tm.checkSlowQueryTableExists(
			ctx, cacheKey, project, table, region, cacheGeneration, time.Time{},
		)
	}
	if !prefetched.usableFor(cacheGeneration, time.Now()) {
		// Do not consume a result for a newer or much later hard-TTL request.
		// Flights are generation-scoped, so the synchronous retry cannot collide
		// with work from another cache generation.
		if !prefetchHardDeadline.IsZero() {
			tm.prefetches.CompareAndDelete(cacheKey, prefetched)
			return nil, errMaxComputeTablePrefetchExpired
		}
		return tm.checkSlowQueryTableExists(
			ctx, cacheKey, project, table, region, cacheGeneration, time.Time{},
		)
	}
	return prefetched, nil
}

func (p *maxComputeTablePrefetch) usableFor(cacheGeneration int64, now time.Time) bool {
	if p.cacheGeneration != cacheGeneration {
		return false
	}
	return p.validUntil.IsZero() || !now.After(p.validUntil)
}

func (tm *MaxComputeTableManager) checkMaxComputeTableExistsSafely(
	ctx context.Context,
	project, table, region string,
) (exists bool, err error) {
	defer func() {
		if recovered := recover(); recovered != nil {
			err = fmt.Errorf("panic while checking MaxCompute table existence (%T)", recovered)
		}
	}()
	return tm.engine.CheckTableExists(ctx, project, table, region)
}

func (tm *MaxComputeTableManager) loadSlowQueryTablePrefetch(
	cacheKey string,
	cacheGeneration int64,
) (*maxComputeTablePrefetch, bool) {
	value, ok := tm.prefetches.Load(cacheKey)
	if !ok {
		return nil, false
	}
	prefetched, ok := value.(*maxComputeTablePrefetch)
	if !ok || prefetched.validUntil.IsZero() {
		return nil, false
	}
	if !prefetched.usableFor(cacheGeneration, time.Now()) {
		if prefetched.cacheGeneration == cacheGeneration {
			tm.prefetches.CompareAndDelete(cacheKey, prefetched)
		}
		return nil, false
	}
	return prefetched, true
}

func (tm *MaxComputeTableManager) completeSlowQueryTableMaintenance(
	ctx context.Context,
	project, table, dataSource, region string,
	existence *maxComputeTablePrefetch,
	hardPathStartedAt time.Time,
) error {
	cacheKey := maxComputeTableCacheKey(project, table, region)

	if existence.exists {
		tableFreshnessAt := time.Now()
		partitionCheckAt := tableFreshnessAt
		forcePartitionRepair := false
		if !existence.validUntil.IsZero() {
			// Reconstruct when this foreground request's old-path DESC would have
			// completed. If MSCK would be due then, wait until that instant before
			// repairing so partitions arriving during the removed DESC interval are
			// still discovered for this same request.
			partitionCheckAt = hardPathStartedAt.Add(existence.checkDuration)
			tableFreshnessAt = partitionCheckAt
			if tableFreshnessAt.After(existence.validUntil) {
				// Do not extend table-existence freshness beyond the bounded prefetch
				// reuse window when the first hard request arrived late.
				tableFreshnessAt = existence.validUntil
			}
			forcePartitionRepair = tm.partitionRepairDueBy(
				project, table, region, partitionCheckAt,
			)
			if forcePartitionRepair {
				waitForMaxComputeMaintenanceTime(partitionCheckAt)
			}
		}
		// Preserve the existing ordering: existing-table freshness is published
		// before the cached partition-repair check.
		tm.setCachedMetadata(cacheKey, &TableMetadata{
			Exists:      true,
			LastChecked: tableFreshnessAt,
		})
		if forcePartitionRepair {
			_ = tm.repairPartitionsNow(ctx, project, table, region)
		} else {
			_ = tm.repairPartitions(ctx, project, table, region)
		}
		tm.prefetches.CompareAndDelete(cacheKey, existence)
		return nil
	}

	tm.mu.Lock()
	defer tm.mu.Unlock()
	if err := tm.createTableIfNotExists(ctx, project, table, dataSource, region); err != nil {
		if isMaxComputeAlreadyExistsError(err) {
			tm.setCachedMetadata(cacheKey, &TableMetadata{
				Exists:      true,
				LastChecked: time.Now(),
			})
			tm.prefetches.CompareAndDelete(cacheKey, existence)
			return nil
		}
		return fmt.Errorf("failed to create table: %w", err)
	}

	log.Info(ctx, "MaxCompute external table created successfully",
		zap.String("project", project),
		zap.String("table", table),
		zap.String("region", region),
	)
	_ = tm.repairPartitions(ctx, project, table, region)
	tm.setCachedMetadata(cacheKey, &TableMetadata{
		Exists:      true,
		LastChecked: time.Now(),
	})
	tm.prefetches.CompareAndDelete(cacheKey, existence)
	return nil
}

func waitForMaxComputeMaintenanceTime(deadline time.Time) {
	delay := time.Until(deadline)
	if delay <= 0 {
		return
	}
	timer := time.NewTimer(delay)
	defer timer.Stop()
	<-timer.C
}

func isMaxComputeAlreadyExistsError(err error) bool {
	if err == nil {
		return false
	}
	message := err.Error()
	return strings.Contains(message, "already exists") ||
		strings.Contains(message, "Table already exists") ||
		strings.Contains(message, "already exist")
}

func maxComputeTableCacheKey(project, table, region string) string {
	return fmt.Sprintf("table:%s.%s@%s", project, table, region)
}

func maxComputePartitionCacheKey(project, table, region string) string {
	return fmt.Sprintf("partition:%s.%s@%s:repaired", project, table, region)
}

func (tm *MaxComputeTableManager) cacheTTL() time.Duration {
	return time.Duration(tm.config.CacheTTLSeconds) * time.Second
}

func slowQueryPrefetchThreshold(ttl time.Duration) time.Duration {
	if ttl <= 0 {
		return 0
	}
	refreshAhead := ttl / 10
	if refreshAhead > 30*time.Second {
		refreshAhead = 30 * time.Second
	}
	return ttl - refreshAhead
}

// IsViewTemplate 检查表名是否对应 VIEW 类型的模板（与 Athena TableManager 行为一致）
func (tm *MaxComputeTableManager) IsViewTemplate(tableName string) bool {
	template, ok := tm.findTableTemplate(tableName)
	return ok && template.IsView()
}

// EnsureViewExists 确保 VIEW 存在：先创建 DependsOn 中的物理表，再执行 CREATE VIEW（与 Athena 一致）
func (tm *MaxComputeTableManager) EnsureViewExists(ctx context.Context, project, viewName, dataSource, region, clusterId string) error {
	if !tm.config.Enabled || !tm.config.AutoCreate {
		return nil
	}

	cacheKey := fmt.Sprintf("table:%s.%s@%s", project, viewName, region)
	if metadata, ok := tm.getCachedMetadata(cacheKey); ok {
		if metadata.Exists {
			return nil
		}
	}

	template, ok := tm.findTableTemplate(viewName)
	if !ok {
		return fmt.Errorf("no template found for view: %s", viewName)
	}
	if !template.IsView() {
		return fmt.Errorf("template for %s is not a view type", viewName)
	}

	for _, depName := range template.DependsOn {
		actualDepTable := fmt.Sprintf("%s_%s", depName, clusterId)
		depDataSource := ResolveTopSQLDataSource(dataSource, actualDepTable)
		if err := tm.EnsureTableExists(ctx, project, actualDepTable, depDataSource, region); err != nil {
			log.Warn(ctx, "failed to ensure MaxCompute dependency table exists",
				zap.String("dependency", actualDepTable),
			)
		}
	}

	exists, err := tm.engine.CheckTableExists(ctx, project, viewName, region)
	if err != nil {
		return fmt.Errorf("failed to check view existence: %w", err)
	}
	if exists {
		tm.setCachedMetadata(cacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
		return nil
	}

	tm.mu.Lock()
	defer tm.mu.Unlock()

	exists, _ = tm.engine.CheckTableExists(ctx, project, viewName, region)
	if exists {
		tm.setCachedMetadata(cacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
		return nil
	}

	params := map[string]string{
		"Database": project,
		"Table":    viewName,
	}
	for _, depName := range template.DependsOn {
		paramKey := "Dep" + snakeToPascal(depName)
		params[paramKey] = fmt.Sprintf("%s_%s", depName, clusterId)
	}

	ddl := tm.renderDDL(template.DDL, params)
	if err := tm.engine.ExecuteStatement(ctx, ddl, region); err != nil {
		if strings.Contains(err.Error(), "already exists") ||
			strings.Contains(err.Error(), "AlreadyExists") {
			tm.setCachedMetadata(cacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
			return nil
		}
		return fmt.Errorf("execute CREATE VIEW failed: %w", err)
	}

	log.Info(ctx, "MaxCompute view created successfully",
		zap.String("project", project),
		zap.String("view", viewName),
		zap.String("region", region),
	)

	tm.setCachedMetadata(cacheKey, &TableMetadata{Exists: true, LastChecked: time.Now()})
	return nil
}

// createTable 创建 OSS Parquet 外部表
func (tm *MaxComputeTableManager) createTableIfNotExists(ctx context.Context, project, table, dataSource, region string) error {
	template, ok := tm.findTableTemplate(table)
	if !ok {
		return fmt.Errorf("no template found for table: %s", table)
	}

	location := tm.extractOSSLocation(dataSource)
	if location == "" {
		return fmt.Errorf("invalid OSS data source: %s", dataSource)
	}

	ddl := tm.renderDDL(template.DDL, map[string]string{
		"Database": project,
		"Table":    table,
		"Location": location,
	})

	return tm.engine.ExecuteStatement(ctx, ddl, region)
}

// repairPartitions 执行 MSCK REPAIR TABLE ... ADD PARTITIONS 同步分区
func (tm *MaxComputeTableManager) repairPartitions(ctx context.Context, project, table, region string) error {
	partitionKey := maxComputePartitionCacheKey(project, table, region)
	if metadata, ok := tm.getCachedPartitionMetadata(partitionKey); ok {
		if metadata.Loaded {
			return nil
		}
	}
	return tm.repairPartitionsNow(ctx, project, table, region)
}

func (tm *MaxComputeTableManager) repairPartitionsNow(ctx context.Context, project, table, region string) error {
	partitionKey := maxComputePartitionCacheKey(project, table, region)
	// MaxCompute 语法：MSCK REPAIR TABLE table_name ADD PARTITIONS
	sql := fmt.Sprintf("MSCK REPAIR TABLE `%s` ADD PARTITIONS", table)
	if err := tm.engine.ExecuteStatement(ctx, sql, region); err != nil {
		log.Warn(ctx, "MSCK REPAIR TABLE failed (partition sync may be incomplete)",
			zap.String("project", project),
			zap.String("table", table),
		)
		return err
	}

	tm.setCachedPartitionMetadata(partitionKey, &PartitionMetadata{
		Loaded:      true,
		LastChecked: time.Now(),
	})

	log.Info(ctx, "MSCK REPAIR TABLE completed",
		zap.String("project", project),
		zap.String("table", table),
	)

	return nil
}

func (tm *MaxComputeTableManager) partitionRepairDueBy(
	project, table, region string,
	checkAt time.Time,
) bool {
	if tm.cacheTTL() <= 0 {
		return false
	}
	metadata, ok := tm.getRawCachedPartitionMetadata(
		maxComputePartitionCacheKey(project, table, region),
	)
	if !ok || !metadata.Loaded {
		return true
	}
	return !checkAt.Before(metadata.LastChecked.Add(tm.cacheTTL()))
}

// extractOSSLocation 从 data_source 得到外部表 LOCATION（与 Athena extractLocation 一致：保留完整路径，含 org=/cluster=/type=/component= 等多级目录）
func (tm *MaxComputeTableManager) extractOSSLocation(dataSource string) string {
	dataSource = strings.TrimRight(dataSource, "/")
	if !strings.HasPrefix(dataSource, "oss://") {
		return ""
	}
	return dataSource + "/"
}

func (tm *MaxComputeTableManager) renderDDL(template string, params map[string]string) string {
	result := template
	for key, value := range params {
		placeholder := fmt.Sprintf("{{.%s}}", key)
		result = strings.ReplaceAll(result, placeholder, value)
	}
	return result
}

func (tm *MaxComputeTableManager) findTableTemplate(tableName string) (config.TableTemplate, bool) {
	if template, ok := tm.config.TableTemplates[tableName]; ok {
		return template, true
	}
	var bestTemplate config.TableTemplate
	bestPrefix := ""
	prefixFound := false
	for templateName, template := range tm.config.TableTemplates {
		prefix := templateName + "_"
		if strings.HasPrefix(tableName, prefix) && len(prefix) > len(bestPrefix) {
			bestTemplate = template
			bestPrefix = prefix
			prefixFound = true
		}
	}
	if prefixFound {
		return bestTemplate, true
	}
	if strings.Contains(tableName, "_") {
		parts := strings.Split(tableName, "_")
		lastPart := parts[len(parts)-1]
		isNumeric := true
		for _, c := range lastPart {
			if c < '0' || c > '9' {
				isNumeric = false
				break
			}
		}
		if isNumeric && len(parts) > 1 {
			baseTableName := strings.Join(parts[:len(parts)-1], "_")
			if template, ok := tm.config.TableTemplates[baseTableName]; ok {
				return template, true
			}
		}
	}
	return config.TableTemplate{}, false
}

func (tm *MaxComputeTableManager) getCachedMetadata(key string) (*TableMetadata, bool) {
	if value, ok := tm.cache.Load(key); ok {
		metadata := value.(*TableMetadata)
		if tm.config.CacheTTLSeconds == 0 {
			return metadata, true
		}
		ttl := time.Duration(tm.config.CacheTTLSeconds) * time.Second
		if time.Since(metadata.LastChecked) < ttl {
			return metadata, true
		}
	}
	return nil, false
}

func (tm *MaxComputeTableManager) setCachedMetadata(key string, metadata *TableMetadata) {
	tm.cache.Store(key, metadata)
}

func (tm *MaxComputeTableManager) getRawCachedMetadata(key string) (*TableMetadata, bool) {
	value, ok := tm.cache.Load(key)
	if !ok {
		return nil, false
	}
	metadata, ok := value.(*TableMetadata)
	return metadata, ok
}

func (tm *MaxComputeTableManager) getCachedPartitionMetadata(key string) (*PartitionMetadata, bool) {
	if value, ok := tm.partitionCache.Load(key); ok {
		metadata := value.(*PartitionMetadata)
		if tm.config.CacheTTLSeconds == 0 {
			return metadata, true
		}
		ttl := time.Duration(tm.config.CacheTTLSeconds) * time.Second
		if time.Since(metadata.LastChecked) < ttl {
			return metadata, true
		}
	}
	return nil, false
}

func (tm *MaxComputeTableManager) getRawCachedPartitionMetadata(key string) (*PartitionMetadata, bool) {
	value, ok := tm.partitionCache.Load(key)
	if !ok {
		return nil, false
	}
	metadata, ok := value.(*PartitionMetadata)
	return metadata, ok
}

func (tm *MaxComputeTableManager) setCachedPartitionMetadata(key string, metadata *PartitionMetadata) {
	tm.partitionCache.Store(key, metadata)
}