package engine

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

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

	"cloud.google.com/go/bigquery"
	gcs "cloud.google.com/go/storage"
	"golang.org/x/sync/singleflight"
	"google.golang.org/api/googleapi"
	"google.golang.org/api/iterator"
)

var (
	bigQueryUnrecognizedNamePattern = regexp.MustCompile(`(?i)^Unrecognized name: ([A-Za-z_][A-Za-z0-9_]*)(?:[\s;]|$)`)
	bigQueryNameNotFoundPattern     = regexp.MustCompile(`(?i)^Name ([A-Za-z_][A-Za-z0-9_]*) not found inside\b`)
	bigQueryReasonPattern           = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_]{0,63}$`)
	bigQueryQueryIDPattern          = regexp.MustCompile(`^[A-Za-z0-9_-]{1,256}$`)
)

const (
	bigQueryMessageShapeEmpty              = "empty"
	bigQueryMessageShapeUnrecognizedName   = "unrecognized_name"
	bigQueryMessageShapeNameNotFoundInside = "name_not_found_inside"
	bigQueryMessageShapePartitionFilter    = "required_partition_filter"
	bigQueryMessageShapeDatasetRequired    = "dataset_qualification_required"
	bigQueryMessageShapeDatasetNotFound    = "dataset_not_found"
	bigQueryMessageShapeLocation           = "location_mismatch"
	bigQueryMessageShapeFilePattern        = "file_pattern_expansion_failed"
	bigQueryMessageShapeOther              = "other"
	bigQueryErrorSourceBigQuery            = "bigquery"
	bigQueryErrorSourceGoogleAPI           = "googleapi"
	bigQueryErrorSourceGCS                 = "gcs"
	bigQueryErrorSourceAuth                = "auth"
	bigQueryErrorSourceNetwork             = "network"
	bigQueryErrorSourceValidation          = "validation"
	bigQueryErrorSourceUnknown             = "unknown"
)

// BigQueryQueryError retains only safe, structured diagnostics for an
// unclassified BigQuery failure. The upstream message is deliberately omitted.
type BigQueryQueryError struct {
	QueryID       string
	ErrorSource   string
	Reason        string
	ErrorCount    int
	MessageShape  string
	HasSuggestion bool
}

func (e *BigQueryQueryError) Error() string {
	return "BigQuery query failed"
}

type gcsObjectNameIterator interface {
	Next() (string, error)
}

type gcsObjectLister interface {
	List(ctx context.Context, bucket, prefix string) gcsObjectNameIterator
	ListPrefixes(ctx context.Context, bucket, prefix string) gcsObjectNameIterator
}

type gcsStorageObjectLister struct {
	client *gcs.Client
}

func (l *gcsStorageObjectLister) List(ctx context.Context, bucket, prefix string) gcsObjectNameIterator {
	return &gcsStorageObjectNameIterator{
		iterator: l.client.Bucket(bucket).Objects(ctx, &gcs.Query{Prefix: prefix}),
	}
}

func (l *gcsStorageObjectLister) ListPrefixes(ctx context.Context, bucket, prefix string) gcsObjectNameIterator {
	return &gcsStorageObjectNameIterator{
		iterator:     l.client.Bucket(bucket).Objects(ctx, &gcs.Query{Prefix: prefix, Delimiter: "/"}),
		prefixesOnly: true,
	}
}

type gcsStorageObjectNameIterator struct {
	iterator     *gcs.ObjectIterator
	prefixesOnly bool
}

func (it *gcsStorageObjectNameIterator) Next() (string, error) {
	for {
		attrs, err := it.iterator.Next()
		if err != nil {
			return "", err
		}
		if attrs == nil {
			return "", errors.New("GCS object iterator returned nil attributes")
		}
		if it.prefixesOnly {
			if attrs.Prefix == "" {
				continue
			}
			return attrs.Prefix, nil
		}
		return attrs.Name, nil
	}
}

type gcsObjectListerFactory func(context.Context) (gcsObjectLister, error)

func newGCSObjectLister(ctx context.Context) (gcsObjectLister, error) {
	client, err := gcs.NewClient(ctx)
	if err != nil {
		return nil, err
	}
	return &gcsStorageObjectLister{client: client}, nil
}

// BigQueryEngine Google BigQuery 查询引擎
type BigQueryEngine struct {
	client         *bigquery.Client
	projectID      string
	dataset        string
	maxBytesBilled int64

	gcsMu              sync.Mutex
	gcsObjectLister    gcsObjectLister
	newGCSObjectLister gcsObjectListerFactory
	topSQLSourceGroup  singleflight.Group
	topSQLSources      sync.Map // key = dataset/table/root, value = source URI fingerprint
	topSQLSetupJobs    topSQLSetupJobRunner
	topSQLNow          func() time.Time
}

// BigQueryConfig BigQuery 引擎配置
type BigQueryConfig struct {
	ProjectID      string
	Dataset        string
	MaxBytesBilled int64
	// GCP 认证通常使用 GOOGLE_APPLICATION_CREDENTIALS 环境变量
}

// NewBigQueryEngine 创建 BigQuery 引擎实例
func NewBigQueryEngine(cfg *BigQueryConfig) (*BigQueryEngine, error) {
	ctx := context.Background()

	client, err := bigquery.NewClient(ctx, cfg.ProjectID)
	if err != nil {
		return nil, fmt.Errorf("failed to create BigQuery client: %w", err)
	}

	engine := &BigQueryEngine{
		client:             client,
		projectID:          cfg.ProjectID,
		dataset:            cfg.Dataset,
		maxBytesBilled:     cfg.MaxBytesBilled,
		newGCSObjectLister: newGCSObjectLister,
		topSQLNow:          time.Now,
	}
	engine.topSQLSetupJobs = &bigQuerySetupJobRunner{client: client}
	return engine, nil
}

func (e *BigQueryEngine) getGCSObjectLister(ctx context.Context) (gcsObjectLister, error) {
	e.gcsMu.Lock()
	defer e.gcsMu.Unlock()

	if e.gcsObjectLister != nil {
		return e.gcsObjectLister, nil
	}

	factory := e.newGCSObjectLister
	if factory == nil {
		factory = newGCSObjectLister
	}
	lister, err := factory(ctx)
	if err != nil {
		return nil, err
	}
	e.gcsObjectLister = lister
	return lister, nil
}

// HasDeltaLog reports whether a GCS data root contains a Delta transaction
// log. A successfully listed, empty root returns false. A non-empty root with
// no transaction log is malformed and returns ErrInvalidDeltaSource. Storage
// errors are returned unchanged so permission and timeout failures can never be
// mistaken for an optional data source that has not produced data yet.
func (e *BigQueryEngine) HasDeltaLog(ctx context.Context, dataSource *dto.DataSourceInfo) (bool, error) {
	if dataSource == nil {
		return false, errors.New("GCS data source is required")
	}
	if dataSource.Bucket == "" {
		return false, errors.New("GCS bucket is required")
	}

	lister, err := e.getGCSObjectLister(ctx)
	if err != nil {
		return false, err
	}

	rootPrefix := strings.Trim(dataSource.Path, "/")
	if rootPrefix != "" {
		rootPrefix += "/"
	}
	deltaLogPrefix := rootPrefix + "_delta_log/"
	available, err := inspectGCSDeltaRoot(
		lister.List(ctx, dataSource.Bucket, deltaLogPrefix),
		lister.List(ctx, dataSource.Bucket, rootPrefix),
		deltaLogPrefix,
	)
	if errors.Is(err, ErrInvalidDeltaSource) {
		return false, fmt.Errorf("%w at gs://%s/%s", ErrInvalidDeltaSource, dataSource.Bucket, strings.TrimSuffix(rootPrefix, "/"))
	}
	return available, err
}

func inspectGCSDeltaRoot(deltaObjects, rootObjects gcsObjectNameIterator, deltaLogPrefix string) (bool, error) {
	hasDeltaObjects := false
	for {
		name, err := deltaObjects.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return false, err
		}
		if !strings.HasPrefix(name, deltaLogPrefix) || isGCSDirectoryMarker(name) {
			continue
		}
		hasDeltaObjects = true
		if isDeltaTransactionLogObject(strings.TrimPrefix(name, deltaLogPrefix)) {
			return true, nil
		}
	}
	if hasDeltaObjects {
		return false, ErrInvalidDeltaSource
	}

	for {
		name, err := rootObjects.Next()
		if err == iterator.Done {
			return false, nil
		}
		if err != nil {
			return false, err
		}
		if isGCSDirectoryMarker(name) {
			continue
		}
		return false, ErrInvalidDeltaSource
	}
}

func isGCSDirectoryMarker(name string) bool {
	return name == "" || strings.HasSuffix(name, "/")
}

func isDeltaTransactionLogObject(relativeName string) bool {
	if relativeName == "_last_checkpoint" {
		return true
	}
	if strings.Contains(relativeName, "/") || len(relativeName) <= 20 {
		return false
	}
	for _, c := range relativeName[:20] {
		if c < '0' || c > '9' {
			return false
		}
	}

	suffix := relativeName[20:]
	return suffix == ".json" ||
		(strings.HasPrefix(suffix, ".checkpoint") &&
			(strings.HasSuffix(suffix, ".parquet") || strings.HasSuffix(suffix, ".json")))
}

// Name 返回引擎名称
func (e *BigQueryEngine) Name() string {
	return "bigquery"
}

// Execute 执行查询
func (e *BigQueryEngine) Execute(ctx context.Context, sql string, dataSource *dto.DataSourceInfo) (*dto.DataProxyQueryResult, error) {
	startTime := time.Now()

	// 创建查询
	q := e.client.Query(sql)
	q.UseStandardSQL = true
	q.MaxBytesBilled = e.maxBytesBilled
	e.configureQueryRouting(q, dataSource)

	// 执行查询
	job, err := q.Run(ctx)
	if err != nil {
		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
			return nil, fmt.Errorf("query run failed: %w", err)
		}
		if unknownErr := classifyBigQueryUnknownField(err, ""); unknownErr != nil {
			return nil, unknownErr
		}
		return nil, fmt.Errorf("query run failed: %w", newBigQueryQueryError("", err, nil))
	}

	// 等待完成
	status, err := job.Wait(ctx)
	if err != nil {
		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
			return nil, fmt.Errorf("query wait failed: %w", err)
		}
		if unknownErr := classifyBigQueryWaitUnknownField(err, job.ID()); unknownErr != nil {
			return nil, unknownErr
		}
		return nil, fmt.Errorf("query wait failed: %w", newBigQueryQueryError(job.ID(), err, nil))
	}

	if err := status.Err(); err != nil {
		if unknownErr := classifyBigQueryStatusUnknownField(err, status.Errors, job.ID()); unknownErr != nil {
			return nil, unknownErr
		}
		return nil, fmt.Errorf("query execution failed: %w", newBigQueryQueryError(job.ID(), err, status.Errors))
	}

	// 获取结果
	it, err := job.Read(ctx)
	if err != nil {
		return nil, fmt.Errorf("read results failed: %w", err)
	}

	elapsed := time.Since(startTime)
	result := &dto.DataProxyQueryResult{
		Metadata: dto.DataProxyResultMetadata{
			QueryID:             job.ID(),
			Engine:              e.Name(),
			Vendor:              dataSource.Vendor,
			Region:              dataSource.Region,
			ExecutionTime:       dto.FormattedDuration(elapsed),
			EngineExecutionTime: dto.FormattedDuration(status.Statistics.EndTime.Sub(status.Statistics.StartTime)),
			BytesScanned:        int64(status.Statistics.TotalBytesProcessed),
		},
	}

	// 读取列名
	if it.Schema != nil {
		for _, field := range it.Schema {
			result.Columns = append(result.Columns, field.Name)
		}
	}

	// 读取数据行
	for {
		var row []bigquery.Value
		err := it.Next(&row)
		if err == iterator.Done {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("iterate results failed: %w", err)
		}

		rowData := make([]interface{}, len(row))
		for i, v := range row {
			rowData[i] = v
		}
		result.Rows = append(result.Rows, rowData)
	}

	result.Metadata.RowCount = int64(len(result.Rows))
	return result, nil
}

func (e *BigQueryEngine) configureQueryRouting(q *bigquery.Query, dataSource *dto.DataSourceInfo) {
	if q == nil {
		return
	}
	region := ""
	if dataSource != nil {
		region = strings.TrimSpace(dataSource.Region)
	}
	dataset := e.GetDatabase(region)
	if dataset != "" {
		q.DefaultProjectID = e.projectID
		q.DefaultDatasetID = dataset
	}
	if region != "" {
		q.Location = region
	}
}

func classifyBigQueryUnknownField(err error, queryID string) error {
	fields := collectBigQueryUnknownFields(err)
	return newUnknownFieldError("bigquery", queryID, fields)
}

func classifyBigQueryStatusUnknownField(primary error, supplemental []*bigquery.Error, queryID string) error {
	fields := collectBigQueryUnknownFields(primary)
	if len(fields) == 0 {
		return nil
	}
	for _, item := range supplemental {
		fields = append(fields, collectBigQueryUnknownFields(item)...)
	}
	return newUnknownFieldError("bigquery", queryID, fields)
}

func classifyBigQueryWaitUnknownField(err error, queryID string) error {
	var apiErr *googleapi.Error
	if !errors.As(err, &apiErr) || apiErr == nil || apiErr.Code != 400 || len(apiErr.Errors) == 0 {
		return nil
	}
	var fields []string
	for _, item := range apiErr.Errors {
		if item.Reason != "invalidQuery" {
			return nil
		}
		field := bigQueryUnknownFieldFromMessage(item.Message)
		if field == "" {
			return nil
		}
		fields = append(fields, field)
	}
	return newUnknownFieldError("bigquery", queryID, fields)
}

func collectBigQueryUnknownFields(err error) []string {
	var fields []string
	var multi bigquery.MultiError
	if errors.As(err, &multi) {
		for _, item := range multi {
			fields = append(fields, bigQueryUnknownField(item)...)
		}
	} else {
		fields = append(fields, bigQueryUnknownField(err)...)
	}
	return fields
}

func bigQueryUnknownField(err error) []string {
	var queryErr *bigquery.Error
	if !errors.As(err, &queryErr) || queryErr == nil || queryErr.Reason != "invalidQuery" {
		return nil
	}
	field := bigQueryUnknownFieldFromMessage(queryErr.Message)
	if field != "" {
		return []string{field}
	}
	return nil
}

func bigQueryUnknownFieldFromMessage(message string) string {
	for _, pattern := range []*regexp.Regexp{bigQueryUnrecognizedNamePattern, bigQueryNameNotFoundPattern} {
		match := pattern.FindStringSubmatch(message)
		if len(match) == 2 && isSafeSQLIdentifier(match[1]) {
			return match[1]
		}
	}
	return ""
}

func newBigQueryQueryError(queryID string, primary error, supplemental []*bigquery.Error) *BigQueryQueryError {
	errorSource, reason, message, errorCount := bigQueryErrorDetails(primary, supplemental)
	if !bigQueryReasonPattern.MatchString(reason) {
		reason = "unknown"
	}
	return &BigQueryQueryError{
		QueryID:       queryID,
		ErrorSource:   errorSource,
		Reason:        reason,
		ErrorCount:    errorCount,
		MessageShape:  bigQueryMessageShape(message),
		HasSuggestion: strings.Contains(strings.ToLower(message), "did you mean"),
	}
}

func bigQueryErrorDetails(err error, supplemental []*bigquery.Error) (string, string, string, int) {
	var queryErr *bigquery.Error
	if errors.As(err, &queryErr) && queryErr != nil {
		return bigQueryErrorSourceBigQuery, queryErr.Reason, queryErr.Message, max(1, len(supplemental))
	}
	var multi bigquery.MultiError
	if errors.As(err, &multi) {
		for _, item := range multi {
			if errors.As(item, &queryErr) && queryErr != nil {
				return bigQueryErrorSourceBigQuery, queryErr.Reason, queryErr.Message, len(multi)
			}
		}
	}
	var apiErr *googleapi.Error
	if errors.As(err, &apiErr) && apiErr != nil {
		if len(apiErr.Errors) > 0 {
			return bigQueryErrorSourceGoogleAPI, apiErr.Errors[0].Reason, apiErr.Errors[0].Message, len(apiErr.Errors)
		}
		return bigQueryErrorSourceGoogleAPI, "", "", 1
	}
	if err != nil {
		return bigQueryErrorSourceUnknown, "", "", 1
	}
	return bigQueryErrorSourceUnknown, "", "", 0
}

func bigQueryMessageShape(message string) string {
	lowerMessage := strings.ToLower(message)
	switch {
	case message == "":
		return bigQueryMessageShapeEmpty
	case bigQueryUnrecognizedNamePattern.MatchString(message):
		return bigQueryMessageShapeUnrecognizedName
	case bigQueryNameNotFoundPattern.MatchString(message):
		return bigQueryMessageShapeNameNotFoundInside
	case strings.Contains(lowerMessage, "without a filter over column") ||
		(strings.Contains(lowerMessage, "partition") && strings.Contains(lowerMessage, "filter")):
		return bigQueryMessageShapePartitionFilter
	case strings.Contains(lowerMessage, "must be qualified with a dataset") || strings.Contains(lowerMessage, "default dataset is not set"):
		return bigQueryMessageShapeDatasetRequired
	case strings.Contains(lowerMessage, "dataset") && strings.Contains(lowerMessage, "not found"):
		return bigQueryMessageShapeDatasetNotFound
	case strings.Contains(lowerMessage, "location"):
		return bigQueryMessageShapeLocation
	case strings.Contains(lowerMessage, "failed to expand table") && strings.Contains(lowerMessage, "file pattern"):
		return bigQueryMessageShapeFilePattern
	default:
		return bigQueryMessageShapeOther
	}
}

func normalizeBigQueryErrorSource(source string) string {
	switch source {
	case bigQueryErrorSourceBigQuery,
		bigQueryErrorSourceGoogleAPI,
		bigQueryErrorSourceGCS,
		bigQueryErrorSourceAuth,
		bigQueryErrorSourceNetwork,
		bigQueryErrorSourceValidation:
		return source
	default:
		return bigQueryErrorSourceUnknown
	}
}

func normalizeBigQueryMessageShape(shape string) string {
	switch shape {
	case bigQueryMessageShapeEmpty,
		bigQueryMessageShapeUnrecognizedName,
		bigQueryMessageShapeNameNotFoundInside,
		bigQueryMessageShapePartitionFilter,
		bigQueryMessageShapeDatasetRequired,
		bigQueryMessageShapeDatasetNotFound,
		bigQueryMessageShapeLocation,
		bigQueryMessageShapeFilePattern,
		bigQueryMessageShapeOther:
		return shape
	default:
		return bigQueryMessageShapeOther
	}
}

// GetDatabase 根据 region 返回 BigQuery dataset 名（prefix_region 格式，region 为空时仅 prefix）
func (e *BigQueryEngine) GetDatabase(region string) string {
	if region == "" {
		return e.dataset
	}
	return e.dataset + "_" + strings.ReplaceAll(region, "-", "_")
}

// Health 健康检查
func (e *BigQueryEngine) Health(ctx context.Context) error {
	// 简单查询测试连接
	q := e.client.Query("SELECT 1")
	_, err := q.Run(ctx)
	if err != nil {
		return fmt.Errorf("bigquery health check failed: %w", err)
	}
	return nil
}

// EnsureDatasetExists 确保 dataset 存在，不存在则创建
// region 为 BigQuery Location，如 asia-east1
func (e *BigQueryEngine) EnsureDatasetExists(ctx context.Context, datasetID, region string) error {
	ds := e.client.Dataset(datasetID)
	_, err := ds.Metadata(ctx)
	if err == nil {
		return nil // 已存在
	}
	if !strings.Contains(err.Error(), "notFound") && !strings.Contains(err.Error(), "Not found") {
		return fmt.Errorf("check dataset failed: %w", err)
	}

	meta := &bigquery.DatasetMetadata{
		Location: region,
	}
	if err := ds.Create(ctx, meta); err != nil {
		if strings.Contains(err.Error(), "Already Exists") || strings.Contains(err.Error(), "already exists") {
			return nil
		}
		return fmt.Errorf("create dataset failed: %w", err)
	}
	return nil
}

// CheckTableExists 检查表是否存在
func (e *BigQueryEngine) CheckTableExists(ctx context.Context, dataset, table, region string) (bool, error) {
	tbl := e.client.Dataset(dataset).Table(table)
	_, err := tbl.Metadata(ctx)
	if err != nil {
		if strings.Contains(err.Error(), "notFound") || strings.Contains(err.Error(), "Not found") {
			return false, nil
		}
		return false, fmt.Errorf("check table exists failed: %w", err)
	}
	return true, nil
}

// GetTableSchema 获取表结构
func (e *BigQueryEngine) GetTableSchema(ctx context.Context, dataset, table, _ string) (*dto.SchemaResponse, error) {
	tbl := e.client.Dataset(dataset).Table(table)
	meta, err := tbl.Metadata(ctx)
	if err != nil {
		return nil, fmt.Errorf("get table metadata failed: %w", err)
	}

	resp := &dto.SchemaResponse{
		Database: dataset,
		Table:    table,
	}

	if meta.Schema != nil {
		for _, f := range meta.Schema {
			col := dto.ColumnInfo{
				Name: f.Name,
				Type: string(f.Type), // FieldType 是 string 的 type alias
			}
			if f.Description != "" {
				col.Comment = f.Description
			}
			resp.Columns = append(resp.Columns, col)
		}
	}

	// 从 INFORMATION_SCHEMA 获取分区列（Delta Lake / Hive 分区）
	partitions, err := e.getPartitionColumns(ctx, dataset, table)
	if err != nil {
		return nil, err
	}
	resp.Partitions = partitions

	// 删除分区列
	resp.Columns = slices.DeleteFunc(resp.Columns, func(col dto.ColumnInfo) bool {
		return slices.ContainsFunc(partitions, func(p dto.PartitionDef) bool {
			return p.Name == col.Name
		})
	})

	return resp, nil
}

// getPartitionColumns 从 INFORMATION_SCHEMA.COLUMNS 查询分区列
func (e *BigQueryEngine) getPartitionColumns(ctx context.Context, dataset, table string) ([]dto.PartitionDef, error) {
	sql := fmt.Sprintf(
		"SELECT column_name, data_type FROM `%s.%s.INFORMATION_SCHEMA.COLUMNS` WHERE table_name = '%s' AND is_partitioning_column = 'YES'",
		e.projectID, dataset, table,
	)
	q := e.client.Query(sql)
	q.UseStandardSQL = true
	job, err := q.Run(ctx)
	if err != nil {
		return nil, err
	}
	status, err := job.Wait(ctx)
	if err != nil {
		return nil, err
	}
	if status.Err() != nil {
		return nil, status.Err()
	}
	it, err := job.Read(ctx)
	if err != nil {
		return nil, err
	}

	var partitions []dto.PartitionDef
	for {
		var row []bigquery.Value
		if err := it.Next(&row); err == iterator.Done {
			break
		} else if err != nil {
			return nil, err
		}
		if len(row) >= 2 {
			name, _ := row[0].(string)
			typ, _ := row[1].(string)
			if name != "" {
				partitions = append(partitions, dto.PartitionDef{Name: name, Type: typ})
			}
		}
	}
	return partitions, nil
}

// DeltaLakeFormat BigQuery Delta Lake 格式（SDK 暂未导出常量）
const DeltaLakeFormat bigquery.DataFormat = "DELTA_LAKE"

// CreateExternalTable 创建外部表（使用 Delta Lake 格式，兼容 _delta_log）
func (e *BigQueryEngine) CreateExternalTable(ctx context.Context, dataset, table, gcsURI string) error {
	baseURI := strings.TrimRight(gcsURI, "/")

	extConfig := &bigquery.ExternalDataConfig{
		SourceFormat: DeltaLakeFormat,
		SourceURIs:   []string{baseURI},
		AutoDetect:   true,
	}

	meta := &bigquery.TableMetadata{
		ExternalDataConfig:     extConfig,
		RequirePartitionFilter: true,
	}

	tbl := e.client.Dataset(dataset).Table(table)
	if err := tbl.Create(ctx, meta); err != nil {
		if strings.Contains(err.Error(), "Already Exists") || strings.Contains(err.Error(), "already exists") {
			return nil
		}
		return fmt.Errorf("create external table failed: %w", err)
	}

	return nil
}

// RefreshExternalTableSchema asks BigQuery to re-autodetect the schema of an
// existing Delta Lake external table. The current external configuration and
// ETag are preserved to avoid overwriting concurrent metadata changes.
func (e *BigQueryEngine) RefreshExternalTableSchema(
	ctx context.Context,
	dataset, table, gcsURI string,
) (bool, error) {
	tbl := e.client.Dataset(dataset).Table(table)
	meta, err := tbl.Metadata(ctx)
	if err != nil {
		return false, fmt.Errorf("get table metadata failed: %w", err)
	}
	if meta.ExternalDataConfig == nil ||
		meta.ExternalDataConfig.SourceFormat != DeltaLakeFormat {
		return false, nil
	}
	if !sameGCSDeltaRoot(meta.ExternalDataConfig.SourceURIs, gcsURI) {
		return false, nil
	}

	configCopy := externalDataConfigForSchemaRefresh(meta.ExternalDataConfig)
	_, err = tbl.Update(ctx, bigquery.TableMetadataToUpdate{
		ExternalDataConfig: configCopy,
	}, meta.ETag, bigquery.WithAutoDetectSchema(true))
	if err != nil {
		return false, fmt.Errorf("refresh external table schema failed: %w", err)
	}
	return true, nil
}

func externalDataConfigForSchemaRefresh(current *bigquery.ExternalDataConfig) *bigquery.ExternalDataConfig {
	configCopy := *current
	configCopy.SourceURIs = append([]string(nil), current.SourceURIs...)
	configCopy.AutoDetect = true
	// The inferred schema itself must not be sent back in the patch; otherwise
	// it can override the autodetection we are asking BigQuery to perform.
	configCopy.Schema = nil
	return &configCopy
}

func sameGCSDeltaRoot(sourceURIs []string, requestedURI string) bool {
	if len(sourceURIs) != 1 {
		return false
	}
	return strings.TrimRight(sourceURIs[0], "/") ==
		strings.TrimRight(requestedURI, "/")
}