RWM Console cluster: risingwave.platform.svc.cluster.local

← cluster insights objects benchmark_twrr_mv explain
Overview Objects Graph History
materialized view · insights.benchmark_twrr_mv profiled over 5s
seconds (1–30)

Job is idle — throughput ~0; structure shown.

Stateful hash join (4 state tables) — consider a temporal join for dimension lookupsWindow state — add a WHERE rank <= N to bound it
17 operators
Materialize · insights.benchmark_twrr_mv
0% idle 2 actors
Project · Inner · Not(IsNull(first_value)) AND (first_value <> 0:Decimal)
— — 2 actors
Filter · Inner · Not(IsNull(first_value)) AND (first_value <> 0:Decimal)
0% idle 2 actors
OverWindow · Inner · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) Window state — add a WHERE rank <= N to bound it
0% idle 2 actors
HashJoin · Inner · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) Stateful hash join (4 state tables) — consider a temporal join for dimension lookups
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · benchmarks_dm
— — 2 actors
Filter · benchmarks_dm
0% idle 2 actors
StreamScan · benchmarks_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
StreamScan · benchmark_values_ft
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Heat = the operator's output-buffer backpressure over the sampling window. Click a node to fold its subtree.
Materialize · insights.benchmark_twrr_mv Materialize insights.benchmark_twrr… idle · 2 actors Project · Inner · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) Project Inner · Not(IsNull(firs… — · 2 actors Filter · Inner · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) Filter Inner · Not(IsNull(firs… idle · 2 actors OverWindow · Inner · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) OverWindow Inner · Not(IsNull(firs… idle · 2 actors HashJoin · Inner · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) HashJoin Inner · Not(IsNull(firs… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · benchmarks_dm Project benchmarks_dm — · 2 actors Filter · benchmarks_dm Filter benchmarks_dm idle · 2 actors StreamScan · benchmarks_dm StreamScan benchmarks_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · benchmark_values_ft StreamScan benchmark_values_ft idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors
Streaming operator plan from EXPLAIN ANALYZE. Node heat = backpressure. Drag to pan, scroll to zoom.
Fragments (DESCRIBE FRAGMENTS) — click to expand 
Fragment 11092 (Actor 32802,32801)
StreamMaterialize { columns: [benchmark_id, fact_date, daily_subperiod_return], stream_key: [benchmark_id, fact_date], pk_columns: [benchmark_id, fact_date], pk_conflict: NoCheck }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, $expr1 ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
└── StreamProject { exprs: [benchmark_values_ft.benchmark_id, benchmark_values_ft.date, ((benchmark_values_ft.amount - first_value) / first_value) as $expr1] }
    ├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, $expr1 ]
    ├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
    └── StreamFilter { predicate: Not(IsNull(first_value)) AND (first_value <> 0:Decimal) }
        ├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id, first_value ]
        ├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
        └── StreamOverWindow { window_functions: [first_value(benchmark_values_ft.amount) OVER(PARTITION BY benchmark_values_ft.benchmark_id ORDER BY benchmark_values_ft.date ASC ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING)] }
            ├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id, first_value ]
            ├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
            └── StreamHashJoin { type: Inner, predicate: benchmark_values_ft.benchmark_id = benchmarks_dm.id }
                ├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id ]
                ├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
                ├── MergeExecutor { output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount ], stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ] }
                └── MergeExecutor { output: [ benchmarks_dm.id ], stream key: [ benchmarks_dm.id ] }

Fragment 11093 (Actor 32805,32806)
StreamTableScan { table: benchmark_values_ft, columns: [benchmark_id, date, amount] }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
├── Upstream { output: [ benchmark_id, date, amount ], stream key: [] }
└── BatchPlanNode { output: [ benchmark_id, date, amount ], stream key: [] }

Fragment 11094 (Actor 32927,32926)
StreamProject { exprs: [benchmarks_dm.id] } { output: [ benchmarks_dm.id ], stream key: [ benchmarks_dm.id ] }
└── StreamFilter { predicate: IsNull(benchmarks_dm.disabled_at) } { output: [ benchmarks_dm.id, benchmarks_dm.disabled_at ], stream key: [ benchmarks_dm.id ] }
    └── StreamTableScan { table: benchmarks_dm, columns: [id, disabled_at] } { output: [ benchmarks_dm.id, benchmarks_dm.disabled_at ], stream key: [ benchmarks_dm.id ] }
        ├── Upstream { output: [ id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ id, disabled_at ], stream key: [] }