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

← cluster search objects active_identifier_edges_mv explain
Overview Objects Graph History
materialized view · search.active_identifier_edges_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 lookups
221 operators
Materialize · search.active_identifier_edges_mv
0% idle 2 actors
Union
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · party_holder_edges_mv_next
— — 2 actors
StreamScan · party_holder_edges_mv_next
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 = accounts_dm.account_id
— — 2 actors
HashJoin · Inner · account_to_portfolios_dm.account_id = accounts_dm.account_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 · accounts_dm
— — 2 actors
Filter · accounts_dm
0% idle 2 actors
StreamScan · accounts_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · account_to_portfolios_dm.portfolio_id = portfolios_dm.portf…
— — 2 actors
HashJoin · Inner · account_to_portfolios_dm.portfolio_id = portfolios_dm.portf… 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 · portfolios_dm
— — 2 actors
Filter · portfolios_dm
0% idle 2 actors
StreamScan · portfolios_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 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
Project
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · clients_portfolios_dm.client_id = clients_dm.id
— — 2 actors
HashJoin · Inner · clients_portfolios_dm.client_id = clients_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 · clients_dm
— — 2 actors
Filter · clients_dm
0% idle 2 actors
StreamScan · clients_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli…
— — 2 actors
HashJoin · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli… 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 · portfolios_dm
— — 2 actors
Filter · portfolios_dm
0% idle 2 actors
StreamScan · portfolios_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 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
Project
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli…
— — 2 actors
HashJoin · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli… 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 · portfolios_dm
— — 2 actors
Filter · portfolios_dm
0% idle 2 actors
StreamScan · portfolios_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · clients_portfolios_dm.client_id = clients_dm.id
— — 2 actors
HashJoin · Inner · clients_portfolios_dm.client_id = clients_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 · clients_dm
— — 2 actors
Filter · clients_dm
0% idle 2 actors
StreamScan · clients_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 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
Project
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_id
— — 2 actors
HashJoin · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_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 · accounts_dm
— — 2 actors
Filter · accounts_dm
0% idle 2 actors
StreamScan · accounts_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · accounts_to_clients_dm.client_id = clients_dm.id
— — 2 actors
HashJoin · Inner · accounts_to_clients_dm.client_id = clients_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 · clients_dm
— — 2 actors
Filter · clients_dm
0% idle 2 actors
StreamScan · clients_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 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
Project
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · account_to_portfolios_dm.portfolio_id = portfolios_dm.portf…
— — 2 actors
HashJoin · Inner · account_to_portfolios_dm.portfolio_id = portfolios_dm.portf… 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 · portfolios_dm
— — 2 actors
Filter · portfolios_dm
0% idle 2 actors
StreamScan · portfolios_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · account_to_portfolios_dm.account_id = accounts_dm.account_id
— — 2 actors
HashJoin · Inner · account_to_portfolios_dm.account_id = accounts_dm.account_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 · accounts_dm
— — 2 actors
Filter · accounts_dm
0% idle 2 actors
StreamScan · accounts_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 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
Project
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · accounts_to_clients_dm.client_id = clients_dm.id
— — 2 actors
HashJoin · Inner · accounts_to_clients_dm.client_id = clients_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 · clients_dm
— — 2 actors
Filter · clients_dm
0% idle 2 actors
StreamScan · clients_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_id
— — 2 actors
HashJoin · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_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 · accounts_dm
— — 2 actors
Filter · accounts_dm
0% idle 2 actors
StreamScan · accounts_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 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
Project · portfolios_dm
— — 2 actors
Filter · portfolios_dm
0% idle 2 actors
StreamScan · portfolios_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · clients_dm
— — 2 actors
Filter · clients_dm
0% idle 2 actors
StreamScan · clients_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · accounts_dm
— — 2 actors
Filter · accounts_dm
0% idle 2 actors
StreamScan · accounts_dm
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 · search.active_identifier_edges_mv Materialize search.active_identifie… idle · 2 actors Union Union idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · party_holder_edges_mv_next Project party_holder_edges_mv_n… — · 2 actors StreamScan · party_holder_edges_mv_next StreamScan party_holder_edges_mv_n… 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 = accounts_dm.account_id SyncLogStore Inner · account_to_port… — · 2 actors HashJoin · Inner · account_to_portfolios_dm.account_id = accounts_dm.account_id HashJoin Inner · account_to_port… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · accounts_dm Project accounts_dm — · 2 actors Filter · accounts_dm Filter accounts_dm idle · 2 actors StreamScan · accounts_dm StreamScan accounts_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · account_to_portfolios_dm.portfolio_id = portfolios_dm.portf… SyncLogStore Inner · account_to_port… — · 2 actors HashJoin · Inner · account_to_portfolios_dm.portfolio_id = portfolios_dm.portf… HashJoin Inner · account_to_port… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · portfolios_dm Project portfolios_dm — · 2 actors Filter · portfolios_dm Filter portfolios_dm idle · 2 actors StreamScan · portfolios_dm StreamScan portfolios_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 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 Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · clients_portfolios_dm.client_id = clients_dm.id SyncLogStore Inner · clients_portfol… — · 2 actors HashJoin · Inner · clients_portfolios_dm.client_id = clients_dm.id HashJoin Inner · clients_portfol… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · clients_dm Project clients_dm — · 2 actors Filter · clients_dm Filter clients_dm idle · 2 actors StreamScan · clients_dm StreamScan clients_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli… SyncLogStore Inner · clients_portfol… — · 2 actors HashJoin · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli… HashJoin Inner · clients_portfol… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · portfolios_dm Project portfolios_dm — · 2 actors Filter · portfolios_dm Filter portfolios_dm idle · 2 actors StreamScan · portfolios_dm StreamScan portfolios_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 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 Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli… SyncLogStore Inner · clients_portfol… — · 2 actors HashJoin · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli… HashJoin Inner · clients_portfol… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · portfolios_dm Project portfolios_dm — · 2 actors Filter · portfolios_dm Filter portfolios_dm idle · 2 actors StreamScan · portfolios_dm StreamScan portfolios_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · clients_portfolios_dm.client_id = clients_dm.id SyncLogStore Inner · clients_portfol… — · 2 actors HashJoin · Inner · clients_portfolios_dm.client_id = clients_dm.id HashJoin Inner · clients_portfol… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · clients_dm Project clients_dm — · 2 actors Filter · clients_dm Filter clients_dm idle · 2 actors StreamScan · clients_dm StreamScan clients_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 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 Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_id SyncLogStore Inner · accounts_to_cli… — · 2 actors HashJoin · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_id HashJoin Inner · accounts_to_cli… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · accounts_dm Project accounts_dm — · 2 actors Filter · accounts_dm Filter accounts_dm idle · 2 actors StreamScan · accounts_dm StreamScan accounts_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · accounts_to_clients_dm.client_id = clients_dm.id SyncLogStore Inner · accounts_to_cli… — · 2 actors HashJoin · Inner · accounts_to_clients_dm.client_id = clients_dm.id HashJoin Inner · accounts_to_cli… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · clients_dm Project clients_dm — · 2 actors Filter · clients_dm Filter clients_dm idle · 2 actors StreamScan · clients_dm StreamScan clients_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 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 Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · account_to_portfolios_dm.portfolio_id = portfolios_dm.portf… SyncLogStore Inner · account_to_port… — · 2 actors HashJoin · Inner · account_to_portfolios_dm.portfolio_id = portfolios_dm.portf… HashJoin Inner · account_to_port… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · portfolios_dm Project portfolios_dm — · 2 actors Filter · portfolios_dm Filter portfolios_dm idle · 2 actors StreamScan · portfolios_dm StreamScan portfolios_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · account_to_portfolios_dm.account_id = accounts_dm.account_id SyncLogStore Inner · account_to_port… — · 2 actors HashJoin · Inner · account_to_portfolios_dm.account_id = accounts_dm.account_id HashJoin Inner · account_to_port… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · accounts_dm Project accounts_dm — · 2 actors Filter · accounts_dm Filter accounts_dm idle · 2 actors StreamScan · accounts_dm StreamScan accounts_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 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 Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · accounts_to_clients_dm.client_id = clients_dm.id SyncLogStore Inner · accounts_to_cli… — · 2 actors HashJoin · Inner · accounts_to_clients_dm.client_id = clients_dm.id HashJoin Inner · accounts_to_cli… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · clients_dm Project clients_dm — · 2 actors Filter · clients_dm Filter clients_dm idle · 2 actors StreamScan · clients_dm StreamScan clients_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_id SyncLogStore Inner · accounts_to_cli… — · 2 actors HashJoin · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_id HashJoin Inner · accounts_to_cli… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · accounts_dm Project accounts_dm — · 2 actors Filter · accounts_dm Filter accounts_dm idle · 2 actors StreamScan · accounts_dm StreamScan accounts_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 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 Project · portfolios_dm Project portfolios_dm — · 2 actors Filter · portfolios_dm Filter portfolios_dm idle · 2 actors StreamScan · portfolios_dm StreamScan portfolios_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · clients_dm Project clients_dm — · 2 actors Filter · clients_dm Filter clients_dm idle · 2 actors StreamScan · clients_dm StreamScan clients_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · accounts_dm Project accounts_dm — · 2 actors Filter · accounts_dm Filter accounts_dm idle · 2 actors StreamScan · accounts_dm StreamScan accounts_dm 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 25440 (Actor 125187,125186)
StreamMaterialize { columns: [target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, null:Int32(hidden), accounts_dm.account_id(hidden), null:Varchar(hidden), null:Date(hidden), $src(hidden)], stream_key: [accounts_dm.account_id, null:Varchar, null:Date, null:Int32, $src], pk_columns: [accounts_dm.account_id, null:Varchar, null:Date, null:Int32, $src], pk_conflict: NoCheck }
├── output: [ accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, accounts_dm.account_id, null:Varchar, null:Date, $src ]
├── stream key: [ accounts_dm.account_id, null:Varchar, null:Date, null:Int32, $src ]
└── StreamUnion { all: true } { output: [ accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, accounts_dm.account_id, null:Varchar, null:Date, $src ], stream key: [ accounts_dm.account_id, null:Varchar, null:Date, null:Int32, $src ] }
    ├── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, accounts_dm.account_id, null:Varchar, null:Date, 0:Int32 ], stream key: [ accounts_dm.account_id ] }
    ├── MergeExecutor { output: [ clients_dm.id, 'client':Varchar, clients_dm.id, 'client':Varchar, null:Int32, clients_dm.id, null:Varchar, null:Date, 1:Int32 ], stream key: [ clients_dm.id ] }
    ├── MergeExecutor { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, portfolios_dm.portfolio_id, null:Varchar, null:Date, 2:Int32 ], stream key: [ portfolios_dm.portfolio_id ] }
    ├── MergeExecutor
    │   ├── output: [ accounts_to_clients_dm.account_id, 'account':Varchar, accounts_to_clients_dm.client_id, 'client':Varchar, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, 3:Int32 ]
    │   └── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ account_to_portfolios_dm.account_id, 'account':Varchar, account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, 4:Int32 ]
    │   └── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ accounts_to_clients_dm.client_id, 'client':Varchar, accounts_to_clients_dm.account_id, 'account':Varchar, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, 5:Int32 ]
    │   └── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ clients_portfolios_dm.client_id, 'client':Varchar, clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, 6:Int32 ]
    │   └── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, clients_portfolios_dm.client_id, 'client':Varchar, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, 7:Int32 ]
    │   └── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, account_to_portfolios_dm.account_id, 'account':Varchar, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, 8:Int32 ]
    │   └── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    └── MergeExecutor
        ├── output: [ party_holder_edges_mv_next.party_id, 'party':Varchar, party_holder_edges_mv_next.entity_id, party_holder_edges_mv_next.entity_type, party_holder_edges_mv_next.$src, party_holder_edges_mv_next.party_id, party_holder_edges_mv_next.entity_id, null:Date, 9:Int32 ]
        └── stream key: [ party_holder_edges_mv_next.party_id, party_holder_edges_mv_next.entity_id, party_holder_edges_mv_next.$src ]

Fragment 25441 (Actor 125182,125183)
StreamProject { exprs: [accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, accounts_dm.account_id, null:Varchar, null:Date, 0:Int32] } { output: [ accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, accounts_dm.account_id, null:Varchar, null:Date, 0:Int32 ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 25442 (Actor 125185,125184)
StreamProject { exprs: [clients_dm.id, 'client':Varchar, clients_dm.id, 'client':Varchar, null:Int32, clients_dm.id, null:Varchar, null:Date, 1:Int32] } { output: [ clients_dm.id, 'client':Varchar, clients_dm.id, 'client':Varchar, null:Int32, clients_dm.id, null:Varchar, null:Date, 1:Int32 ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, closing_date] } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, closing_date ], stream key: [] }

Fragment 25443 (Actor 125227,125226)
StreamProject { exprs: [portfolios_dm.portfolio_id, 'portfolio':Varchar, portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, portfolios_dm.portfolio_id, null:Varchar, null:Date, 2:Int32] }
├── output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, portfolios_dm.portfolio_id, null:Varchar, null:Date, 2:Int32 ]
├── stream key: [ portfolios_dm.portfolio_id ]
└── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, disabled_at] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
        ├── Upstream { output: [ portfolio_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, disabled_at ], stream key: [] }

Fragment 25444 (Actor 125190,125191)
StreamProject { exprs: [accounts_to_clients_dm.account_id, 'account':Varchar, accounts_to_clients_dm.client_id, 'client':Varchar, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, 3:Int32] }
├── output: [ accounts_to_clients_dm.account_id, 'account':Varchar, accounts_to_clients_dm.client_id, 'client':Varchar, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, 3:Int32 ]
├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 25445 (Actor 125189,125188)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.client_id = clients_dm.id } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 25446 (Actor 125192,125193)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.account_id = accounts_dm.account_id } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    ├── 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 ] }
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 25447 (Actor 125229,125228)
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) AND IsNull(accounts_to_clients_dm.effective_end_date) }
    ├── 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, accounts_to_clients_dm.effective_end_date ]
    ├── 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, effective_end_date] }
        ├── 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, accounts_to_clients_dm.effective_end_date ]
        ├── 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, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, client_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

Fragment 25448 (Actor 125231,125230)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 25449 (Actor 125233,125232)
StreamProject { exprs: [clients_dm.id] } { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, closing_date] } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, closing_date ], stream key: [] }

Fragment 25450 (Actor 125194,125195)
StreamProject { exprs: [account_to_portfolios_dm.account_id, 'account':Varchar, account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, 4:Int32] }
├── output: [ account_to_portfolios_dm.account_id, 'account':Varchar, account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, 4:Int32 ]
├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── MergeExecutor { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 25451 (Actor 125197,125196)
StreamSyncLogStore { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id }
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ]
    ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 25452 (Actor 125199,125198)
StreamSyncLogStore { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.account_id = accounts_dm.account_id }
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ]
    ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── 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 ] }
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 25453 (Actor 125235,125234)
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) AND IsNull(account_to_portfolios_dm.effective_end_date) }
    ├── 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, account_to_portfolios_dm.effective_end_date ]
    ├── 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, effective_end_date] }
        ├── 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, account_to_portfolios_dm.effective_end_date ]
        ├── 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, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, portfolio_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

Fragment 25454 (Actor 125236,125237)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 25455 (Actor 125239,125238)
StreamProject { exprs: [portfolios_dm.portfolio_id] } { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, disabled_at] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
        ├── Upstream { output: [ portfolio_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, disabled_at ], stream key: [] }

Fragment 25456 (Actor 125203,125202)
StreamProject { exprs: [accounts_to_clients_dm.client_id, 'client':Varchar, accounts_to_clients_dm.account_id, 'account':Varchar, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, 5:Int32] }
├── output: [ accounts_to_clients_dm.client_id, 'client':Varchar, accounts_to_clients_dm.account_id, 'account':Varchar, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, 5:Int32 ]
├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 25457 (Actor 125201,125200)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.account_id = accounts_dm.account_id } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 25458 (Actor 125205,125204)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.client_id = clients_dm.id } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    ├── 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 ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 25459 (Actor 125240,125241)
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) AND IsNull(accounts_to_clients_dm.effective_end_date) }
    ├── 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, accounts_to_clients_dm.effective_end_date ]
    ├── 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, effective_end_date] }
        ├── 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, accounts_to_clients_dm.effective_end_date ]
        ├── 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, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, client_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

Fragment 25460 (Actor 125243,125242)
StreamProject { exprs: [clients_dm.id] } { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, closing_date] } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, closing_date ], stream key: [] }

Fragment 25461 (Actor 125245,125244)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 25462 (Actor 125209,125208)
StreamProject { exprs: [clients_portfolios_dm.client_id, 'client':Varchar, clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, 6:Int32] }
├── output: [ clients_portfolios_dm.client_id, 'client':Varchar, clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, 6:Int32 ]
├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 25463 (Actor 125207,125206)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 25464 (Actor 125211,125210)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.client_id = clients_dm.id } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── 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 ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 25465 (Actor 125247,125246)
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) AND IsNull(clients_portfolios_dm.effective_end_date) }
    ├── output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at, clients_portfolios_dm.effective_end_date ]
    ├── 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, effective_end_date] }
        ├── output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at, clients_portfolios_dm.effective_end_date ]
        ├── 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, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ client_id, portfolio_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

Fragment 25466 (Actor 125249,125248)
StreamProject { exprs: [clients_dm.id] } { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, closing_date] } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, closing_date ], stream key: [] }

Fragment 25467 (Actor 125251,125250)
StreamProject { exprs: [portfolios_dm.portfolio_id] } { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, disabled_at] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
        ├── Upstream { output: [ portfolio_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, disabled_at ], stream key: [] }

Fragment 25468 (Actor 125213,125212)
StreamProject { exprs: [clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, clients_portfolios_dm.client_id, 'client':Varchar, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, 7:Int32] }
├── output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, clients_portfolios_dm.client_id, 'client':Varchar, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, 7:Int32 ]
├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 25469 (Actor 125215,125214)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.client_id = clients_dm.id } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 25470 (Actor 125217,125216)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── 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 ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 25471 (Actor 125252,125253)
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) AND IsNull(clients_portfolios_dm.effective_end_date) }
    ├── output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at, clients_portfolios_dm.effective_end_date ]
    ├── 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, effective_end_date] }
        ├── output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at, clients_portfolios_dm.effective_end_date ]
        ├── 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, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ client_id, portfolio_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

Fragment 25472 (Actor 125255,125254)
StreamProject { exprs: [portfolios_dm.portfolio_id] } { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, disabled_at] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
        ├── Upstream { output: [ portfolio_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, disabled_at ], stream key: [] }

Fragment 25473 (Actor 125257,125256)
StreamProject { exprs: [clients_dm.id] } { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, closing_date] } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, closing_date ], stream key: [] }

Fragment 25474 (Actor 125221,125220)
StreamProject { exprs: [account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, account_to_portfolios_dm.account_id, 'account':Varchar, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, 8:Int32] }
├── output: [ account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, account_to_portfolios_dm.account_id, 'account':Varchar, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, 8:Int32 ]
├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── MergeExecutor { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 25475 (Actor 125219,125218)
StreamSyncLogStore { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.account_id = accounts_dm.account_id }
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ]
    ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 25476 (Actor 125222,125223)
StreamSyncLogStore { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id }
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ]
    ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── 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 ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 25477 (Actor 125259,125258)
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) AND IsNull(account_to_portfolios_dm.effective_end_date) }
    ├── 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, account_to_portfolios_dm.effective_end_date ]
    ├── 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, effective_end_date] }
        ├── 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, account_to_portfolios_dm.effective_end_date ]
        ├── 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, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, portfolio_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

Fragment 25478 (Actor 125261,125260)
StreamProject { exprs: [portfolios_dm.portfolio_id] } { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, disabled_at] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
        ├── Upstream { output: [ portfolio_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, disabled_at ], stream key: [] }

Fragment 25479 (Actor 125262,125263)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 25480 (Actor 125224,125225)
StreamProject { exprs: [party_holder_edges_mv_next.party_id, 'party':Varchar, party_holder_edges_mv_next.entity_id, party_holder_edges_mv_next.entity_type, party_holder_edges_mv_next.$src, party_holder_edges_mv_next.party_id, party_holder_edges_mv_next.entity_id, null:Date, 9:Int32] }
├── output: [ party_holder_edges_mv_next.party_id, 'party':Varchar, party_holder_edges_mv_next.entity_id, party_holder_edges_mv_next.entity_type, party_holder_edges_mv_next.$src, party_holder_edges_mv_next.party_id, party_holder_edges_mv_next.entity_id, null:Date, 9:Int32 ]
├── stream key: [ party_holder_edges_mv_next.party_id, party_holder_edges_mv_next.entity_id, party_holder_edges_mv_next.$src ]
└── StreamTableScan { table: party_holder_edges_mv_next, columns: [party_id, entity_id, entity_type, $src] } { output: [ party_holder_edges_mv_next.party_id, party_holder_edges_mv_next.entity_id, party_holder_edges_mv_next.entity_type, party_holder_edges_mv_next.$src ], stream key: [ party_holder_edges_mv_next.party_id, party_holder_edges_mv_next.entity_id, party_holder_edges_mv_next.$src ] }
    ├── Upstream { output: [ party_id, entity_id, entity_type, $src ], stream key: [] }
    └── BatchPlanNode { output: [ party_id, entity_id, entity_type, $src ], stream key: [] }