+------------------------------------------------------------------+
|                    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 definition"""
    metric_id: str = Field(default_factory=lambda: str(uuid4()))
    name: str                          # unique name: monthly_active_customers
    display_name: str                  # display: Monthly Active Customers
    description: str | None = None
    object_type: str                   # owning object type: Customer
    category: MetricCategory           # classification
    data_type: MetricDataType          # return value type
    expression: MetricExpression       # computation expression
    strategies: list[ComputationStrategy]  # allowed strategies (priority ordered)
    freshness_requirement: FreshnessLevel  # freshness requirement
    cache_ttl: int = 300               # cache TTL (seconds)
    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 latency
    NEAR_REALTIME = "NEAR_REALTIME"  # < 1min
    HOURLY = "HOURLY"
    DAILY = "DAILY"
    WEEKLY = "WEEKLY"

class MetricExpression(BaseModel):
    """Metric computation expression"""
    type: ExpressionType
    sql: str | None = None           # SQL expression
    aggregation: AggregationType | None = None
    source_field: str | None = None  # source field
    filter_clause: str | None = None # filter condition
    group_by: list[str] = []         # group-by fields
    window: WindowSpec | None = None # window specification

class ExpressionType(str, Enum):
    SIMPLE_AGG = "SIMPLE_AGG"       # COUNT/SUM/AVG/MAX/MIN
    SQL = "SQL"                      # custom SQL
    COMPOSITE = "COMPOSITE"          # compose other metrics
    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"

# Example 1: Simple aggregation - customer total order count
order_count_metric = MetricDefinition(
    name="customer_order_count",
    display_name="Customer Total Orders",
    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,
)

# Example 2: SQL expression - customer 30-day GMV
gmv_30d_metric = MetricDefinition(
    name="customer_gmv_30d",
    display_name="Customer 30-Day 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,
)

# Example 3: Composite metric - average GMV per customer
avg_gmv_metric = MetricDefinition(
    name="average_customer_gmv",
    display_name="Average Customer 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):
    """Real-time computation executor"""

    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 virtual column executor"""

    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 computation executor"""

    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_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 registration example
@register_metric_udf("customer_health_score")
async def compute_health_score(
    data: MetricData, context: ComputeContext
) -> float:
    """Compute customer health score across multiple dimensions"""
    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-first computation executor"""

    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),
            )

        # Cache miss: fall back to real-time
        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

class RollupExecutor(StrategyExecutor):
    """Pre-aggregation executor"""

    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:
            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"

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):
    """Materialized view executor"""

    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(),
            strategy_used=ComputationStrategy.MATERIALIZED,
            latency_ms=0,
            is_stale=False,
        )

class ComputationCoordinator:
    """Metric computation coordinator with priority-based routing"""

    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:
        """Try computation strategies in priority order"""
        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 "
                    f"{metric.name}: {e}"
                )
                continue

        raise MetricComputeError(
            f"All strategies failed for metric {metric.name}: "
            + "; ".join(errors)
        )

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
  [4] REALTIME --> execute live query --> return

class BatchComputationCoordinator:
    """Batch metric computation"""

    async def compute_batch(
        self,
        metric: MetricDefinition,
        object_ids: list[str],
    ) -> dict[str, MetricResult]:
        # Batch cache lookup first
        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 real-time compute for cache misses
        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}

-- User-written 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

-- Rewritten SQL (strategy=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 rewriter"""

    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:
        """Rewrite metric() calls in OQL"""
        # 1. Extract all metric references
        metric_refs = self.METRIC_PATTERN.findall(oql)
        if not metric_refs:
            return oql

        # 2. Load metric definitions
        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. Determine optimal strategy for each metric
        strategy_map = {}
        for name, metric_def in metrics.items():
            strategy = await self._coordinator.select_strategy(
                metric_def, context
            )
            strategy_map[name] = strategy

        # 4. Generate rewritten SQL based on strategies
        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. Insert JOIN clauses
        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]:
        """Generate SQL replacement for a metric"""
        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:
            sub = metric.expression.sql.replace(
                "{object_id}", "main.object_id"
            )
            replacement = f"({sub})"
            return replacement, None

class MetricRegistryService:
    """Metric registration and management service"""

    async def register(
        self, definition: MetricDefinition
    ) -> MetricDefinition:
        """Register a new metric"""
        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)

        # Auto-create materialized view if strategy includes MATERIALIZED
        if ComputationStrategy.MATERIALIZED in definition.strategies:
            await self._mv_service.create_for_metric(definition)

        # Register rollup task if strategy includes ROLLUP
        if ComputationStrategy.ROLLUP in definition.strategies:
            await self._rollup_scheduler.register(definition)

        # Create virtual column if strategy includes 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:
        """Update a metric definition"""
        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:
        """Delete a metric"""
        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)

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:
    """Metric system observability"""

    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: 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}, ...]