+------------------------------------------------------------------+
|                    Metric System Architecture                     |
|                                                                   |
|  +---------------------+     +-----------------------------+      |
|  | MetricRegistryService|     | ComputationCoordinator     |      |
|  |                     |     |                             |      |
|  | - register()        |     | - route(metric, context)   |      |
|  | - update()          |     | - evaluate(strategy)       |      |
|  | - delete()          |     | - fallback()               |      |
|  | - list()            |     +-----------------------------+      |
|  | - get()             |          |                               |
|  +---------------------+          v                               |
|           |              +-----------------------------+          |
|           v              |   Strategy Executors        |          |
|  +------------------+   |                             |          |
|  | OQL Rewriter     |   | REALTIME    -> live query   |          |
|  |                  |   | VIRTUAL_COL -> Doris VC     |          |
|  | metric('x')  ──> |   | UDF         -> custom func  |          |
|  | expanded SQL     |   | CACHED      -> Redis cache  |          |
|  +------------------+   | ROLLUP      -> pre-agg      |          |
|                         | MATERIALIZED-> MV table     |          |
|                         +-----------------------------+          |
+------------------------------------------------------------------+

class MetricDefinition(BaseModel):
    """指标定义"""
    metric_id: str = Field(default_factory=lambda: str(uuid4()))
    name: str                          # 唯一名称: monthly_active_customers
    display_name: str                  # 显示名: 月活跃客户数
    description: str | None = None
    object_type: str                   # 所属对象类型: Customer
    category: MetricCategory           # 分类
    data_type: MetricDataType          # 返回值类型
    expression: MetricExpression       # 计算表达式
    strategies: list[ComputationStrategy]  # 允许的计算策略（按优先级排序）
    freshness_requirement: FreshnessLevel  # 新鲜度要求
    cache_ttl: int = 300               # 缓存 TTL（秒）
    tags: list[str] = []
    owner: str | None = None
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)
    version: int = 1

class MetricCategory(str, Enum):
    BUSINESS = "BUSINESS"        # 业务指标
    OPERATIONAL = "OPERATIONAL"  # 运营指标
    FINANCIAL = "FINANCIAL"      # 财务指标
    TECHNICAL = "TECHNICAL"      # 技术指标
    CUSTOM = "CUSTOM"            # 自定义

class MetricDataType(str, Enum):
    INTEGER = "INTEGER"
    DECIMAL = "DECIMAL"
    PERCENTAGE = "PERCENTAGE"
    CURRENCY = "CURRENCY"
    DURATION = "DURATION"

class FreshnessLevel(str, Enum):
    REALTIME = "REALTIME"        # 实时（< 1s 延迟）
    NEAR_REALTIME = "NEAR_REALTIME"  # 准实时（< 1min）
    HOURLY = "HOURLY"            # 小时级
    DAILY = "DAILY"              # 天级
    WEEKLY = "WEEKLY"            # 周级

class MetricExpression(BaseModel):
    """指标计算表达式"""
    type: ExpressionType
    sql: str | None = None           # SQL 表达式
    aggregation: AggregationType | None = None
    source_field: str | None = None  # 源字段
    filter_clause: str | None = None # 过滤条件
    group_by: list[str] = []         # 分组字段
    window: WindowSpec | None = None # 窗口规格

class ExpressionType(str, Enum):
    SIMPLE_AGG = "SIMPLE_AGG"       # COUNT/SUM/AVG/MAX/MIN
    SQL = "SQL"                      # 自定义 SQL
    COMPOSITE = "COMPOSITE"          # 组合其他指标
    PYTHON_UDF = "PYTHON_UDF"        # Python UDF

class AggregationType(str, Enum):
    COUNT = "COUNT"
    COUNT_DISTINCT = "COUNT_DISTINCT"
    SUM = "SUM"
    AVG = "AVG"
    MAX = "MAX"
    MIN = "MIN"
    MEDIAN = "MEDIAN"
    PERCENTILE = "PERCENTILE"

# 示例 1: 简单聚合 — 客户总订单数
order_count_metric = MetricDefinition(
    name="customer_order_count",
    display_name="客户总订单数",
    object_type="Customer",
    category=MetricCategory.BUSINESS,
    data_type=MetricDataType.INTEGER,
    expression=MetricExpression(
        type=ExpressionType.SIMPLE_AGG,
        aggregation=AggregationType.COUNT,
        source_field="orders",  # 关系名
    ),
    strategies=[
        ComputationStrategy.CACHED,
        ComputationStrategy.ROLLUP,
        ComputationStrategy.REALTIME,
    ],
    freshness_requirement=FreshnessLevel.HOURLY,
    cache_ttl=3600,
)

# 示例 2: SQL 表达式 — 客户 30 天 GMV
gmv_30d_metric = MetricDefinition(
    name="customer_gmv_30d",
    display_name="客户30天GMV",
    object_type="Customer",
    category=MetricCategory.FINANCIAL,
    data_type=MetricDataType.CURRENCY,
    expression=MetricExpression(
        type=ExpressionType.SQL,
        sql="""
        SELECT SUM(o.total_amount)
        FROM orders o
        WHERE o.customer_id = {object_id}
          AND o.created_at >= DATE_SUB(NOW(), INTERVAL 30 DAY)
          AND o.status != 'CANCELLED'
        """,
    ),
    strategies=[
        ComputationStrategy.MATERIALIZED,
        ComputationStrategy.CACHED,
        ComputationStrategy.REALTIME,
    ],
    freshness_requirement=FreshnessLevel.DAILY,
    cache_ttl=86400,
)

# 示例 3: 组合指标 — 客均 GMV
avg_gmv_metric = MetricDefinition(
    name="average_customer_gmv",
    display_name="客均GMV",
    object_type="Customer",
    category=MetricCategory.FINANCIAL,
    data_type=MetricDataType.CURRENCY,
    expression=MetricExpression(
        type=ExpressionType.COMPOSITE,
        sql="metric('total_gmv') / metric('active_customer_count')",
    ),
    strategies=[
        ComputationStrategy.CACHED,
        ComputationStrategy.REALTIME,
    ],
    freshness_requirement=FreshnessLevel.DAILY,
)

class RealtimeExecutor(StrategyExecutor):
    """实时计算执行器"""

    async def execute(
        self, metric: MetricDefinition, context: ComputeContext
    ) -> MetricResult:
        sql = self._build_sql(metric, context)

        start = time.monotonic()
        result = await self._doris.execute(sql)
        elapsed_ms = (time.monotonic() - start) * 1000

        return MetricResult(
            metric_name=metric.name,
            value=result[0]["value"] if result else None,
            computed_at=datetime.utcnow(),
            strategy_used=ComputationStrategy.REALTIME,
            latency_ms=elapsed_ms,
            is_stale=False,
        )

    def _build_sql(
        self, metric: MetricDefinition, context: ComputeContext
    ) -> str:
        expr = metric.expression

        if expr.type == ExpressionType.SIMPLE_AGG:
            agg = expr.aggregation.value
            source = expr.source_field
            return f"""
            SELECT {agg}({source}) AS value
            FROM {metric.object_type.lower()}_objects
            WHERE object_id = '{context.object_id}'
            """

        elif expr.type == ExpressionType.SQL:
            return expr.sql.replace("{object_id}", context.object_id)

        raise MetricError(f"Unsupported expression type for REALTIME: {expr.type}")

-- 创建虚拟列
ALTER TABLE customer_objects ADD COLUMN
  order_total_amount DECIMAL(18,2)
  GENERATED ALWAYS AS (
    SELECT COALESCE(SUM(amount), 0)
    FROM order_objects
    WHERE customer_id = customer_objects.object_id
  ) VIRTUAL;

class VirtualColumnExecutor(StrategyExecutor):
    """Doris 虚拟列执行器"""

    async def execute(
        self, metric: MetricDefinition, context: ComputeContext
    ) -> MetricResult:
        column_name = self._get_virtual_column_name(metric)

        sql = f"""
        SELECT {column_name} AS value
        FROM {metric.object_type.lower()}_objects
        WHERE object_id = '{context.object_id}'
        """

        result = await self._doris.execute(sql)
        return MetricResult(
            metric_name=metric.name,
            value=result[0]["value"] if result else None,
            computed_at=datetime.utcnow(),
            strategy_used=ComputationStrategy.VIRTUAL_COLUMN,
            latency_ms=0,
            is_stale=False,
        )

    def _get_virtual_column_name(self, metric: MetricDefinition) -> str:
        return f"vc_{metric.name}"

class UdfExecutor(StrategyExecutor):
    """UDF 计算执行器"""

    def __init__(self, udf_registry: dict[str, Callable]):
        self._udf_registry = udf_registry

    async def execute(
        self, metric: MetricDefinition, context: ComputeContext
    ) -> MetricResult:
        udf_name = metric.expression.sql  # UDF 名称存在 sql 字段
        udf_func = self._udf_registry.get(udf_name)
        if udf_func is None:
            raise MetricError(f"UDF not found: {udf_name}")

        # 获取计算所需的数据
        data = await self._fetch_data(metric, context)

        start = time.monotonic()
        value = await udf_func(data, context)
        elapsed_ms = (time.monotonic() - start) * 1000

        return MetricResult(
            metric_name=metric.name,
            value=value,
            computed_at=datetime.utcnow(),
            strategy_used=ComputationStrategy.UDF,
            latency_ms=elapsed_ms,
            is_stale=False,
        )

# UDF 注册示例
@register_metric_udf("customer_health_score")
async def compute_health_score(
    data: MetricData, context: ComputeContext
) -> float:
    """计算客户健康分数 — 综合多个维度"""
    order_frequency = data.get("order_frequency", 0)
    avg_order_value = data.get("avg_order_value", 0)
    days_since_last_order = data.get("days_since_last_order", 999)
    support_tickets = data.get("support_tickets", 0)

    frequency_score = min(order_frequency / 10, 1.0) * 30
    value_score = min(avg_order_value / 10000, 1.0) * 25
    recency_score = max(0, 1 - days_since_last_order / 365) * 30
    satisfaction_score = max(0, 1 - support_tickets / 20) * 15

    return round(frequency_score + value_score + recency_score + satisfaction_score, 2)

class CachedExecutor(StrategyExecutor):
    """缓存计算执行器"""

    CACHE_PREFIX = "metric:cache:"

    async def execute(
        self, metric: MetricDefinition, context: ComputeContext
    ) -> MetricResult:
        cache_key = self._cache_key(metric, context)

        # 尝试读取缓存
        cached = await self._redis.get(cache_key)
        if cached:
            data = json.loads(cached)
            return MetricResult(
                metric_name=metric.name,
                value=data["value"],
                computed_at=datetime.fromisoformat(data["computed_at"]),
                strategy_used=ComputationStrategy.CACHED,
                latency_ms=0,
                is_stale=self._is_stale(data, metric),
            )

        # 缓存未命中，回退到实时计算
        result = await self._realtime_executor.execute(metric, context)

        # 写入缓存
        await self._redis.setex(
            cache_key,
            metric.cache_ttl,
            json.dumps({
                "value": result.value,
                "computed_at": result.computed_at.isoformat(),
            }),
        )

        result.strategy_used = ComputationStrategy.CACHED
        return result

    def _cache_key(
        self, metric: MetricDefinition, context: ComputeContext
    ) -> str:
        return f"{self.CACHE_PREFIX}{metric.name}:{context.object_id}"

    def _is_stale(
        self, data: dict, metric: MetricDefinition
    ) -> bool:
        computed = datetime.fromisoformat(data["computed_at"])
        age = (datetime.utcnow() - computed).total_seconds()
        return age > metric.cache_ttl * 0.8  # 80% TTL 后标记为 stale

class RollupExecutor(StrategyExecutor):
    """预聚合执行器"""

    async def execute(
        self, metric: MetricDefinition, context: ComputeContext
    ) -> MetricResult:
        rollup_table = self._get_rollup_table(metric)

        sql = f"""
        SELECT metric_value AS value, computed_at
        FROM {rollup_table}
        WHERE metric_name = '{metric.name}'
          AND object_id = '{context.object_id}'
        ORDER BY computed_at DESC
        LIMIT 1
        """

        result = await self._doris.execute(sql)
        if not result:
            # Rollup 表中无数据，回退
            raise StrategyUnavailableError("No rollup data available")

        return MetricResult(
            metric_name=metric.name,
            value=result[0]["value"],
            computed_at=result[0]["computed_at"],
            strategy_used=ComputationStrategy.ROLLUP,
            latency_ms=0,
            is_stale=self._check_staleness(result[0]["computed_at"], metric),
        )

    def _get_rollup_table(self, metric: MetricDefinition) -> str:
        freshness = metric.freshness_requirement
        if freshness in (FreshnessLevel.HOURLY, FreshnessLevel.NEAR_REALTIME):
            return "metric_rollup_hourly"
        elif freshness == FreshnessLevel.DAILY:
            return "metric_rollup_daily"
        else:
            return "metric_rollup_weekly"

-- 每小时 Rollup 任务
INSERT INTO metric_rollup_hourly (metric_name, object_id, metric_value, computed_at)
SELECT
    'customer_order_count' AS metric_name,
    c.object_id,
    COUNT(o.object_id) AS metric_value,
    NOW() AS computed_at
FROM customer_objects c
LEFT JOIN order_objects o ON o.customer_id = c.object_id
GROUP BY c.object_id
ON DUPLICATE KEY UPDATE
    metric_value = VALUES(metric_value),
    computed_at = VALUES(computed_at);

class MaterializedViewExecutor(StrategyExecutor):
    """物化视图执行器"""

    async def execute(
        self, metric: MetricDefinition, context: ComputeContext
    ) -> MetricResult:
        mv_name = f"mv_{metric.name}"

        sql = f"""
        SELECT metric_value AS value
        FROM {mv_name}
        WHERE object_id = '{context.object_id}'
        """

        result = await self._doris.execute(sql)
        if not result:
            raise StrategyUnavailableError(f"MV {mv_name} has no data")

        return MetricResult(
            metric_name=metric.name,
            value=result[0]["value"],
            computed_at=datetime.utcnow(),  # MV 由 Doris 自动维护
            strategy_used=ComputationStrategy.MATERIALIZED,
            latency_ms=0,
            is_stale=False,
        )

class ComputationCoordinator:
    """指标计算协调器 — 策略优先级路由"""

    def __init__(self):
        self._executors: dict[ComputationStrategy, StrategyExecutor] = {
            ComputationStrategy.REALTIME: RealtimeExecutor(),
            ComputationStrategy.VIRTUAL_COLUMN: VirtualColumnExecutor(),
            ComputationStrategy.UDF: UdfExecutor(),
            ComputationStrategy.CACHED: CachedExecutor(),
            ComputationStrategy.ROLLUP: RollupExecutor(),
            ComputationStrategy.MATERIALIZED: MaterializedViewExecutor(),
        }

    async def compute(
        self,
        metric: MetricDefinition,
        context: ComputeContext,
    ) -> MetricResult:
        """按优先级依次尝试计算策略"""
        errors: list[str] = []

        for strategy in metric.strategies:
            executor = self._executors.get(strategy)
            if executor is None:
                continue

            # 检查策略是否可用
            if not await self._is_strategy_available(strategy, metric, context):
                errors.append(f"{strategy.value}: unavailable")
                continue

            try:
                result = await executor.execute(metric, context)

                # 检查结果新鲜度
                if result.is_stale and self._has_fresher_strategy(
                    metric, strategy
                ):
                    # 结果过期且有更新鲜的策略，触发异步刷新
                    await self._schedule_refresh(metric, context)

                return result

            except StrategyUnavailableError as e:
                errors.append(f"{strategy.value}: {e}")
                continue
            except Exception as e:
                errors.append(f"{strategy.value}: error - {e}")
                logger.warning(
                    f"Strategy {strategy.value} failed for {metric.name}: {e}"
                )
                continue

        # 所有策略都失败
        raise MetricComputeError(
            f"All strategies failed for metric {metric.name}: "
            + "; ".join(errors)
        )

    async def _is_strategy_available(
        self,
        strategy: ComputationStrategy,
        metric: MetricDefinition,
        context: ComputeContext,
    ) -> bool:
        """检查策略是否可用"""
        if strategy == ComputationStrategy.VIRTUAL_COLUMN:
            return await self._has_virtual_column(metric)
        if strategy == ComputationStrategy.MATERIALIZED:
            return await self._has_materialized_view(metric)
        if strategy == ComputationStrategy.ROLLUP:
            return await self._has_rollup_data(metric, context)
        if strategy == ComputationStrategy.UDF:
            return metric.expression.type == ExpressionType.PYTHON_UDF
        return True

    def _has_fresher_strategy(
        self,
        metric: MetricDefinition,
        current: ComputationStrategy,
    ) -> bool:
        """当前策略之后是否还有更实时的策略"""
        fresher = {
            ComputationStrategy.REALTIME,
            ComputationStrategy.VIRTUAL_COLUMN,
            ComputationStrategy.UDF,
        }
        idx = metric.strategies.index(current)
        remaining = metric.strategies[idx + 1:]
        return any(s in fresher for s in remaining)

ComputationCoordinator.compute(metric, context)
       |
       v
  strategies = [MATERIALIZED, CACHED, ROLLUP, REALTIME]
       |
       v
  [1] MATERIALIZED available? ──NO──> skip
       |YES
       v
  Execute MV query
       |
  Success? ──NO──> [2] CACHED
       |YES
       v
  Result stale? ──YES──> schedule async refresh
       |NO
       v
  Return result
       |
  [2] CACHED available? ──YES──> check Redis
       |                           |
       |                    Hit? ──YES──> return cached
       |                           |NO
       |                           v
       |                    Fallback to REALTIME, cache result
       |
  [3] ROLLUP available? ──YES──> query rollup table
       |                           |
       |                    Data found? ──YES──> return
       |                                  |NO
       |                                  v
       v                            fallback
  [4] REALTIME ──> execute live query ──> return

class BatchComputationCoordinator:
    """批量指标计算"""

    async def compute_batch(
        self,
        metric: MetricDefinition,
        object_ids: list[str],
    ) -> dict[str, MetricResult]:
        # 先批量查缓存
        cached_results = await self._batch_cache_lookup(metric, object_ids)
        missing_ids = [
            oid for oid in object_ids if oid not in cached_results
        ]

        if not missing_ids:
            return cached_results

        # 对未命中的批量实时计算
        batch_sql = self._build_batch_sql(metric, missing_ids)
        rows = await self._doris.execute(batch_sql)

        computed_results = {}
        for row in rows:
            oid = row["object_id"]
            result = MetricResult(
                metric_name=metric.name,
                value=row["value"],
                computed_at=datetime.utcnow(),
                strategy_used=ComputationStrategy.REALTIME,
            )
            computed_results[oid] = result
            # 写入缓存
            await self._cache_result(metric, oid, result)

        return {**cached_results, **computed_results}

-- 用户写的 OQL
SELECT
    customer.name,
    metric('customer_order_count') AS order_count,
    metric('customer_gmv_30d') AS gmv_30d
FROM Customer customer
WHERE metric('customer_gmv_30d') > 10000

-- 重写后的 SQL（策略=MATERIALIZED）
SELECT
    c.name,
    mv_order_count.metric_value AS order_count,
    mv_gmv_30d.metric_value AS gmv_30d
FROM customer_objects c
LEFT JOIN mv_customer_order_count mv_order_count
    ON mv_order_count.object_id = c.object_id
LEFT JOIN mv_customer_gmv_30d mv_gmv_30d
    ON mv_gmv_30d.object_id = c.object_id
WHERE mv_gmv_30d.metric_value > 10000

class OqlMetricRewriter:
    """OQL 指标重写器"""

    METRIC_PATTERN = re.compile(r"metric\(\s*'([^']+)'\s*\)")

    def __init__(
        self,
        registry: MetricRegistryService,
        coordinator: ComputationCoordinator,
    ):
        self._registry = registry
        self._coordinator = coordinator

    async def rewrite(self, oql: str, context: QueryContext) -> str:
        """重写 OQL 中的 metric() 调用"""
        # 1. 提取所有指标引用
        metric_refs = self.METRIC_PATTERN.findall(oql)
        if not metric_refs:
            return oql  # 无指标引用，原样返回

        # 2. 加载指标定义
        metrics = {}
        for name in metric_refs:
            metric_def = await self._registry.get_by_name(name)
            if metric_def is None:
                raise OqlError(f"Unknown metric: {name}")
            metrics[name] = metric_def

        # 3. 为每个指标确定最优策略
        strategy_map = {}
        for name, metric_def in metrics.items():
            strategy = await self._coordinator.select_strategy(
                metric_def, context
            )
            strategy_map[name] = strategy

        # 4. 根据策略生成重写后的 SQL
        rewritten = oql
        join_clauses = []
        for name, metric_def in metrics.items():
            strategy = strategy_map[name]
            replacement, join = self._generate_replacement(
                name, metric_def, strategy
            )
            rewritten = rewritten.replace(
                f"metric('{name}')", replacement
            )
            if join:
                join_clauses.append(join)

        # 5. 插入 JOIN 子句
        if join_clauses:
            rewritten = self._inject_joins(rewritten, join_clauses)

        return rewritten

    def _generate_replacement(
        self,
        name: str,
        metric: MetricDefinition,
        strategy: ComputationStrategy,
    ) -> tuple[str, str | None]:
        """生成指标的 SQL 替换"""
        if strategy == ComputationStrategy.MATERIALIZED:
            alias = f"mv_{name}"
            replacement = f"{alias}.metric_value"
            join = (
                f"LEFT JOIN mv_{name} {alias} "
                f"ON {alias}.object_id = main.object_id"
            )
            return replacement, join

        elif strategy == ComputationStrategy.VIRTUAL_COLUMN:
            replacement = f"main.vc_{name}"
            return replacement, None

        elif strategy == ComputationStrategy.ROLLUP:
            alias = f"rollup_{name}"
            replacement = f"{alias}.metric_value"
            join = (
                f"LEFT JOIN metric_rollup_daily {alias} "
                f"ON {alias}.object_id = main.object_id "
                f"AND {alias}.metric_name = '{name}'"
            )
            return replacement, join

        else:
            # REALTIME / CACHED / UDF: 使用子查询
            sub = metric.expression.sql.replace(
                "{object_id}", "main.object_id"
            )
            replacement = f"({sub})"
            return replacement, None

    def _inject_joins(
        self, sql: str, join_clauses: list[str]
    ) -> str:
        """在 FROM 子句后注入 JOIN"""
        joins = "\n".join(join_clauses)
        # 简单的 FROM 子句检测
        from_pattern = re.compile(
            r"(FROM\s+\w+\s+\w+)", re.IGNORECASE
        )
        match = from_pattern.search(sql)
        if match:
            insert_pos = match.end()
            return sql[:insert_pos] + "\n" + joins + sql[insert_pos:]
        return sql

class MetricRegistryService:
    """指标注册与管理服务"""

    async def register(
        self, definition: MetricDefinition
    ) -> MetricDefinition:
        """注册新指标"""
        # 验证
        await self._validate(definition)

        # 检查重名
        existing = await self._repo.find_by_name(definition.name)
        if existing:
            raise MetricError(
                f"Metric '{definition.name}' already exists"
            )

        # 保存定义
        await self._repo.save(definition)

        # 如果策略包含 MATERIALIZED，自动创建物化视图
        if ComputationStrategy.MATERIALIZED in definition.strategies:
            await self._mv_service.create_for_metric(definition)

        # 如果策略包含 ROLLUP，注册 Rollup 任务
        if ComputationStrategy.ROLLUP in definition.strategies:
            await self._rollup_scheduler.register(definition)

        # 如果策略包含 VIRTUAL_COLUMN，创建虚拟列
        if ComputationStrategy.VIRTUAL_COLUMN in definition.strategies:
            await self._doris_schema.add_virtual_column(definition)

        logger.info(f"Registered metric: {definition.name}")
        return definition

    async def update(
        self, metric_id: str, updates: MetricUpdateRequest
    ) -> MetricDefinition:
        """更新指标定义"""
        existing = await self._repo.get(metric_id)
        if existing is None:
            raise MetricNotFoundError(metric_id)

        # 应用更新
        updated = existing.model_copy(update=updates.model_dump(exclude_unset=True))
        updated.version = existing.version + 1
        updated.updated_at = datetime.utcnow()

        # 检查策略变更
        old_strategies = set(existing.strategies)
        new_strategies = set(updated.strategies)

        added = new_strategies - old_strategies
        removed = old_strategies - new_strategies

        # 处理新增策略
        if ComputationStrategy.MATERIALIZED in added:
            await self._mv_service.create_for_metric(updated)
        if ComputationStrategy.ROLLUP in added:
            await self._rollup_scheduler.register(updated)

        # 处理移除策略
        if ComputationStrategy.MATERIALIZED in removed:
            await self._mv_service.drop_for_metric(existing)
        if ComputationStrategy.ROLLUP in removed:
            await self._rollup_scheduler.unregister(existing)

        await self._repo.save(updated)
        return updated

    async def delete(self, metric_id: str) -> None:
        """删除指标"""
        existing = await self._repo.get(metric_id)
        if existing is None:
            raise MetricNotFoundError(metric_id)

        # 检查是否被其他指标引用（组合指标）
        dependents = await self._repo.find_dependents(existing.name)
        if dependents:
            names = [d.name for d in dependents]
            raise MetricError(
                f"Cannot delete: referenced by {names}"
            )

        # 清理关联资源
        await self._cleanup(existing)
        await self._repo.delete(metric_id)

    async def _validate(self, definition: MetricDefinition) -> None:
        """验证指标定义"""
        # 检查对象类型存在
        obj_type = await self._ontology.get_type(definition.object_type)
        if obj_type is None:
            raise MetricError(
                f"Object type '{definition.object_type}' not found"
            )

        # 检查 SQL 表达式语法
        if definition.expression.type == ExpressionType.SQL:
            await self._sql_validator.validate(definition.expression.sql)

        # 检查组合指标的依赖
        if definition.expression.type == ExpressionType.COMPOSITE:
            deps = self.METRIC_PATTERN.findall(definition.expression.sql)
            for dep in deps:
                dep_metric = await self._repo.find_by_name(dep)
                if dep_metric is None:
                    raise MetricError(
                        f"Dependent metric '{dep}' not found"
                    )

service MetricRegistryService {
  rpc RegisterMetric(RegisterMetricRequest) returns (MetricDefinition);
  rpc UpdateMetric(UpdateMetricRequest) returns (MetricDefinition);
  rpc DeleteMetric(DeleteMetricRequest) returns (google.protobuf.Empty);
  rpc GetMetric(GetMetricRequest) returns (MetricDefinition);
  rpc ListMetrics(ListMetricsRequest) returns (ListMetricsResponse);
  rpc ComputeMetric(ComputeMetricRequest) returns (MetricResult);
  rpc ComputeMetricBatch(ComputeMetricBatchRequest)
      returns (ComputeMetricBatchResponse);
}

class MetricObservability:
    """指标系统可观测性"""

    async def record_computation(
        self, metric_name: str, strategy: ComputationStrategy,
        latency_ms: float, success: bool,
    ) -> None:
        labels = {
            "metric": metric_name,
            "strategy": strategy.value,
            "success": str(success),
        }
        self._histogram.observe(latency_ms, labels)
        self._counter.inc(labels)

    async def get_strategy_stats(
        self, metric_name: str
    ) -> dict[str, StrategyStats]:
        """获取各策略的使用统计"""
        return {
            "REALTIME": StrategyStats(
                call_count=1200, avg_latency_ms=450,
                success_rate=0.98, cache_hit_rate=0.0
            ),
            "CACHED": StrategyStats(
                call_count=8500, avg_latency_ms=2,
                success_rate=0.99, cache_hit_rate=0.85
            ),
            "MATERIALIZED": StrategyStats(
                call_count=3200, avg_latency_ms=12,
                success_rate=0.97, cache_hit_rate=0.0
            ),
        }

User: SELECT name, metric('customer_gmv_30d') FROM Customer WHERE metric('customer_gmv_30d') > 50000

  |
  v
OqlMetricRewriter
  ├── Extract: ['customer_gmv_30d']
  ├── Load definition: gmv_30d (strategies=[MATERIALIZED, CACHED, REALTIME])
  ├── Select strategy: MATERIALIZED (MV exists + data fresh)
  └── Rewrite:
      SELECT c.name, mv.metric_value AS customer_gmv_30d
      FROM customer_objects c
      LEFT JOIN mv_customer_gmv_30d mv ON mv.object_id = c.object_id
      WHERE mv.metric_value > 50000
  |
  v
Doris Execution (< 20ms, index scan on MV)
  |
  v
Result: [{name: "Acme Corp", customer_gmv_30d: 125000}, ...]