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

← cluster adib_rm objects client_cash_tiles_mv explain
Overview Objects Graph History
materialized view · adib_rm.client_cash_tiles_mv profiled over 5s
seconds (1–30)
Stateful hash join (4 state tables) — consider a temporal join for dimension lookupsAggregation state — unbounded unless keyed or temporally filteredDynamic filter — verify it pairs with a temporal condition to clean state
65 operators
Materialize · adib_rm.client_cash_tiles_mv
0% idle 2 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
Project
— — 2 actors
Project
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · LeftSemi · accounts_plain_mv.product_type_id = cash_product_type_ids_m…
— — 2 actors
HashJoin · LeftSemi · accounts_plain_mv.product_type_id = cash_product_type_ids_m… 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 · cash_product_type_ids_mv
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · LeftOuter · accounts_plain_mv.account_id = investment_account_balance_s…
— — 2 actors
HashJoin · LeftOuter · accounts_plain_mv.account_id = investment_account_balance_s… 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 · investment_account_balance_snapshot_journal_mv_next
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
TemporalJoin · LeftOuter · accounts_plain_mv.base_currency_code = currencies_dm.code, …
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
StreamScan · currencies_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_plain_mv.accou…
— — 2 actors
HashJoin · Inner · accounts_to_clients_dm.account_id = accounts_plain_mv.accou… 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 · accounts_plain_mv
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · clients_dm.id = accounts_to_clients_dm.client_id
— — 2 actors
HashJoin · Inner · clients_dm.id = accounts_to_clients_dm.client_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_to_clients_dm
— — 2 actors
DynamicFilter · accounts_to_clients_dm Dynamic filter — verify it pairs with a temporal condition to clean state
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% 2/s 0 actors
Now
0% 2/s 1 actor
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 · clients_dm
— — 2 actors
Filter · clients_dm
0% idle 2 actors
StreamScan · clients_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 · adib_rm.client_cash_tiles_mv Materialize adib_rm.client_cash_til… idle · 2 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · LeftSemi · accounts_plain_mv.product_type_id = cash_product_type_ids_m… SyncLogStore LeftSemi · accounts_pla… — · 2 actors HashJoin · LeftSemi · accounts_plain_mv.product_type_id = cash_product_type_ids_m… HashJoin LeftSemi · accounts_pla… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · cash_product_type_ids_mv StreamScan cash_product_type_ids_mv idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · LeftOuter · accounts_plain_mv.account_id = investment_account_balance_s… SyncLogStore LeftOuter · accounts_pl… — · 2 actors HashJoin · LeftOuter · accounts_plain_mv.account_id = investment_account_balance_s… HashJoin LeftOuter · accounts_pl… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · investment_account_balance_snapshot_journal_mv_next StreamScan investment_account_bala… idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors TemporalJoin · LeftOuter · accounts_plain_mv.base_currency_code = currencies_dm.code, … TemporalJoin LeftOuter · accounts_pl… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · currencies_dm StreamScan currencies_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_plain_mv.accou… SyncLogStore Inner · accounts_to_cli… — · 2 actors HashJoin · Inner · accounts_to_clients_dm.account_id = accounts_plain_mv.accou… HashJoin Inner · accounts_to_cli… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · accounts_plain_mv StreamScan accounts_plain_mv idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · clients_dm.id = accounts_to_clients_dm.client_id SyncLogStore Inner · clients_dm.id =… — · 2 actors HashJoin · Inner · clients_dm.id = accounts_to_clients_dm.client_id HashJoin Inner · clients_dm.id =… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · accounts_to_clients_dm Project accounts_to_clients_dm — · 2 actors DynamicFilter · accounts_to_clients_dm DynamicFilter accounts_to_clients_dm idle · 2 actors Merge Merge — · 2 actors Exchange Exchange 2/s · 0 actors Now Now 2/s · 1 actor 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 · 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
Streaming operator plan from EXPLAIN ANALYZE. Node heat = backpressure. Drag to pan, scroll to zoom.
Fragments (DESCRIBE FRAGMENTS) — click to expand 
Fragment 23653 (Actor 104896,104895)
StreamMaterialize { columns: [client_id, cash_tiles], stream_key: [client_id], pk_columns: [client_id], pk_conflict: NoCheck }
├── output: [ clients_dm.id, jsonb_agg($expr3) ]
├── stream key: [ clients_dm.id ]
└── StreamProject { exprs: [clients_dm.id, jsonb_agg($expr3)] }
    ├── output: [ clients_dm.id, jsonb_agg($expr3) ]
    ├── stream key: [ clients_dm.id ]
    └── StreamHashAgg { group_key: [clients_dm.id], aggs: [jsonb_agg($expr3), count] }
        ├── output: [ clients_dm.id, jsonb_agg($expr3), count ]
        ├── stream key: [ clients_dm.id ]
        └── MergeExecutor
            ├── output:
            │   ┌── clients_dm.id
            │   ├── $expr3
            │   ├── accounts_to_clients_dm.account_id
            │   ├── accounts_to_clients_dm.effective_start_date
            │   ├── accounts_plain_mv.base_currency_code
            │   ├── accounts_plain_mv.account_id
            │   └── accounts_plain_mv.product_type_id
            └── stream key:
                ┌── clients_dm.id
                ├── accounts_to_clients_dm.account_id
                ├── accounts_to_clients_dm.effective_start_date
                ├── accounts_plain_mv.base_currency_code
                ├── accounts_plain_mv.account_id
                └── accounts_plain_mv.product_type_id

Fragment 23654 (Actor 104897,104898)
StreamProject { exprs: [clients_dm.id, JsonbBuildObject('accountId':Varchar, accounts_plain_mv.account_id, 'name':Varchar, accounts_plain_mv.name, 'number':Varchar, accounts_plain_mv.number, 'value':Varchar, JsonbBuildObject('amount':Varchar, investment_account_balance_snapshot_journal_mv_next.available_balance::Varchar, 'currency':Varchar, $expr2), 'baseCurrency':Varchar, $expr2) as $expr3, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id, accounts_plain_mv.product_type_id] }
├── output: [ clients_dm.id, $expr3, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id, accounts_plain_mv.product_type_id ]
├── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id, accounts_plain_mv.product_type_id ]
└── StreamProject { exprs: [clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.base_currency_code, currencies_dm.symbol, investment_account_balance_snapshot_journal_mv_next.available_balance, JsonbBuildObject('code':Varchar, accounts_plain_mv.base_currency_code, 'symbol':Varchar, currencies_dm.symbol) as $expr2, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.product_type_id] }
    ├── output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.base_currency_code, currencies_dm.symbol, investment_account_balance_snapshot_journal_mv_next.available_balance, $expr2, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.product_type_id ]
    ├── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id, accounts_plain_mv.product_type_id ]
    └── MergeExecutor
        ├── output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.base_currency_code, currencies_dm.symbol, investment_account_balance_snapshot_journal_mv_next.available_balance, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.product_type_id ]
        └── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id, accounts_plain_mv.product_type_id ]

Fragment 23655 (Actor 104899,104900)
StreamSyncLogStore
├── output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.base_currency_code, currencies_dm.symbol, investment_account_balance_snapshot_journal_mv_next.available_balance, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.product_type_id ]
├── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id, accounts_plain_mv.product_type_id ]
└── StreamHashJoin { type: LeftSemi, predicate: accounts_plain_mv.product_type_id = cash_product_type_ids_mv.product_type_id }
    ├── output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.base_currency_code, currencies_dm.symbol, investment_account_balance_snapshot_journal_mv_next.available_balance, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.product_type_id ]
    ├── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id, accounts_plain_mv.product_type_id ]
    ├── MergeExecutor
    │   ├── output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code, currencies_dm.symbol, investment_account_balance_snapshot_journal_mv_next.available_balance, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, investment_account_balance_snapshot_journal_mv_next.account_id ]
    │   └── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id ]
    └── MergeExecutor { output: [ cash_product_type_ids_mv.product_type_id, cash_product_type_ids_mv._row_id, cash_product_type_ids_mv.$src ], stream key: [ cash_product_type_ids_mv._row_id, cash_product_type_ids_mv.$src ] }

Fragment 23656 (Actor 104902,104901)
StreamSyncLogStore
├── output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code, currencies_dm.symbol, investment_account_balance_snapshot_journal_mv_next.available_balance, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, investment_account_balance_snapshot_journal_mv_next.account_id ]
├── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id ]
└── StreamHashJoin { type: LeftOuter, predicate: accounts_plain_mv.account_id = investment_account_balance_snapshot_journal_mv_next.account_id }
    ├── output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code, currencies_dm.symbol, investment_account_balance_snapshot_journal_mv_next.available_balance, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, investment_account_balance_snapshot_journal_mv_next.account_id ]
    ├── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code, accounts_plain_mv.account_id ]
    ├── MergeExecutor { output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code, currencies_dm.symbol, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, currencies_dm.code ], stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code ] }
    └── MergeExecutor { output: [ investment_account_balance_snapshot_journal_mv_next.account_id, investment_account_balance_snapshot_journal_mv_next.available_balance ], stream key: [ investment_account_balance_snapshot_journal_mv_next.account_id ] }

Fragment 23657 (Actor 104827,104826)
StreamTemporalJoin { type: LeftOuter, append_only: false, predicate: accounts_plain_mv.base_currency_code = currencies_dm.code, nested_loop: false }
├── output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code, currencies_dm.symbol, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, currencies_dm.code ]
├── stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, accounts_plain_mv.base_currency_code ]
├── MergeExecutor { output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ], stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── MergeExecutor { output: [ currencies_dm.code, currencies_dm.symbol ], stream key: [ currencies_dm.code ] }

Fragment 23658 (Actor 104904,104903)
StreamSyncLogStore { output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ], stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.account_id = accounts_plain_mv.account_id } { output: [ clients_dm.id, accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ], stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
    └── MergeExecutor { output: [ accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code ], stream key: [ accounts_plain_mv.account_id ] }

Fragment 23659 (Actor 104905,104906)
StreamSyncLogStore { output: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_dm.id = accounts_to_clients_dm.client_id } { output: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ clients_dm.id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.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 23660 (Actor 104907,104908)
StreamProject { exprs: [clients_dm.id] } { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.disabled_at) } { output: [ clients_dm.id, clients_dm.disabled_at ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, disabled_at] } { output: [ clients_dm.id, clients_dm.disabled_at ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ id, disabled_at ], stream key: [] }

Fragment 23661 (Actor 104910,104909)
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 ] }
└── StreamDynamicFilter { predicate: ($expr1 <= now), output: [accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, $expr1, accounts_to_clients_dm.effective_start_date], cleaned_by_watermark: true } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, $expr1, 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 ] }
    ├── StreamProject { exprs: [accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, AtTimeZone(accounts_to_clients_dm.effective_start_date::Timestamp, 'UTC':Varchar) as $expr1, accounts_to_clients_dm.effective_start_date] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, $expr1, 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: [] }
    └── MergeExecutor { output: [ now ], stream key: [] }

Fragment 23662 (Actor 104911)
StreamNow { output: [ now ], stream key: [] }

Fragment 23663 (Actor 104913,104912)
StreamTableScan { table: accounts_plain_mv, columns: [account_id, name, number, product_type_id, base_currency_code] } { output: [ accounts_plain_mv.account_id, accounts_plain_mv.name, accounts_plain_mv.number, accounts_plain_mv.product_type_id, accounts_plain_mv.base_currency_code ], stream key: [ accounts_plain_mv.account_id ] }
├── Upstream { output: [ account_id, name, number, product_type_id, base_currency_code ], stream key: [] }
└── BatchPlanNode { output: [ account_id, name, number, product_type_id, base_currency_code ], stream key: [] }

Fragment 23664 (Actor 104825,104824)
StreamTableScan { table: currencies_dm, columns: [code, symbol] } { output: [ currencies_dm.code, currencies_dm.symbol ], stream key: [ currencies_dm.code ] }
├── Upstream { output: [ code, symbol ], stream key: [] }
└── BatchPlanNode { output: [ code, symbol ], stream key: [] }

Fragment 23665 (Actor 104914,104915)
StreamTableScan { table: investment_account_balance_snapshot_journal_mv_next, columns: [account_id, available_balance] } { output: [ investment_account_balance_snapshot_journal_mv_next.account_id, investment_account_balance_snapshot_journal_mv_next.available_balance ], stream key: [ investment_account_balance_snapshot_journal_mv_next.account_id ] }
├── Upstream { output: [ account_id, available_balance ], stream key: [] }
└── BatchPlanNode { output: [ account_id, available_balance ], stream key: [] }

Fragment 23666 (Actor 104916,104917)
StreamTableScan { table: cash_product_type_ids_mv, columns: [product_type_id, _row_id, $src] } { output: [ cash_product_type_ids_mv.product_type_id, cash_product_type_ids_mv._row_id, cash_product_type_ids_mv.$src ], stream key: [ cash_product_type_ids_mv._row_id, cash_product_type_ids_mv.$src ] }
├── Upstream { output: [ product_type_id, _row_id, $src ], stream key: [] }
└── BatchPlanNode { output: [ product_type_id, _row_id, $src ], stream key: [] }