# Clinic Data Proxy 代码架构设计

## 一、概述

### 1.1 设计目标

Clinic Data Proxy 是一个全局服务（Global Service），旨在为 Clinic 系统提供统一的数据访问能力。核心目标包括：

1. **统一查询入口**：提供标准化的 API 接口，屏蔽底层多云、多 Region、多数据源的复杂性
2. **智能路由**：根据数据源 URI 自动识别云厂商、Region，并选择合适的查询引擎
3. **SQL 方言适配**：自动将逻辑 SQL 转换为目标引擎的特定方言（Presto、BigQuery 等）
4. **结果聚合**：统一返回格式，支持跨 Region 数据查询与聚合
5. **高性能**：无状态设计，不落地存储数据，仅承担查询代理与聚合职责

### 1.2 架构定位

在 Clinic 架构中的定位：
- **层级**：全局控制层（Global Control Plane）
- **职责**：查询路由 + 结果聚合（不承担实际查询执行）
- **上游**：Clinic 业务模块（巡检、诊断、告警分析等）
- **下游**：Regional Data Query Engine（或直接对接云查询引擎）

## 二、总体架构

### 2.1 架构图

```
┌─────────────────────────────────────────────────────────┐
│              Clinic Business Modules                    │
│   (Health Check, Diagnosis, Alert Analysis, etc.)       │
└────────────────────┬────────────────────────────────────┘
                     │ Unified Query API
                     ▼
┌─────────────────────────────────────────────────────────┐
│           Clinic Data Proxy (Global Service)            │
│  ┌─────────────┐  ┌──────────────┐  ┌────────────────┐ │
│  │   Router    │→ │   Dialect    │→ │    Engine      │ │
│  │  (识别云)    │  │  (SQL转换)    │  │   (查询执行)    │ │
│  └─────────────┘  └──────────────┘  └────────────────┘ │
└────────────────────┬────────────────────────────────────┘
                     │ Region-specific Queries
          ┌──────────┼──────────┬────────────┐
          ▼          ▼          ▼            ▼
     ┌────────┐ ┌────────┐ ┌────────┐  ┌─────────┐
     │ Athena │ │BigQuery│ │  DLA   │  │ DuckDB  │
     │  (AWS) │ │ (GCP)  │ │(AliYun)│  │(Direct) │
     └────────┘ └────────┘ └────────┘  └─────────┘
          │          │          │            │
          ▼          ▼          ▼            ▼
     ┌────────┐ ┌────────┐ ┌────────┐  ┌─────────┐
     │   S3   │ │   GCS  │ │  OSS   │  │   S3    │
     └────────┘ └────────┘ └────────┘  └─────────┘
```

### 2.2 核心流程

```
1. 业务请求 → Data Proxy
   ├─ QueryRequest {SQL, DataSource, Options}

2. Router 解析 DataSource
   ├─ s3://bucket/path     → AWS + Region
   ├─ gs://bucket/path     → GCP + Region
   └─ oss://bucket/path    → AliCloud + Region

3. Dialect 转换 SQL
   ├─ AWS/AliCloud → Presto Dialect
   ├─ GCP          → BigQuery Dialect
   └─ Direct       → DuckDB Dialect

4. Engine 执行查询
   ├─ AWS      → Athena
   ├─ GCP      → BigQuery
   ├─ AliCloud → DLA
   └─ Direct   → DuckDB

5. 结果聚合并返回
   └─ QueryResult {Columns, Rows, Metadata}
```

## 三、模块设计

### 3.1 目录结构

```
pkg/dataproxy/
├── model/
│   ├── query.go           # 查询请求/响应模型
│   ├── datasource.go      # 数据源模型
│   └── error.go           # 错误定义
├── dialect/
│   ├── dialect.go         # 方言接口
│   ├── presto.go          # Presto 方言（Athena、DLA）
│   ├── bigquery.go        # BigQuery 方言
│   └── duckdb.go          # DuckDB 方言
├── engine/
│   ├── engine.go          # 查询引擎接口
│   ├── athena.go          # AWS Athena 引擎
│   ├── bigquery.go        # GCP BigQuery 引擎
│   ├── dla.go             # AliCloud DLA 引擎
│   └── duckdb.go          # DuckDB 引擎（直接查询）
├── router/
│   ├── router.go          # 路由器（选择引擎）
│   ├── datasource.go      # 数据源解析
│   └── region.go          # Region 识别
├── service/
│   └── dataproxy_service.go  # Data Proxy 核心服务
├── handler/
│   └── dataproxy_handler.go  # HTTP 处理器
└── config/
    └── dataproxy_config.go   # 配置管理
```

### 3.2 模块职责

| 模块 | 职责 | 输入 | 输出 |
|------|------|------|------|
| **Model** | 定义统一的数据模型 | - | QueryRequest/QueryResult |
| **Router** | 解析数据源，选择引擎 | DataSource URI | Vendor + Region + Engine |
| **Dialect** | SQL 方言转换 | 逻辑 SQL | 引擎特定 SQL |
| **Engine** | 执行查询 | SQL + 认证信息 | QueryResult |
| **Service** | 编排整体流程 | QueryRequest | QueryResult |
| **Handler** | HTTP 接口 | HTTP Request | HTTP Response |

## 四、详细设计

### 4.1 数据模型 (model/)

#### 4.1.1 查询请求模型

```go
// QueryRequest 统一查询请求
type QueryRequest struct {
    SQL        string        `json:"sql"`         // 逻辑 SQL（方言无关）
    DataSource string        `json:"data_source"` // 数据源 URI (s3://, gs://, oss://)
    Timeout    time.Duration `json:"timeout"`     // 查询超时时间
    Options    QueryOptions  `json:"options"`     // 查询选项
}

// QueryOptions 查询选项
type QueryOptions struct {
    MaxRows       int               `json:"max_rows"`        // 最大返回行数
    UseCache      bool              `json:"use_cache"`       // 是否使用缓存
    OutputFormat  string            `json:"output_format"`   // 输出格式 (json, csv)
    Authentication *AuthConfig      `json:"authentication"`  // 认证配置
    Region        string            `json:"region"`          // 强制指定 Region（可选）
    Vendor        string            `json:"vendor"`          // 强制指定云厂商（可选）
}

// AuthConfig 认证配置
type AuthConfig struct {
    Type         string            `json:"type"`          // 认证类型 (iam, explicit)
    AccessKey    string            `json:"access_key"`    // 访问密钥
    SecretKey    string            `json:"secret_key"`    // 密钥
    SessionToken string            `json:"session_token"` // 会话令牌
    ProjectID    string            `json:"project_id"`    // GCP 项目 ID
}
```

#### 4.1.2 查询响应模型

```go
// QueryResult 统一查询结果
type QueryResult struct {
    Columns  []string        `json:"columns"`   // 列名
    Rows     [][]interface{} `json:"rows"`      // 数据行
    Metadata ResultMetadata  `json:"metadata"`  // 元数据
}

// ResultMetadata 结果元数据
type ResultMetadata struct {
    RowCount      int64         `json:"row_count"`      // 行数
    BytesScanned  int64         `json:"bytes_scanned"`  // 扫描字节数
    ExecutionTime time.Duration `json:"execution_time"` // 执行时间
    QueryID       string        `json:"query_id"`       // 查询 ID
    Engine        string        `json:"engine"`         // 使用的引擎
    Vendor        string        `json:"vendor"`         // 云厂商
    Region        string        `json:"region"`         // Region
}
```

#### 4.1.3 数据源模型

```go
// DataSourceInfo 数据源信息
type DataSourceInfo struct {
    Scheme   string // s3, gs, oss
    Vendor   string // aws, gcp, alicloud
    Region   string // us-east-1, asia-northeast1, ap-southeast-1
    Bucket   string // bucket 名称
    Path     string // 对象路径
    FullURI  string // 完整 URI
}

// ParseDataSource 解析数据源 URI
func ParseDataSource(uri string) (*DataSourceInfo, error)
```

### 4.2 路由层 (router/)

#### 4.2.1 数据源解析器

```go
// router/datasource.go

// ParseDataSource 解析数据源 URI，识别云厂商和 Region
func ParseDataSource(uri string) (*model.DataSourceInfo, error) {
    u, err := url.Parse(uri)
    if err != nil {
        return nil, fmt.Errorf("invalid URI: %w", err)
    }

    info := &model.DataSourceInfo{
        Scheme:  u.Scheme,
        Bucket:  u.Host,
        Path:    strings.TrimPrefix(u.Path, "/"),
        FullURI: uri,
    }

    // 识别云厂商
    switch u.Scheme {
    case "s3":
        info.Vendor = constant.ProviderAWS
        info.Region = DetectAWSRegion(u.Host)
    case "gs":
        info.Vendor = constant.ProviderGCP
        info.Region = DetectGCPRegion(u.Host)
    case "oss":
        info.Vendor = constant.ProviderAliCloud
        info.Region = DetectAliyunRegion(u.Host)
    default:
        return nil, fmt.Errorf("unsupported scheme: %s", u.Scheme)
    }

    return info, nil
}

// DetectAWSRegion 从 S3 Bucket 名称推断 AWS Region
// 可选策略：
// 1. 从 bucket 命名规则推断（如 bucket-us-east-1）
// 2. 调用 S3 GetBucketLocation API
// 3. 从配置文件读取映射关系
func DetectAWSRegion(bucket string) string {
    // 实现 Region 检测逻辑
    return "us-east-1" // 默认值
}
```

#### 4.2.2 路由器

```go
// router/router.go

// Router 查询路由器，负责选择合适的查询引擎
type Router struct {
    engines map[string]engine.QueryEngine
    config  *config.DataProxyConfig
}

// NewRouter 创建路由器
func NewRouter(cfg *config.DataProxyConfig) (*Router, error) {
    r := &Router{
        engines: make(map[string]engine.QueryEngine),
        config:  cfg,
    }

    // 初始化各云厂商的查询引擎
    if cfg.AWS.Enabled {
        athena, err := engine.NewAthenaEngine(&cfg.AWS)
        if err != nil {
            return nil, err
        }
        r.engines[constant.ProviderAWS] = athena
    }

    if cfg.GCP.Enabled {
        bq, err := engine.NewBigQueryEngine(&cfg.GCP)
        if err != nil {
            return nil, err
        }
        r.engines[constant.ProviderGCP] = bq
    }

    if cfg.AliCloud.Enabled {
        dla, err := engine.NewDLAEngine(&cfg.AliCloud)
        if err != nil {
            return nil, err
        }
        r.engines[constant.ProviderAliCloud] = dla
    }

    // DuckDB 作为 fallback 引擎
    duckdb, err := engine.NewDuckDBEngine(&cfg.DuckDB)
    if err != nil {
        return nil, err
    }
    r.engines["duckdb"] = duckdb

    return r, nil
}

// Route 根据云厂商选择查询引擎
func (r *Router) Route(vendor string) (engine.QueryEngine, error) {
    eng, ok := r.engines[vendor]
    if !ok {
        return nil, fmt.Errorf("no engine available for vendor: %s", vendor)
    }
    return eng, nil
}

// GetDialect 根据云厂商获取 SQL 方言
func (r *Router) GetDialect(vendor string) dialect.Dialect {
    switch vendor {
    case constant.ProviderAWS, constant.ProviderAliCloud:
        return dialect.NewPrestoDialect()
    case constant.ProviderGCP:
        return dialect.NewBigQueryDialect()
    default:
        return dialect.NewDuckDBDialect()
    }
}
```

### 4.3 SQL 方言适配层 (dialect/)

#### 4.3.1 方言接口

```go
// dialect/dialect.go

// Dialect SQL 方言接口，负责将逻辑 SQL 转换为特定引擎的 SQL
type Dialect interface {
    // Rewrite 重写 SQL
    Rewrite(logicalSQL string) (string, error)

    // Name 方言名称
    Name() string

    // Validate 验证 SQL 是否符合方言语法
    Validate(sql string) error
}
```

#### 4.3.2 Presto 方言（Athena、DLA）

```go
// dialect/presto.go

type PrestoDialect struct{}

func NewPrestoDialect() *PrestoDialect {
    return &PrestoDialect{}
}

func (d *PrestoDialect) Name() string {
    return "presto"
}

func (d *PrestoDialect) Rewrite(sql string) (string, error) {
    // 函数名映射
    replacements := map[string]string{
        "json_get(":      "json_extract(",
        "to_timestamp(":  "from_unixtime(",
        "regex_match(":   "regexp_like(",
        "date_format(":   "date_format(",
        "current_time()": "now()",
    }

    result := sql
    for old, new := range replacements {
        result = strings.ReplaceAll(result, old, new)
    }

    // 处理 LIMIT 子句（Presto 支持）
    // 处理 DATE/TIMESTAMP 类型转换

    return result, nil
}

func (d *PrestoDialect) Validate(sql string) error {
    // 验证 SQL 基本语法
    sql = strings.TrimSpace(strings.ToUpper(sql))
    if !strings.HasPrefix(sql, "SELECT") &&
       !strings.HasPrefix(sql, "WITH") {
        return fmt.Errorf("only SELECT/WITH queries are supported")
    }
    return nil
}
```

#### 4.3.3 BigQuery 方言

```go
// dialect/bigquery.go

type BigQueryDialect struct{}

func NewBigQueryDialect() *BigQueryDialect {
    return &BigQueryDialect{}
}

func (d *BigQueryDialect) Name() string {
    return "bigquery"
}

func (d *BigQueryDialect) Rewrite(sql string) (string, error) {
    // 函数名映射
    replacements := map[string]string{
        "json_get(":      "JSON_VALUE(",
        "json_extract(":  "JSON_EXTRACT(",
        "to_timestamp(":  "TIMESTAMP_SECONDS(",
        "regex_match(":   "REGEXP_CONTAINS(",
        "date_format(":   "FORMAT_TIMESTAMP(",
        "from_unixtime(": "TIMESTAMP_SECONDS(",
    }

    result := sql
    for old, new := range replacements {
        result = strings.ReplaceAll(result, old, new)
    }

    // BigQuery 特殊处理
    // 1. 时间函数转换
    // 2. 字符串连接运算符 (|| → CONCAT)
    // 3. LIMIT 子句

    return result, nil
}

func (d *BigQueryDialect) Validate(sql string) error {
    sql = strings.TrimSpace(strings.ToUpper(sql))
    if !strings.HasPrefix(sql, "SELECT") &&
       !strings.HasPrefix(sql, "WITH") {
        return fmt.Errorf("only SELECT/WITH queries are supported")
    }
    return nil
}
```

#### 4.3.4 DuckDB 方言

```go
// dialect/duckdb.go

type DuckDBDialect struct{}

func NewDuckDBDialect() *DuckDBDialect {
    return &DuckDBDialect{}
}

func (d *DuckDBDialect) Name() string {
    return "duckdb"
}

func (d *DuckDBDialect) Rewrite(sql string) (string, error) {
    // DuckDB 支持大部分标准 SQL，无需大量转换
    // 主要处理：
    // 1. S3 路径处理 (s3://bucket/path)
    // 2. Parquet/JSON 格式读取

    return sql, nil
}

func (d *DuckDBDialect) Validate(sql string) error {
    sql = strings.TrimSpace(strings.ToUpper(sql))
    if !strings.HasPrefix(sql, "SELECT") &&
       !strings.HasPrefix(sql, "WITH") {
        return fmt.Errorf("only SELECT/WITH queries are supported")
    }
    return nil
}
```

### 4.4 查询引擎层 (engine/)

#### 4.4.1 引擎接口

```go
// engine/engine.go

// QueryEngine 查询引擎接口
type QueryEngine interface {
    // Execute 执行查询
    Execute(ctx context.Context, sql string, dataSource *model.DataSourceInfo) (*model.QueryResult, error)

    // Name 引擎名称
    Name() string

    // Health 健康检查
    Health(ctx context.Context) error
}
```

#### 4.4.2 Athena 引擎

```go
// engine/athena.go

type AthenaEngine struct {
    client      *athena.Client
    s3OutputLoc string // 查询结果输出位置
    database    string // 默认数据库
    workgroup   string // 工作组
}

func NewAthenaEngine(cfg *config.AWSConfig) (*AthenaEngine, error) {
    // 初始化 AWS SDK
    awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(),
        awsconfig.WithRegion(cfg.Region),
    )
    if err != nil {
        return nil, err
    }

    return &AthenaEngine{
        client:      athena.NewFromConfig(awsCfg),
        s3OutputLoc: cfg.AthenaOutputLocation,
        database:    cfg.AthenaDatabase,
        workgroup:   cfg.AthenaWorkgroup,
    }, nil
}

func (e *AthenaEngine) Name() string {
    return "athena"
}

func (e *AthenaEngine) Execute(ctx context.Context, sql string, dataSource *model.DataSourceInfo) (*model.QueryResult, error) {
    // 1. 启动查询
    startInput := &athena.StartQueryExecutionInput{
        QueryString: aws.String(sql),
        QueryExecutionContext: &types.QueryExecutionContext{
            Database: aws.String(e.database),
        },
        ResultConfiguration: &types.ResultConfiguration{
            OutputLocation: aws.String(e.s3OutputLoc),
        },
        WorkGroup: aws.String(e.workgroup),
    }

    startOutput, err := e.client.StartQueryExecution(ctx, startInput)
    if err != nil {
        return nil, fmt.Errorf("start query failed: %w", err)
    }

    queryID := *startOutput.QueryExecutionId

    // 2. 轮询查询状态
    if err := e.waitForQuery(ctx, queryID); err != nil {
        return nil, err
    }

    // 3. 获取查询结果
    result, err := e.fetchResults(ctx, queryID)
    if err != nil {
        return nil, err
    }

    return result, nil
}

func (e *AthenaEngine) waitForQuery(ctx context.Context, queryID string) error {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-ticker.C:
            getInput := &athena.GetQueryExecutionInput{
                QueryExecutionId: aws.String(queryID),
            }

            output, err := e.client.GetQueryExecution(ctx, getInput)
            if err != nil {
                return err
            }

            status := output.QueryExecution.Status.State
            switch status {
            case types.QueryExecutionStateSucceeded:
                return nil
            case types.QueryExecutionStateFailed, types.QueryExecutionStateCancelled:
                reason := ""
                if output.QueryExecution.Status.StateChangeReason != nil {
                    reason = *output.QueryExecution.Status.StateChangeReason
                }
                return fmt.Errorf("query failed: %s - %s", status, reason)
            }
        }
    }
}

func (e *AthenaEngine) fetchResults(ctx context.Context, queryID string) (*model.QueryResult, error) {
    input := &athena.GetQueryResultsInput{
        QueryExecutionId: aws.String(queryID),
        MaxResults:       aws.Int32(1000),
    }

    output, err := e.client.GetQueryResults(ctx, input)
    if err != nil {
        return nil, err
    }

    // 解析结果
    result := &model.QueryResult{
        Metadata: model.ResultMetadata{
            QueryID: queryID,
            Engine:  "athena",
            Vendor:  constant.ProviderAWS,
        },
    }

    // 提取列名（第一行）
    if len(output.ResultSet.Rows) > 0 {
        headerRow := output.ResultSet.Rows[0]
        for _, col := range headerRow.Data {
            result.Columns = append(result.Columns, *col.VarCharValue)
        }

        // 提取数据行
        for _, row := range output.ResultSet.Rows[1:] {
            var rowData []interface{}
            for _, col := range row.Data {
                if col.VarCharValue != nil {
                    rowData = append(rowData, *col.VarCharValue)
                } else {
                    rowData = append(rowData, nil)
                }
            }
            result.Rows = append(result.Rows, rowData)
        }
    }

    result.Metadata.RowCount = int64(len(result.Rows))

    return result, nil
}

func (e *AthenaEngine) Health(ctx context.Context) error {
    _, err := e.client.ListWorkGroups(ctx, &athena.ListWorkGroupsInput{
        MaxResults: aws.Int32(1),
    })
    return err
}
```

#### 4.4.3 BigQuery 引擎

```go
// engine/bigquery.go

type BigQueryEngine struct {
    client    *bigquery.Client
    projectID string
    dataset   string
}

func NewBigQueryEngine(cfg *config.GCPConfig) (*BigQueryEngine, error) {
    ctx := context.Background()

    client, err := bigquery.NewClient(ctx, cfg.ProjectID)
    if err != nil {
        return nil, err
    }

    return &BigQueryEngine{
        client:    client,
        projectID: cfg.ProjectID,
        dataset:   cfg.BigQueryDataset,
    }, nil
}

func (e *BigQueryEngine) Name() string {
    return "bigquery"
}

func (e *BigQueryEngine) Execute(ctx context.Context, sql string, dataSource *model.DataSourceInfo) (*model.QueryResult, error) {
    // 创建查询
    q := e.client.Query(sql)
    q.UseStandardSQL = true

    // 执行查询
    job, err := q.Run(ctx)
    if err != nil {
        return nil, fmt.Errorf("query run failed: %w", err)
    }

    // 等待完成
    status, err := job.Wait(ctx)
    if err != nil {
        return nil, fmt.Errorf("query wait failed: %w", err)
    }

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

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

    result := &model.QueryResult{
        Metadata: model.ResultMetadata{
            QueryID: job.ID(),
            Engine:  "bigquery",
            Vendor:  constant.ProviderGCP,
        },
    }

    // 读取列名
    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, 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))
    result.Metadata.BytesScanned = int64(status.Statistics.TotalBytesProcessed)

    return result, nil
}

func (e *BigQueryEngine) Health(ctx context.Context) error {
    // 简单查询测试连接
    q := e.client.Query("SELECT 1")
    _, err := q.Run(ctx)
    return err
}
```

#### 4.4.4 DLA 引擎（阿里云）

```go
// engine/dla.go

type DLAEngine struct {
    endpoint  string
    accessKey string
    secretKey string
}

func NewDLAEngine(cfg *config.AliCloudConfig) (*DLAEngine, error) {
    return &DLAEngine{
        endpoint:  cfg.DLAEndpoint,
        accessKey: cfg.AccessKeyID,
        secretKey: cfg.AccessKeySecret,
    }, nil
}

func (e *DLAEngine) Name() string {
    return "dla"
}

func (e *DLAEngine) Execute(ctx context.Context, sql string, dataSource *model.DataSourceInfo) (*model.QueryResult, error) {
    // DLA 基于 Presto，实现类似 Athena
    // 1. 提交查询
    // 2. 轮询状态
    // 3. 获取结果

    // TODO: 实现阿里云 DLA SDK 调用
    return nil, fmt.Errorf("DLA engine not implemented yet")
}

func (e *DLAEngine) Health(ctx context.Context) error {
    // TODO: 健康检查
    return nil
}
```

#### 4.4.5 DuckDB 引擎

```go
// engine/duckdb.go

type DuckDBEngine struct {
    // DuckDB 可以直接查询 S3 上的 Parquet/JSON 文件
    // 复用现有的 pkg/duckdb/duckdb_query.go
    s3AccessKey string
    s3SecretKey string
}

func NewDuckDBEngine(cfg *config.DuckDBConfig) (*DuckDBEngine, error) {
    return &DuckDBEngine{
        s3AccessKey: cfg.S3AccessKey,
        s3SecretKey: cfg.S3SecretKey,
    }, nil
}

func (e *DuckDBEngine) Name() string {
    return "duckdb"
}

func (e *DuckDBEngine) Execute(ctx context.Context, sql string, dataSource *model.DataSourceInfo) (*model.QueryResult, error) {
    // 使用现有的 DuckDB 查询能力
    // 参考 pkg/duckdb/duckdb_query.go

    // TODO: 集成现有 DuckDB 实现
    return nil, fmt.Errorf("DuckDB engine not implemented yet")
}

func (e *DuckDBEngine) Health(ctx context.Context) error {
    return nil
}
```

### 4.5 核心服务 (service/)

```go
// service/dataproxy_service.go

// DataProxyService Clinic Data Proxy 核心服务
type DataProxyService struct {
    router *router.Router
    config *config.DataProxyConfig
    logger *zap.Logger
}

// NewDataProxyService 创建服务实例
func NewDataProxyService(cfg *config.DataProxyConfig) (*DataProxyService, error) {
    r, err := router.NewRouter(cfg)
    if err != nil {
        return nil, err
    }

    return &DataProxyService{
        router: r,
        config: cfg,
        logger: log.L(),
    }, nil
}

// Query 执行查询（核心方法）
func (s *DataProxyService) Query(ctx context.Context, req *model.QueryRequest) (*model.QueryResult, error) {
    startTime := time.Now()

    // 1. 解析数据源
    dsInfo, err := router.ParseDataSource(req.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
    }

    s.logger.Info("query routing",
        zap.String("vendor", dsInfo.Vendor),
        zap.String("region", dsInfo.Region),
        zap.String("data_source", req.DataSource),
    )

    // 2. 获取 SQL 方言
    dialect := s.router.GetDialect(dsInfo.Vendor)

    // 3. 验证并转换 SQL
    if err := dialect.Validate(req.SQL); err != nil {
        return nil, fmt.Errorf("SQL validation failed: %w", err)
    }

    rewrittenSQL, err := dialect.Rewrite(req.SQL)
    if err != nil {
        return nil, fmt.Errorf("SQL rewrite failed: %w", err)
    }

    s.logger.Info("SQL rewritten",
        zap.String("dialect", dialect.Name()),
        zap.String("original", req.SQL),
        zap.String("rewritten", rewrittenSQL),
    )

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

    // 5. 执行查询
    queryCtx := ctx
    if req.Timeout > 0 {
        var cancel context.CancelFunc
        queryCtx, cancel = context.WithTimeout(ctx, req.Timeout)
        defer cancel()
    }

    result, err := engine.Execute(queryCtx, rewrittenSQL, dsInfo)
    if err != nil {
        return nil, fmt.Errorf("query execution failed: %w", err)
    }

    // 6. 填充元数据
    result.Metadata.ExecutionTime = time.Since(startTime)
    result.Metadata.Vendor = dsInfo.Vendor
    result.Metadata.Region = dsInfo.Region

    s.logger.Info("query completed",
        zap.String("query_id", result.Metadata.QueryID),
        zap.Int64("row_count", result.Metadata.RowCount),
        zap.Duration("execution_time", result.Metadata.ExecutionTime),
    )

    return result, nil
}

// BatchQuery 批量查询（跨 Region 聚合）
func (s *DataProxyService) BatchQuery(ctx context.Context, requests []*model.QueryRequest) ([]*model.QueryResult, error) {
    results := make([]*model.QueryResult, len(requests))
    errChan := make(chan error, len(requests))

    var wg sync.WaitGroup
    for i, req := range requests {
        wg.Add(1)
        go func(index int, request *model.QueryRequest) {
            defer wg.Done()
            result, err := s.Query(ctx, request)
            if err != nil {
                errChan <- fmt.Errorf("query %d failed: %w", index, err)
                return
            }
            results[index] = result
        }(i, req)
    }

    wg.Wait()
    close(errChan)

    // 收集错误
    var errs []error
    for err := range errChan {
        errs = append(errs, err)
    }

    if len(errs) > 0 {
        return nil, fmt.Errorf("batch query failed: %v", errs)
    }

    return results, nil
}

// Health 健康检查
func (s *DataProxyService) Health(ctx context.Context) error {
    // 检查所有引擎的健康状态
    for vendor, eng := range s.router.engines {
        if err := eng.Health(ctx); err != nil {
            return fmt.Errorf("engine %s unhealthy: %w", vendor, err)
        }
    }
    return nil
}
```

### 4.6 HTTP 处理器 (handler/)

```go
// handler/dataproxy_handler.go

type DataProxyHandler struct {
    service *service.DataProxyService
}

func NewDataProxyHandler(svc *service.DataProxyService) *DataProxyHandler {
    return &DataProxyHandler{
        service: svc,
    }
}

// Query HTTP 查询接口
// POST /dataproxy/query
func (h *DataProxyHandler) Query(c *gin.Context) {
    var req model.QueryRequest
    if err := c.ShouldBindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{
            "error": fmt.Sprintf("invalid request: %v", err),
        })
        return
    }

    // 设置默认超时
    if req.Timeout == 0 {
        req.Timeout = 5 * time.Minute
    }

    result, err := h.service.Query(c.Request.Context(), &req)
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{
            "error": fmt.Sprintf("query failed: %v", err),
        })
        return
    }

    c.JSON(http.StatusOK, result)
}

// BatchQuery HTTP 批量查询接口
// POST /dataproxy/batch-query
func (h *DataProxyHandler) BatchQuery(c *gin.Context) {
    var requests []*model.QueryRequest
    if err := c.ShouldBindJSON(&requests); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{
            "error": fmt.Sprintf("invalid request: %v", err),
        })
        return
    }

    results, err := h.service.BatchQuery(c.Request.Context(), requests)
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{
            "error": fmt.Sprintf("batch query failed: %v", err),
        })
        return
    }

    c.JSON(http.StatusOK, gin.H{
        "results": results,
    })
}

// Health 健康检查接口
// GET /dataproxy/health
func (h *DataProxyHandler) Health(c *gin.Context) {
    if err := h.service.Health(c.Request.Context()); err != nil {
        c.JSON(http.StatusServiceUnavailable, gin.H{
            "status": "unhealthy",
            "error":  err.Error(),
        })
        return
    }

    c.JSON(http.StatusOK, gin.H{
        "status": "healthy",
    })
}

// RegisterRoutes 注册路由
func (h *DataProxyHandler) RegisterRoutes(r *gin.Engine) {
    group := r.Group("/dataproxy")
    {
        group.POST("/query", h.Query)
        // group.POST("/batch-query", h.BatchQuery)
        group.GET("/health", h.Health)
    }
}
```

## 五、配置管理

### 5.1 配置结构

```go
// config/dataproxy_config.go

type DataProxyConfig struct {
    AWS      AWSConfig      `yaml:"aws"`
    GCP      GCPConfig      `yaml:"gcp"`
    AliCloud AliCloudConfig `yaml:"alicloud"`
    DuckDB   DuckDBConfig   `yaml:"duckdb"`
}

type AWSConfig struct {
    Enabled               bool   `yaml:"enabled"`
    Region                string `yaml:"region"`
    AthenaDatabase        string `yaml:"athena_database"`
    AthenaWorkgroup       string `yaml:"athena_workgroup"`
    AthenaOutputLocation  string `yaml:"athena_output_location"`
}

type GCPConfig struct {
    Enabled          bool   `yaml:"enabled"`
    ProjectID        string `yaml:"project_id"`
    BigQueryDataset  string `yaml:"bigquery_dataset"`
}

type AliCloudConfig struct {
    Enabled         bool   `yaml:"enabled"`
    Region          string `yaml:"region"`
    DLAEndpoint     string `yaml:"dla_endpoint"`
    AccessKeyID     string `yaml:"access_key_id"`
    AccessKeySecret string `yaml:"access_key_secret"`
}

type DuckDBConfig struct {
    S3AccessKey string `yaml:"s3_access_key"`
    S3SecretKey string `yaml:"s3_secret_key"`
}
```

### 5.2 配置文件示例

```yaml
# config-dataproxy.yaml

aws:
  enabled: true
  region: us-east-1
  athena_database: clinic_data
  athena_workgroup: primary
  athena_output_location: s3://clinic-athena-results/

gcp:
  enabled: true
  project_id: clinic-gcp-project
  bigquery_dataset: clinic_data

alicloud:
  enabled: true
  region: ap-southeast-1
  dla_endpoint: https://dla.ap-southeast-1.aliyuncs.com
  access_key_id: ${ALICLOUD_ACCESS_KEY}
  access_key_secret: ${ALICLOUD_SECRET_KEY}

duckdb:
  s3_access_key: ${AWS_ACCESS_KEY}
  s3_secret_key: ${AWS_SECRET_KEY}
```

## 六、API 接口设计

### 6.1 查询接口

**请求**
```http
POST /dataproxy/query
Content-Type: application/json

{
  "sql": "SELECT * FROM topsql WHERE timestamp > to_timestamp(1704067200) LIMIT 100",
  "data_source": "s3://o11y-prod-shared/diagnosis/data/topsql/",
  "timeout": "5m",
  "options": {
    "max_rows": 1000,
    "use_cache": true,
    "output_format": "json"
  }
}
```

**响应**
```json
{
  "columns": ["timestamp", "sql_digest", "cpu_time", "latency"],
  "rows": [
    [1704067200, "abc123", 1500, 200],
    [1704067201, "def456", 2000, 300]
  ],
  "metadata": {
    "row_count": 2,
    "bytes_scanned": 1048576,
    "execution_time": "2.5s",
    "query_id": "query-12345",
    "engine": "athena",
    "vendor": "aws",
    "region": "us-east-1"
  }
}
```

### 6.2 批量查询接口

**请求**
```http
POST /dataproxy/batch-query
Content-Type: application/json

[
  {
    "sql": "SELECT COUNT(*) FROM topsql",
    "data_source": "s3://bucket-us-east-1/topsql/"
  },
  {
    "sql": "SELECT COUNT(*) FROM topsql",
    "data_source": "s3://bucket-ap-southeast-1/topsql/"
  }
]
```

**响应**
```json
{
  "results": [
    {
      "columns": ["count"],
      "rows": [[1000]],
      "metadata": {...}
    },
    {
      "columns": ["count"],
      "rows": [[800]],
      "metadata": {...}
    }
  ]
}
```

## 七、部署方案

### 7.1 部署架构

```
┌──────────────────────────────────────────┐
│       Load Balancer (ALB/NLB)            │
└────────────┬─────────────────────────────┘
             │
     ┌───────┴───────┐
     ▼               ▼
┌─────────┐     ┌─────────┐
│ Proxy 1 │     │ Proxy 2 │  (Auto Scaling)
└─────────┘     └─────────┘
     │               │
     └───────┬───────┘
             ▼
    ┌─────────────────┐
    │  Cloud Engines  │
    │ (Athena/BQ/DLA) │
    └─────────────────┘
```

### 7.2 部署配置

**Docker 镜像**
```dockerfile
FROM golang:1.21-alpine AS builder
WORKDIR /app
COPY . .
RUN go build -o dataproxy cmd/dataproxy/main.go

FROM alpine:latest
RUN apk --no-cache add ca-certificates
COPY --from=builder /app/dataproxy /usr/local/bin/
COPY configs/config-dataproxy.yaml /etc/dataproxy/config.yaml
ENTRYPOINT ["dataproxy"]
CMD ["--config", "/etc/dataproxy/config.yaml"]
```

**Kubernetes Deployment**
```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: clinic_data_proxy
spec:
  replicas: 3
  selector:
    matchLabels:
      app: data-proxy
  template:
    metadata:
      labels:
        app: data-proxy
    spec:
      containers:
      - name: proxy
        image: clinic/data-proxy:latest
        ports:
        - containerPort: 8080
        env:
        - name: AWS_REGION
          value: us-east-1
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
```

## 八、监控与可观测性

### 8.1 Metrics 指标

```go
// 查询延迟
dataproxy_query_duration_seconds{vendor, region, engine}

// 查询成功/失败数
dataproxy_query_total{vendor, region, engine, status}

// 扫描的数据量
dataproxy_bytes_scanned_total{vendor, region, engine}

// 并发查询数
dataproxy_concurrent_queries{vendor, region, engine}

// 引擎健康状态
dataproxy_engine_health{vendor, region, engine}
```

### 8.2 日志规范

```go
// 查询开始
log.Info("query started",
    zap.String("query_id", queryID),
    zap.String("vendor", vendor),
    zap.String("sql", sql),
)

// 查询完成
log.Info("query completed",
    zap.String("query_id", queryID),
    zap.Int64("row_count", rowCount),
    zap.Duration("duration", duration),
)

// 查询失败
log.Error("query failed",
    zap.String("query_id", queryID),
    zap.Error(err),
)
```

## 九、测试策略

### 9.1 单元测试

- 每个模块独立测试
- Mock 外部依赖（AWS SDK、GCP SDK 等）
- 覆盖率目标：80%+

### 9.2 集成测试

- 实际调用云厂商 API（使用测试账号）
- 验证端到端查询流程
- 测试多云场景

### 9.3 性能测试

- 并发查询测试（1000+ QPS）
- 大数据量查询测试（TB 级）
- Region 间延迟测试

## 十、实施计划

### Phase 1：基础框架（Week 1-2）
- [ ] 数据模型定义
- [ ] Router 和 DataSource 解析
- [ ] Dialect 接口和基础实现
- [ ] Engine 接口定义

### Phase 2：引擎实现（Week 3-4）
- [ ] Athena Engine
- [ ] BigQuery Engine
- [ ] DuckDB Engine（复用现有）
- [ ] DLA Engine（可选）

### Phase 3：服务层和 API（Week 5）
- [ ] DataProxyService 实现
- [ ] HTTP Handler 实现
- [ ] 路由注册
- [ ] 配置管理

### Phase 4：测试和优化（Week 6）
- [ ] 单元测试
- [ ] 集成测试
- [ ] 性能优化
- [ ] 文档完善

### Phase 5：部署上线（Week 7）
- [ ] Docker 镜像构建
- [ ] K8s 部署
- [ ] 监控告警配置
- [ ] 灰度发布

## 十一、风险与对策

| 风险 | 影响 | 对策 |
|------|------|------|
| 云厂商 API 限流 | 查询失败 | 实现指数退避重试、请求限流 |
| 大查询超时 | 用户体验差 | 异步查询模式、结果缓存 |
| SQL 方言转换错误 | 查询失败 | 完善测试用例、提供方言调试工具 |
| 跨 Region 延迟高 | 性能下降 | Regional 缓存、预聚合 |
| 成本过高 | 预算超支 | 查询成本监控、限额控制 |

## 十二、后续演进方向

1. **查询优化器**：自动优化 SQL，减少扫描数据量
2. **结果缓存**：Redis/Memcached 缓存热查询结果
3. **成本分析**：查询成本实时统计和优化建议
4. **可视化界面**：Web UI 支持交互式查询
5. **权限管理**：细粒度的数据访问控制
6. **联邦查询**：跨多个数据源的 JOIN 查询

---

## 附录：参考资料

1. AWS Athena SDK 文档：https://docs.aws.amazon.com/athena/
2. GCP BigQuery SDK 文档：https://cloud.google.com/bigquery/docs
3. DuckDB 文档：https://duckdb.org/docs/
4. Presto SQL 参考：https://prestodb.io/docs/current/
5. 现有代码库：`pkg/duckdb/`, `pkg/s3/`, `pkg/thirdapi/`