Repository: risingwavelabs/risingwave
File: grafana/dashboard/dev/streaming_relations.py (Python)

Code before:
    )


def _sum_fragment_metric_by_mv(expr: str) -> str:
    return (
        f"sum(({expr})"
        f"* on(fragment_id) group_left(materialized_view_id)"
        f"max by (fragment_id, materialized_view_id) ({metric('table_info')}))"
        f"by (materialized_view_id)"
    )

@section
def _(outer_panels: Panels):
    panels = outer_panels.sub_panel()
    actor_count_expr = (
        f"clamp_min(sum({metric('stream_actor_count')}) by (fragment_id), 1)"
    )
    poll_duration_rate_expr = (
        f"sum(rate({metric('stream_actor_poll_duration')}[$__rate_interval])) by (fragment_id)"
    )
    poll_duration_expr = (
        f"{poll_duration_rate_expr} / on(fragment_id) {actor_count_expr}"
    )
    zero_mv_expr = f"0 * max by (materialized_view_id) ({metric('table_info')})"
    def _coalesce_mv(expr: str) -> str:
        return f"(({expr}) or on(materialized_view_id) {zero_mv_expr})"

    kv_log_store_buffer_usage_by_fragment_expr = (
        f"sum({metric('kv_log_store_buffer_memory_bytes')}"
        f"  * on(actor_id) group_left(fragment_id) {metric('actor_info')})"
        f" by (fragment_id)"
    )
    sync_kv_log_store_buffer_usage_by_fragment_expr = (
        f"sum({metric('sync_kv_log_store_buffer_memory_bytes')}) by (fragment_id)"
    )
    executor_cache_usage_expr = (
        f"sum({metric('stream_memory_usage')} * on(table_id) group_left(materialized_view_id) "
        f"{metric('table_info')}) by (materialized_view_id)"
    )
    shared_buffer_usage_expr = _sum_fragment_metric_by_mv(
        metric("state_store_per_table_imm_size")
    )
    kv_log_store_buffer_usage_expr = _sum_fragment_metric_by_mv(
        kv_log_store_buffer_usage_by_fragment_expr
    )
    sync_kv_log_store_buffer_usage_expr = _sum_fragment_metric_by_mv(
        sync_kv_log_store_buffer_usage_by_fragment_expr
    )
    total_memory_usage_expr = " + ".join(
        map(
            _coalesce_mv,

Review comment (suggestion):
The variable name `zero_mv_expr` is ambiguous. Consider renaming to `zero_for_all_mvs_expr` or `mv_zero_baseline_expr` to better convey that it creates a zero value for all materialized view IDs.
```suggestion
    zero_for_all_mvs_expr = f"0 * max by (materialized_view_id) ({metric('table_info')})"
    def _coalesce_mv(expr: str) -> str:
        return f"(({expr}) or on(materialized_view_id) {zero_for_all_mvs_expr})"
```

Code after:
    )
    return f"topk(10, (100 * ({peak_expr})))"

def _relation_busy_rate_target(panels: Panels, rate_interval: str):
    return panels.target(
        _relation_busy_rate_expr(rate_interval),
        "name {{name}} id {{id}} type {{type}} fragment {{fragment_id}}",
    )


def _sum_fragment_metric_by_mv(expr: str) -> str:
    return (
        f"sum(({expr})"
        f"* on(fragment_id) group_left(materialized_view_id)"
        f"max by (fragment_id, materialized_view_id) ({metric('table_info')}))"
        f"by (materialized_view_id)"
    )

@section
def _(outer_panels: Panels):
    panels = outer_panels.sub_panel()
    actor_count_expr = (
        f"clamp_min(sum({metric('stream_actor_count')}) by (fragment_id), 1)"
    )
    poll_duration_rate_expr = (
        f"sum(rate({metric('stream_actor_poll_duration')}[$__rate_interval])) by (fragment_id)"
    )
    poll_duration_expr = (
        f"{poll_duration_rate_expr} / on(fragment_id) {actor_count_expr}"
    )
    zero_for_all_mvs_expr = f"0 * max by (materialized_view_id) ({metric('table_info')})"
    def _coalesce_mv(expr: str) -> str:
        return f"(({expr}) or on(materialized_view_id) {zero_for_all_mvs_expr})"
    executor_cache_usage_expr = (
        f"sum({metric('stream_memory_usage')} * on(table_id) group_left(materialized_view_id) "
        f"{metric('table_info')}) by (materialized_view_id)"
    )
    shared_buffer_usage_expr = _sum_fragment_metric_by_mv(
        metric("state_store_per_table_imm_size")
    )
    kv_log_store_buffer_usage_expr = _sum_fragment_metric_by_mv(
        _kv_log_store_buffer_usage_by_fragment_expr()
    )
    sync_kv_log_store_buffer_usage_expr = _sum_fragment_metric_by_mv(
        _sync_kv_log_store_buffer_usage_by_fragment_expr()
    )
    total_memory_usage_expr = " + ".join(
        map(
            _coalesce_mv,
            [
                executor_cache_usage_expr,