RWM Console cluster: risingwave-alinma.alinma-rw.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
24 operators
Materialize · insights.benchmark_twrr_mv
0% idle 2 actors
Project · Not(IsNull(first_value)) AND (first_value <> 0:Decimal)
— — 2 actors
Filter · Not(IsNull(first_value)) AND (first_value <> 0:Decimal)
0% idle 2 actors
OverWindow · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) Window state — add a WHERE rank <= N to bound it
0% idle 2 actors
LocalityProvider · Not(IsNull(first_value)) AND (first_value <> 0:Decimal)
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · benchmark_values_ft.benchmark_id = benchmarks_dm.id
— — 2 actors
HashJoin · Inner · benchmark_values_ft.benchmark_id = benchmarks_dm.id 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
LocalityProvider
0% idle 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 · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) Project Not(IsNull(first_value)… — · 2 actors Filter · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) Filter Not(IsNull(first_value)… idle · 2 actors OverWindow · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) OverWindow Not(IsNull(first_value)… idle · 2 actors LocalityProvider · Not(IsNull(first_value)) AND (first_value <> 0:Decimal) LocalityProvider Not(IsNull(first_value)… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · benchmark_values_ft.benchmark_id = benchmarks_dm.id SyncLogStore Inner · benchmark_value… — · 2 actors HashJoin · Inner · benchmark_values_ft.benchmark_id = benchmarks_dm.id HashJoin Inner · benchmark_value… 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 LocalityProvider LocalityProvider idle · 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 62538 (Actor 743030,743029)
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 ]
            └── StreamLocalityProvider { locality_columns: [benchmark_values_ft.benchmark_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, benchmarks_dm.id ]
                    └── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]

Fragment 62539 (Actor 743033,743034)
StreamSyncLogStore { 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 ] }
└── 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 62540 (Actor 743036,743035)
StreamLocalityProvider { locality_columns: [benchmark_values_ft.benchmark_id] }
├── 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: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount ], stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ] }

Fragment 62541 (Actor 743017,743018)
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 62542 (Actor 743019,743020)
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: [] }