RWM Console cluster: risingwave-adib.adib-rw.svc.cluster.local

← cluster insights objects user_account_membership_mv explain
Overview Objects Graph History
materialized view · insights.user_account_membership_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 lookupsAggregation state — unbounded unless keyed or temporally filtered
115 operators
Materialize · insights.user_account_membership_mv
0% idle 2 actors
Project
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · user_to_portfolios_mv.user_id = users_dm.user_id
2 actors
HashJoin · Inner · user_to_portfolios_mv.user_id = users_dm.user_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 · users_dm
2 actors
Filter · users_dm
0% idle 2 actors
StreamScan · users_dm
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project
2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Union
0% idle 2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = account_to_portfolios_…
2 actors
HashJoin · Inner · clients_portfolios_dm.portfolio_id = account_to_portfolios_… 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
SyncLogStore · Inner · open_accounts_mv.account_id = account_to_portfolios_dm.acco…
2 actors
HashJoin · Inner · open_accounts_mv.account_id = account_to_portfolios_dm.acco… 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 · account_to_portfolios_dm
2 actors
Filter · account_to_portfolios_dm
0% idle 2 actors
StreamScan · account_to_portfolios_dm
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
StreamScan · open_accounts_mv
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · user_to_clients_mv.client_id = clients_portfolios_dm.client…
2 actors
HashJoin · Inner · user_to_clients_mv.client_id = clients_portfolios_dm.client… 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 · clients_portfolios_dm
2 actors
Filter · clients_portfolios_dm
0% idle 2 actors
StreamScan · clients_portfolios_dm
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
StreamScan · user_to_clients_mv
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · accounts_to_clients_dm.account_id = open_accounts_mv.accoun…
2 actors
HashJoin · Inner · accounts_to_clients_dm.account_id = open_accounts_mv.accoun… 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
StreamScan · open_accounts_mv
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · user_to_clients_mv.client_id = accounts_to_clients_dm.clien…
2 actors
HashJoin · Inner · user_to_clients_mv.client_id = accounts_to_clients_dm.clien… 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 · accounts_to_clients_dm
2 actors
Filter · accounts_to_clients_dm
0% idle 2 actors
StreamScan · accounts_to_clients_dm
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
StreamScan · user_to_clients_mv
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · account_to_portfolios_dm.account_id = open_accounts_mv.acco…
2 actors
HashJoin · Inner · account_to_portfolios_dm.account_id = open_accounts_mv.acco… 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
StreamScan · open_accounts_mv
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · user_to_portfolios_mv.portfolio_id = account_to_portfolios_…
2 actors
HashJoin · Inner · user_to_portfolios_mv.portfolio_id = account_to_portfolios_… 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 · account_to_portfolios_dm
2 actors
Filter · account_to_portfolios_dm
0% idle 2 actors
StreamScan · account_to_portfolios_dm
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
StreamScan · user_to_portfolios_mv
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.user_account_membership_mv Materialize insights.user_account_m… idle · 2 actors Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · user_to_portfolios_mv.user_id = users_dm.user_id SyncLogStore Inner · user_to_portfol… — · 2 actors HashJoin · Inner · user_to_portfolios_mv.user_id = users_dm.user_id HashJoin Inner · user_to_portfol… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · users_dm Project users_dm — · 2 actors Filter · users_dm Filter users_dm idle · 2 actors StreamScan · users_dm StreamScan users_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Union Union idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = account_to_portfolios_… SyncLogStore Inner · clients_portfol… — · 2 actors HashJoin · Inner · clients_portfolios_dm.portfolio_id = account_to_portfolios_… HashJoin Inner · clients_portfol… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · open_accounts_mv.account_id = account_to_portfolios_dm.acco… SyncLogStore Inner · open_accounts_m… — · 2 actors HashJoin · Inner · open_accounts_mv.account_id = account_to_portfolios_dm.acco… HashJoin Inner · open_accounts_m… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · account_to_portfolios_dm Project account_to_portfolios_dm — · 2 actors Filter · account_to_portfolios_dm Filter account_to_portfolios_dm idle · 2 actors StreamScan · account_to_portfolios_dm StreamScan account_to_portfolios_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · open_accounts_mv StreamScan open_accounts_mv idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · user_to_clients_mv.client_id = clients_portfolios_dm.client… SyncLogStore Inner · user_to_clients… — · 2 actors HashJoin · Inner · user_to_clients_mv.client_id = clients_portfolios_dm.client… HashJoin Inner · user_to_clients… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · clients_portfolios_dm Project clients_portfolios_dm — · 2 actors Filter · clients_portfolios_dm Filter clients_portfolios_dm idle · 2 actors StreamScan · clients_portfolios_dm StreamScan clients_portfolios_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · user_to_clients_mv StreamScan user_to_clients_mv idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · accounts_to_clients_dm.account_id = open_accounts_mv.accoun… SyncLogStore Inner · accounts_to_cli… — · 2 actors HashJoin · Inner · accounts_to_clients_dm.account_id = open_accounts_mv.accoun… HashJoin Inner · accounts_to_cli… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · open_accounts_mv StreamScan open_accounts_mv idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · user_to_clients_mv.client_id = accounts_to_clients_dm.clien… SyncLogStore Inner · user_to_clients… — · 2 actors HashJoin · Inner · user_to_clients_mv.client_id = accounts_to_clients_dm.clien… HashJoin Inner · user_to_clients… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · accounts_to_clients_dm Project accounts_to_clients_dm — · 2 actors Filter · accounts_to_clients_dm Filter accounts_to_clients_dm idle · 2 actors StreamScan · accounts_to_clients_dm StreamScan accounts_to_clients_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · user_to_clients_mv StreamScan user_to_clients_mv idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · account_to_portfolios_dm.account_id = open_accounts_mv.acco… SyncLogStore Inner · account_to_port… — · 2 actors HashJoin · Inner · account_to_portfolios_dm.account_id = open_accounts_mv.acco… HashJoin Inner · account_to_port… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · open_accounts_mv StreamScan open_accounts_mv idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · user_to_portfolios_mv.portfolio_id = account_to_portfolios_… SyncLogStore Inner · user_to_portfol… — · 2 actors HashJoin · Inner · user_to_portfolios_mv.portfolio_id = account_to_portfolios_… HashJoin Inner · user_to_portfol… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · account_to_portfolios_dm Project account_to_portfolios_dm — · 2 actors Filter · account_to_portfolios_dm Filter account_to_portfolios_dm idle · 2 actors StreamScan · account_to_portfolios_dm StreamScan account_to_portfolios_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · user_to_portfolios_mv StreamScan user_to_portfolios_mv 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 13364 (Actor 99323,99322)
StreamMaterialize { columns: [user_id, account_id, type], stream_key: [user_id, account_id], pk_columns: [user_id, account_id], pk_conflict: NoCheck }
├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, 'all':Varchar ]
├── stream key: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
└── StreamProject { exprs: [user_to_portfolios_mv.user_id, open_accounts_mv.account_id, 'all':Varchar] }
    ├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, 'all':Varchar ]
    ├── stream key: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
    └── MergeExecutor
        ├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, users_dm.user_id ]
        └── stream key: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]

Fragment 13365 (Actor 99324,99325)
StreamSyncLogStore
├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, users_dm.user_id ]
├── stream key: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
└── StreamHashJoin { type: Inner, predicate: user_to_portfolios_mv.user_id = users_dm.user_id }
    ├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, users_dm.user_id ]
    ├── stream key: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
    ├── MergeExecutor
    │   ├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
    │   └── stream key: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
    └── MergeExecutor { output: [ users_dm.user_id ], stream key: [ users_dm.user_id ] }

Fragment 13366 (Actor 99329,99328)
StreamProject { exprs: [user_to_portfolios_mv.user_id, open_accounts_mv.account_id] }
├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
├── stream key: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
└── StreamHashAgg { group_key: [user_to_portfolios_mv.user_id, open_accounts_mv.account_id], aggs: [count] }
    ├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, count ]
    ├── stream key: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id ]
    └── MergeExecutor
        ├── output:
        │   ┌── user_to_portfolios_mv.user_id
        │   ├── open_accounts_mv.account_id
        │   ├── user_to_portfolios_mv.user_id
        │   ├── user_to_portfolios_mv.portfolio_id
        │   ├── account_to_portfolios_dm.account_id
        │   ├── null:Varchar
        │   ├── account_to_portfolios_dm.effective_start_date
        │   ├── null:Date
        │   └── $src
        └── stream key:
            ┌── user_to_portfolios_mv.user_id
            ├── user_to_portfolios_mv.portfolio_id
            ├── account_to_portfolios_dm.account_id
            ├── account_to_portfolios_dm.effective_start_date
            ├── null:Varchar
            ├── null:Date
            └── $src

Fragment 13367 (Actor 99337,99338)
StreamUnion { all: true }
├── output:
│   ┌── user_to_portfolios_mv.user_id
│   ├── open_accounts_mv.account_id
│   ├── user_to_portfolios_mv.user_id
│   ├── user_to_portfolios_mv.portfolio_id
│   ├── account_to_portfolios_dm.account_id
│   ├── null:Varchar
│   ├── account_to_portfolios_dm.effective_start_date
│   ├── null:Date
│   └── $src
├── stream key:
│   ┌── user_to_portfolios_mv.user_id
│   ├── user_to_portfolios_mv.portfolio_id
│   ├── account_to_portfolios_dm.account_id
│   ├── account_to_portfolios_dm.effective_start_date
│   ├── null:Varchar
│   ├── null:Date
│   └── $src
├── MergeExecutor
│   ├── output:
│   │   ┌── user_to_portfolios_mv.user_id
│   │   ├── open_accounts_mv.account_id
│   │   ├── user_to_portfolios_mv.user_id
│   │   ├── user_to_portfolios_mv.portfolio_id
│   │   ├── account_to_portfolios_dm.account_id
│   │   ├── null:Varchar
│   │   ├── account_to_portfolios_dm.effective_start_date
│   │   ├── null:Date
│   │   └── 0:Int32
│   └── stream key:
│       ┌── user_to_portfolios_mv.user_id
│       ├── user_to_portfolios_mv.portfolio_id
│       ├── account_to_portfolios_dm.account_id
│       └── account_to_portfolios_dm.effective_start_date
├── MergeExecutor
│   ├── output:
│   │   ┌── user_to_clients_mv.user_id
│   │   ├── open_accounts_mv.account_id
│   │   ├── user_to_clients_mv.user_id
│   │   ├── user_to_clients_mv.client_id
│   │   ├── accounts_to_clients_dm.account_id
│   │   ├── null:Varchar
│   │   ├── accounts_to_clients_dm.effective_start_date
│   │   ├── null:Date
│   │   └── 1:Int32
│   └── stream key:
│       ┌── user_to_clients_mv.user_id
│       ├── user_to_clients_mv.client_id
│       ├── accounts_to_clients_dm.account_id
│       └── accounts_to_clients_dm.effective_start_date
└── MergeExecutor
    ├── output:
    │   ┌── user_to_clients_mv.user_id
    │   ├── open_accounts_mv.account_id
    │   ├── user_to_clients_mv.user_id
    │   ├── user_to_clients_mv.client_id
    │   ├── clients_portfolios_dm.portfolio_id
    │   ├── open_accounts_mv.account_id
    │   ├── clients_portfolios_dm.effective_start_date
    │   ├── account_to_portfolios_dm.effective_start_date
    │   └── 2:Int32
    └── stream key:
        ┌── user_to_clients_mv.user_id
        ├── user_to_clients_mv.client_id
        ├── clients_portfolios_dm.portfolio_id
        ├── clients_portfolios_dm.effective_start_date
        ├── open_accounts_mv.account_id
        └── account_to_portfolios_dm.effective_start_date

Fragment 13368 (Actor 99349,99348)
StreamProject { exprs: [user_to_portfolios_mv.user_id, open_accounts_mv.account_id, user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, null:Varchar, account_to_portfolios_dm.effective_start_date, null:Date, 0:Int32] }
├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, null:Varchar, account_to_portfolios_dm.effective_start_date, null:Date, 0:Int32 ]
├── stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
└── MergeExecutor
    ├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
    └── stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]

Fragment 13369 (Actor 99347,99346)
StreamSyncLogStore
├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.account_id = open_accounts_mv.account_id }
    ├── output: [ user_to_portfolios_mv.user_id, open_accounts_mv.account_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
    ├── stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ user_to_portfolios_mv.user_id, account_to_portfolios_dm.account_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    │   └── stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
    └── MergeExecutor { output: [ open_accounts_mv.account_id ], stream key: [ open_accounts_mv.account_id ] }

Fragment 13370 (Actor 99351,99350)
StreamSyncLogStore
├── output: [ user_to_portfolios_mv.user_id, account_to_portfolios_dm.account_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: user_to_portfolios_mv.portfolio_id = account_to_portfolios_dm.portfolio_id }
    ├── output: [ user_to_portfolios_mv.user_id, account_to_portfolios_dm.account_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor { output: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id ], stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id ] }
    └── MergeExecutor
        ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
        └── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]

Fragment 13371 (Actor 99394,99393)
StreamTableScan { table: user_to_portfolios_mv, columns: [user_id, portfolio_id] } { output: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id ], stream key: [ user_to_portfolios_mv.user_id, user_to_portfolios_mv.portfolio_id ] }
├── Upstream { output: [ user_id, portfolio_id ], stream key: [] }
└── BatchPlanNode { output: [ user_id, portfolio_id ], stream key: [] }

Fragment 13372 (Actor 99396,99395)
StreamProject { exprs: [account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date] }
├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── StreamFilter { predicate: IsNull(account_to_portfolios_dm.disabled_at) }
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, account_to_portfolios_dm.disabled_at ]
    ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    └── StreamTableScan { table: account_to_portfolios_dm, columns: [account_id, portfolio_id, effective_start_date, disabled_at] }
        ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, account_to_portfolios_dm.disabled_at ]
        ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
        ├── Upstream { output: [ account_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 13373 (Actor 99414,99413)
StreamTableScan { table: open_accounts_mv, columns: [account_id] } { output: [ open_accounts_mv.account_id ], stream key: [ open_accounts_mv.account_id ] }
├── Upstream { output: [ account_id ], stream key: [] }
└── BatchPlanNode { output: [ account_id ], stream key: [] }

Fragment 13374 (Actor 99355,99356)
StreamProject { exprs: [user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, null:Varchar, accounts_to_clients_dm.effective_start_date, null:Date, 1:Int32] }
├── output: [ user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, null:Varchar, accounts_to_clients_dm.effective_start_date, null:Date, 1:Int32 ]
├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
└── MergeExecutor
    ├── output: [ user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
    └── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]

Fragment 13375 (Actor 99357,99358)
StreamSyncLogStore
├── output: [ user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.account_id = open_accounts_mv.account_id }
    ├── output: [ user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
    ├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ user_to_clients_mv.user_id, accounts_to_clients_dm.account_id, user_to_clients_mv.client_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
    │   └── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
    └── MergeExecutor { output: [ open_accounts_mv.account_id ], stream key: [ open_accounts_mv.account_id ] }

Fragment 13376 (Actor 99360,99359)
StreamSyncLogStore
├── output: [ user_to_clients_mv.user_id, accounts_to_clients_dm.account_id, user_to_clients_mv.client_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: user_to_clients_mv.client_id = accounts_to_clients_dm.client_id }
    ├── output: [ user_to_clients_mv.user_id, accounts_to_clients_dm.account_id, user_to_clients_mv.client_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
    ├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
    ├── MergeExecutor { output: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id ], stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id ] }
    └── MergeExecutor
        ├── output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
        └── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]

Fragment 13377 (Actor 99416,99415)
StreamTableScan { table: user_to_clients_mv, columns: [user_id, client_id] } { output: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id ], stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id ] }
├── Upstream { output: [ user_id, client_id ], stream key: [] }
└── BatchPlanNode { output: [ user_id, client_id ], stream key: [] }

Fragment 13378 (Actor 99419,99418)
StreamProject { exprs: [accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date] }
├── output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
└── StreamFilter { predicate: IsNull(accounts_to_clients_dm.disabled_at) }
    ├── output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_to_clients_dm.disabled_at ]
    ├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
    └── StreamTableScan { table: accounts_to_clients_dm, columns: [account_id, client_id, effective_start_date, disabled_at] }
        ├── output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_to_clients_dm.disabled_at ]
        ├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
        ├── Upstream { output: [ account_id, client_id, effective_start_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, client_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 13379 (Actor 99421,99420)
StreamTableScan { table: open_accounts_mv, columns: [account_id] } { output: [ open_accounts_mv.account_id ], stream key: [ open_accounts_mv.account_id ] }
├── Upstream { output: [ account_id ], stream key: [] }
└── BatchPlanNode { output: [ account_id ], stream key: [] }

Fragment 13380 (Actor 99380,99381)
StreamProject { exprs: [user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, open_accounts_mv.account_id, clients_portfolios_dm.effective_start_date, account_to_portfolios_dm.effective_start_date, 2:Int32] }
├── output: [ user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, open_accounts_mv.account_id, clients_portfolios_dm.effective_start_date, account_to_portfolios_dm.effective_start_date, 2:Int32 ]
├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, open_accounts_mv.account_id, account_to_portfolios_dm.effective_start_date ]
└── MergeExecutor
    ├── output: [ user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    └── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, open_accounts_mv.account_id, account_to_portfolios_dm.effective_start_date ]

Fragment 13381 (Actor 99378,99379)
StreamSyncLogStore
├── output: [ user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, open_accounts_mv.account_id, account_to_portfolios_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.portfolio_id = account_to_portfolios_dm.portfolio_id }
    ├── output: [ user_to_clients_mv.user_id, open_accounts_mv.account_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, open_accounts_mv.account_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ user_to_clients_mv.user_id, clients_portfolios_dm.portfolio_id, user_to_clients_mv.client_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ]
    │   └── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
    └── MergeExecutor
        ├── output: [ open_accounts_mv.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
        └── stream key: [ open_accounts_mv.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]

Fragment 13382 (Actor 99383,99382)
StreamSyncLogStore
├── output: [ user_to_clients_mv.user_id, clients_portfolios_dm.portfolio_id, user_to_clients_mv.client_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ]
├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: user_to_clients_mv.client_id = clients_portfolios_dm.client_id }
    ├── output: [ user_to_clients_mv.user_id, clients_portfolios_dm.portfolio_id, user_to_clients_mv.client_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ]
    ├── stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
    ├── MergeExecutor { output: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id ], stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id ] }
    └── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 13383 (Actor 99454,99455)
StreamTableScan { table: user_to_clients_mv, columns: [user_id, client_id] } { output: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id ], stream key: [ user_to_clients_mv.user_id, user_to_clients_mv.client_id ] }
├── Upstream { output: [ user_id, client_id ], stream key: [] }
└── BatchPlanNode { output: [ user_id, client_id ], stream key: [] }

Fragment 13384 (Actor 99536,99535)
StreamProject { exprs: [clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date] }
├── output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
└── StreamFilter { predicate: IsNull(clients_portfolios_dm.disabled_at) }
    ├── output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at ]
    ├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
    └── StreamTableScan { table: clients_portfolios_dm, columns: [client_id, portfolio_id, effective_start_date, disabled_at] }
        ├── output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at ]
        ├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
        ├── Upstream { output: [ client_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ client_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 13385 (Actor 99387,99386)
StreamSyncLogStore
├── output: [ open_accounts_mv.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ open_accounts_mv.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: open_accounts_mv.account_id = account_to_portfolios_dm.account_id }
    ├── output: [ open_accounts_mv.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
    ├── stream key: [ open_accounts_mv.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor { output: [ open_accounts_mv.account_id ], stream key: [ open_accounts_mv.account_id ] }
    └── MergeExecutor { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 13386 (Actor 99545,99546)
StreamTableScan { table: open_accounts_mv, columns: [account_id] } { output: [ open_accounts_mv.account_id ], stream key: [ open_accounts_mv.account_id ] }
├── Upstream { output: [ account_id ], stream key: [] }
└── BatchPlanNode { output: [ account_id ], stream key: [] }

Fragment 13387 (Actor 99552,99551)
StreamProject { exprs: [account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date] }
├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── StreamFilter { predicate: IsNull(account_to_portfolios_dm.disabled_at) }
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, account_to_portfolios_dm.disabled_at ]
    ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    └── StreamTableScan { table: account_to_portfolios_dm, columns: [account_id, portfolio_id, effective_start_date, disabled_at] }
        ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, account_to_portfolios_dm.disabled_at ]
        ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
        ├── Upstream { output: [ account_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 13388 (Actor 99561,99562)
StreamProject { exprs: [users_dm.user_id] } { output: [ users_dm.user_id ], stream key: [ users_dm.user_id ] }
└── StreamFilter { predicate: IsNull(users_dm.disabled_at) } { output: [ users_dm.user_id, users_dm.disabled_at ], stream key: [ users_dm.user_id ] }
    └── StreamTableScan { table: users_dm, columns: [user_id, disabled_at] } { output: [ users_dm.user_id, users_dm.disabled_at ], stream key: [ users_dm.user_id ] }
        ├── Upstream { output: [ user_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ user_id, disabled_at ], stream key: [] }