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

← cluster alinma_bff objects unlinked_account_pairs_mv explain
Overview Objects Graph History
materialized view · alinma_bff.unlinked_account_pairs_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
113 operators
Materialize · alinma_bff.unlinked_account_pairs_mv
0% idle 2 actors
Project
— — 2 actors
MaterializedExprs
0% idle 2 actors
Project
— — 2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · IsNull(account_to_portfolios_dm.account_id)
— — 2 actors
Filter · IsNull(account_to_portfolios_dm.account_id)
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · LeftOuter · accounts_dm.account_id = account_to_portfolios_dm.account_id
— — 2 actors
HashJoin · LeftOuter · accounts_dm.account_id = account_to_portfolios_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
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · account_to_portfolios_dm
— — 2 actors
DynamicFilter · account_to_portfolios_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 · account_to_portfolios_dm
— — 2 actors
DynamicFilter · account_to_portfolios_dm Dynamic filter — verify it pairs with a temporal condition to clean state
1% idle 2 actors
Merge
— — 2 actors
Exchange
0% 2/s 0 actors
Now
0% 2/s 1 actor
Project · account_to_portfolios_dm
— — 2 actors
Filter · account_to_portfolios_dm
1% idle 2 actors
StreamScan · account_to_portfolios_dm
1% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · reference_identifiers.entity_id = accounts_dm.account_id AN…
— — 2 actors
HashJoin · Inner · reference_identifiers.entity_id = accounts_dm.account_id AN… 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
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · account_primary_client_mv_next.account_id = accounts_dm.acc…
— — 2 actors
HashJoin · Inner · account_primary_client_mv_next.account_id = accounts_dm.acc… 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
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · product_types_dm.product_type_id = accounts_dm.product_type…
— — 2 actors
HashJoin · Inner · product_types_dm.product_type_id = accounts_dm.product_type… 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
LocalityProvider
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 · product_types_dm
— — 2 actors
Filter · product_types_dm
0% idle 2 actors
StreamScan · product_types_dm
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
StreamScan · account_primary_client_mv_next
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · accounts_dm.number = reference_identifiers.value
— — 2 actors
HashJoin · Inner · accounts_dm.number = reference_identifiers.value 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
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · reference_identifiers
— — 2 actors
Filter · reference_identifiers
0% idle 2 actors
StreamScan · reference_identifiers
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · account_primary_client_mv_next.account_id = accounts_dm.acc…
— — 2 actors
HashJoin · Inner · account_primary_client_mv_next.account_id = accounts_dm.acc… 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
StreamScan · account_primary_client_mv_next
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 · alinma_bff.unlinked_account_pairs_mv Materialize alinma_bff.unlinked_acc… idle · 2 actors Project Project — · 2 actors MaterializedExprs MaterializedExprs idle · 2 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · IsNull(account_to_portfolios_dm.account_id) Project IsNull(account_to_portf… — · 2 actors Filter · IsNull(account_to_portfolios_dm.account_id) Filter IsNull(account_to_portf… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · LeftOuter · accounts_dm.account_id = account_to_portfolios_dm.account_id SyncLogStore LeftOuter · accounts_dm… — · 2 actors HashJoin · LeftOuter · accounts_dm.account_id = account_to_portfolios_dm.account_id HashJoin LeftOuter · accounts_dm… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · account_to_portfolios_dm Project account_to_portfolios_dm — · 2 actors DynamicFilter · account_to_portfolios_dm DynamicFilter account_to_portfolios_dm idle · 2 actors Merge Merge — · 2 actors Exchange Exchange 2/s · 0 actors Now Now 2/s · 1 actor Project · account_to_portfolios_dm Project account_to_portfolios_dm — · 2 actors DynamicFilter · account_to_portfolios_dm DynamicFilter account_to_portfolios_dm idle · 2 actors Merge Merge — · 2 actors Exchange Exchange 2/s · 0 actors Now Now 2/s · 1 actor 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 LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · reference_identifiers.entity_id = accounts_dm.account_id AN… SyncLogStore Inner · reference_ident… — · 2 actors HashJoin · Inner · reference_identifiers.entity_id = accounts_dm.account_id AN… HashJoin Inner · reference_ident… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · account_primary_client_mv_next.account_id = accounts_dm.acc… SyncLogStore Inner · account_primary… — · 2 actors HashJoin · Inner · account_primary_client_mv_next.account_id = accounts_dm.acc… HashJoin Inner · account_primary… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · product_types_dm.product_type_id = accounts_dm.product_type… SyncLogStore Inner · product_types_d… — · 2 actors HashJoin · Inner · product_types_dm.product_type_id = accounts_dm.product_type… HashJoin Inner · product_types_d… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors 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 · product_types_dm Project product_types_dm — · 2 actors Filter · product_types_dm Filter product_types_dm idle · 2 actors StreamScan · product_types_dm StreamScan product_types_dm idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · account_primary_client_mv_next StreamScan account_primary_client_… idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · accounts_dm.number = reference_identifiers.value SyncLogStore Inner · accounts_dm.num… — · 2 actors HashJoin · Inner · accounts_dm.number = reference_identifiers.value HashJoin Inner · accounts_dm.num… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · reference_identifiers Project reference_identifiers — · 2 actors Filter · reference_identifiers Filter reference_identifiers idle · 2 actors StreamScan · reference_identifiers StreamScan reference_identifiers idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · account_primary_client_mv_next.account_id = accounts_dm.acc… SyncLogStore Inner · account_primary… — · 2 actors HashJoin · Inner · account_primary_client_mv_next.account_id = accounts_dm.acc… HashJoin Inner · account_primary… 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 StreamScan · account_primary_client_mv_next StreamScan account_primary_client_… 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 63498 (Actor 746366,746365)
StreamMaterialize { columns: [id, security_account_id, cash_account_id, registered_by, security_account, cash_account], stream_key: [security_account_id, cash_account_id], pk_columns: [security_account_id, cash_account_id], pk_conflict: NoCheck }
├── output: [ $expr5, accounts_dm.account_id, accounts_dm.account_id, 'risingwave-pipeline':Varchar, $expr6, $expr7 ]
├── stream key: [ accounts_dm.account_id, accounts_dm.account_id ]
└── StreamProject { exprs: [$expr5, accounts_dm.account_id, accounts_dm.account_id, 'risingwave-pipeline':Varchar, JsonbAccess(jsonb_agg($expr3 order_by(reference_identifiers.id ASC)), 0:Int32) as $expr6, JsonbAccess(jsonb_agg($expr4 order_by(reference_identifiers.id ASC)), 0:Int32) as $expr7] }
    ├── output: [ $expr5, accounts_dm.account_id, accounts_dm.account_id, 'risingwave-pipeline':Varchar, $expr6, $expr7 ]
    ├── stream key: [ accounts_dm.account_id, accounts_dm.account_id ]
    └── StreamMaterializedExprs { exprs: [id_from_string_with_prefix('unlinked_account_pair':Varchar, ConcatOp(ConcatOp(accounts_dm.account_id, ':':Varchar), accounts_dm.account_id)) as $expr5] }
        ├── output: [ accounts_dm.account_id, accounts_dm.account_id, jsonb_agg($expr3 order_by(reference_identifiers.id ASC)), jsonb_agg($expr4 order_by(reference_identifiers.id ASC)), $expr5 ]
        ├── stream key: [ accounts_dm.account_id, accounts_dm.account_id ]
        └── StreamProject { exprs: [accounts_dm.account_id, accounts_dm.account_id, jsonb_agg($expr3 order_by(reference_identifiers.id ASC)), jsonb_agg($expr4 order_by(reference_identifiers.id ASC))] }
            ├── output: [ accounts_dm.account_id, accounts_dm.account_id, jsonb_agg($expr3 order_by(reference_identifiers.id ASC)), jsonb_agg($expr4 order_by(reference_identifiers.id ASC)) ]
            ├── stream key: [ accounts_dm.account_id, accounts_dm.account_id ]
            └── StreamHashAgg { group_key: [accounts_dm.account_id, accounts_dm.account_id], aggs: [jsonb_agg($expr3 order_by(reference_identifiers.id ASC)), jsonb_agg($expr4 order_by(reference_identifiers.id ASC)), count] }
                ├── output: [ accounts_dm.account_id, accounts_dm.account_id, jsonb_agg($expr3 order_by(reference_identifiers.id ASC)), jsonb_agg($expr4 order_by(reference_identifiers.id ASC)), count ]
                ├── stream key: [ accounts_dm.account_id, accounts_dm.account_id ]
                └── StreamLocalityProvider { locality_columns: [accounts_dm.account_id, accounts_dm.account_id] }
                    ├── output:
                    │   ┌── accounts_dm.account_id
                    │   ├── accounts_dm.account_id
                    │   ├── reference_identifiers.id
                    │   ├── $expr3
                    │   ├── $expr4
                    │   ├── reference_identifiers.entity_id
                    │   ├── accounts_dm.number
                    │   ├── account_primary_client_mv_next.account_id
                    │   ├── product_types_dm.product_type_id
                    │   ├── account_to_portfolios_dm.portfolio_id
                    │   └── account_to_portfolios_dm.effective_start_date
                    ├── stream key:
                    │   ┌── accounts_dm.account_id
                    │   ├── accounts_dm.account_id
                    │   ├── reference_identifiers.entity_id
                    │   ├── accounts_dm.number
                    │   ├── account_primary_client_mv_next.account_id
                    │   ├── reference_identifiers.id
                    │   ├── product_types_dm.product_type_id
                    │   ├── account_to_portfolios_dm.portfolio_id
                    │   └── account_to_portfolios_dm.effective_start_date
                    └── MergeExecutor
                        ├── output:
                        │   ┌── accounts_dm.account_id
                        │   ├── accounts_dm.account_id
                        │   ├── reference_identifiers.id
                        │   ├── $expr3
                        │   ├── $expr4
                        │   ├── reference_identifiers.entity_id
                        │   ├── accounts_dm.number
                        │   ├── account_primary_client_mv_next.account_id
                        │   ├── product_types_dm.product_type_id
                        │   ├── account_to_portfolios_dm.portfolio_id
                        │   └── account_to_portfolios_dm.effective_start_date
                        └── stream key:
                            ┌── accounts_dm.account_id
                            ├── reference_identifiers.entity_id
                            ├── accounts_dm.number
                            ├── account_primary_client_mv_next.account_id
                            ├── reference_identifiers.id
                            ├── product_types_dm.product_type_id
                            ├── account_to_portfolios_dm.portfolio_id
                            └── account_to_portfolios_dm.effective_start_date

Fragment 63499 (Actor 746369,746370)
StreamProject { exprs: [accounts_dm.account_id, accounts_dm.account_id, reference_identifiers.id, JsonbBuildObject('id':Varchar, accounts_dm.account_id, 'external_id':Varchar, '':Varchar, 'account_number':Varchar, accounts_dm.number, 'currency_code':Varchar, Coalesce(accounts_dm.base_currency_code, '':Varchar), 'opened_at':Varchar, Case(IsNull(accounts_dm.opening_date), null:Jsonb, JsonbBuildObject('seconds':Varchar, Extract('EPOCH':Varchar, accounts_dm.opening_date::Timestamp)::Int64)), 'client':Varchar, account_primary_client_mv_next.client_json) as $expr3, JsonbBuildObject('id':Varchar, accounts_dm.account_id, 'external_id':Varchar, '':Varchar, 'account_number':Varchar, accounts_dm.number, 'currency_code':Varchar, Coalesce(accounts_dm.base_currency_code, '':Varchar), 'opened_at':Varchar, Case(IsNull(accounts_dm.opening_date), null:Jsonb, JsonbBuildObject('seconds':Varchar, Extract('EPOCH':Varchar, accounts_dm.opening_date::Timestamp)::Int64)), 'client':Varchar, account_primary_client_mv_next.client_json) as $expr4, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date] }
├── output: [ accounts_dm.account_id, accounts_dm.account_id, reference_identifiers.id, $expr3, $expr4, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ accounts_dm.account_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── StreamFilter { predicate: IsNull(account_to_portfolios_dm.account_id) } { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, account_to_portfolios_dm.account_id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ], stream key: [ accounts_dm.account_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, account_to_portfolios_dm.account_id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ], stream key: [ accounts_dm.account_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 63500 (Actor 746368,746367)
StreamSyncLogStore { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, account_to_portfolios_dm.account_id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ], stream key: [ accounts_dm.account_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: LeftOuter, predicate: accounts_dm.account_id = account_to_portfolios_dm.account_id } { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, account_to_portfolios_dm.account_id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ], stream key: [ accounts_dm.account_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_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 63501 (Actor 746372,746371)
StreamLocalityProvider { locality_columns: [accounts_dm.account_id] } { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id ] }
└── MergeExecutor { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id ], stream key: [ reference_identifiers.entity_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id ] }

Fragment 63502 (Actor 746378,746377)
StreamSyncLogStore { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id ], stream key: [ reference_identifiers.entity_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id ] }
└── StreamHashJoin { type: Inner, predicate: reference_identifiers.entity_id = accounts_dm.account_id AND reference_identifiers.entity_id = account_primary_client_mv_next.account_id } { output: [ reference_identifiers.id, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.client_json, account_primary_client_mv_next.client_json, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id ], stream key: [ reference_identifiers.entity_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id, product_types_dm.product_type_id ] }
    ├── MergeExecutor { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, reference_identifiers.id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, reference_identifiers.value ], stream key: [ reference_identifiers.entity_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id ] }
    └── MergeExecutor { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id ] }

Fragment 63503 (Actor 746380,746379)
StreamLocalityProvider { locality_columns: [reference_identifiers.entity_id, reference_identifiers.entity_id] } { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, reference_identifiers.id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, reference_identifiers.value ], stream key: [ reference_identifiers.entity_id, reference_identifiers.entity_id, accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id ] }
└── MergeExecutor { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, reference_identifiers.id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, reference_identifiers.value ], stream key: [ accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id ] }

Fragment 63504 (Actor 746381,746382)
StreamSyncLogStore { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, reference_identifiers.id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, reference_identifiers.value ], stream key: [ accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id ] }
└── StreamHashJoin { type: Inner, predicate: accounts_dm.number = reference_identifiers.value } { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, reference_identifiers.id, reference_identifiers.entity_id, account_primary_client_mv_next.account_id, reference_identifiers.value ], stream key: [ accounts_dm.number, account_primary_client_mv_next.account_id, reference_identifiers.id ] }
    ├── MergeExecutor { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.account_id ], stream key: [ accounts_dm.number, account_primary_client_mv_next.account_id ] }
    └── MergeExecutor { output: [ reference_identifiers.id, reference_identifiers.entity_id, reference_identifiers.value ], stream key: [ reference_identifiers.value, reference_identifiers.id ] }

Fragment 63505 (Actor 746383,746384)
StreamLocalityProvider { locality_columns: [accounts_dm.number] } { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.account_id ], stream key: [ accounts_dm.number, account_primary_client_mv_next.account_id ] }
└── MergeExecutor { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.account_id ], stream key: [ account_primary_client_mv_next.account_id ] }

Fragment 63506 (Actor 746386,746385)
StreamSyncLogStore { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.account_id ], stream key: [ account_primary_client_mv_next.account_id ] }
└── StreamHashJoin { type: Inner, predicate: account_primary_client_mv_next.account_id = accounts_dm.account_id } { output: [ account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, account_primary_client_mv_next.account_id ], stream key: [ account_primary_client_mv_next.account_id ] }
    ├── MergeExecutor { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json ], stream key: [ account_primary_client_mv_next.account_id ] }
    └── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date ], stream key: [ accounts_dm.account_id ] }

Fragment 63507 (Actor 746408,746407)
StreamTableScan { table: account_primary_client_mv_next, columns: [account_id, client_json] } { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json ], stream key: [ account_primary_client_mv_next.account_id ] }
├── Upstream { output: [ account_id, client_json ], stream key: [] }
└── BatchPlanNode { output: [ account_id, client_json ], stream key: [] }

Fragment 63508 (Actor 746409,746410)
StreamProject { exprs: [accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date] } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, number, base_currency_code, opening_date, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, number, base_currency_code, opening_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, number, base_currency_code, opening_date, disabled_at ], stream key: [] }

Fragment 63509 (Actor 746387,746388)
StreamLocalityProvider { locality_columns: [reference_identifiers.value] } { output: [ reference_identifiers.id, reference_identifiers.entity_id, reference_identifiers.value ], stream key: [ reference_identifiers.value, reference_identifiers.id ] }
└── MergeExecutor { output: [ reference_identifiers.id, reference_identifiers.entity_id, reference_identifiers.value ], stream key: [ reference_identifiers.id ] }

Fragment 63510 (Actor 746411,746412)
StreamProject { exprs: [reference_identifiers.id, reference_identifiers.entity_id, reference_identifiers.value] } { output: [ reference_identifiers.id, reference_identifiers.entity_id, reference_identifiers.value ], stream key: [ reference_identifiers.id ] }
└── StreamFilter { predicate: IsNotNull(reference_identifiers.entity_id) AND (reference_identifiers.entity_type = 'account':Varchar) AND (reference_identifiers.key = 'cash_account_num':Varchar) } { output: [ reference_identifiers.id, reference_identifiers.entity_id, reference_identifiers.value, reference_identifiers.entity_type, reference_identifiers.key ], stream key: [ reference_identifiers.id ] }
    └── StreamTableScan { table: reference_identifiers, columns: [id, entity_id, value, entity_type, key] } { output: [ reference_identifiers.id, reference_identifiers.entity_id, reference_identifiers.value, reference_identifiers.entity_type, reference_identifiers.key ], stream key: [ reference_identifiers.id ] }
        ├── Upstream { output: [ id, entity_id, value, entity_type, key ], stream key: [] }
        └── BatchPlanNode { output: [ id, entity_id, value, entity_type, key ], stream key: [] }

Fragment 63511 (Actor 746389,746390)
StreamLocalityProvider { locality_columns: [accounts_dm.account_id, account_primary_client_mv_next.account_id] } { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, account_primary_client_mv_next.account_id, product_types_dm.product_type_id ] }
└── MergeExecutor { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id ], stream key: [ account_primary_client_mv_next.account_id, product_types_dm.product_type_id ] }

Fragment 63512 (Actor 746391,746392)
StreamSyncLogStore { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id ], stream key: [ account_primary_client_mv_next.account_id, product_types_dm.product_type_id ] }
└── StreamHashJoin { type: Inner, predicate: account_primary_client_mv_next.account_id = accounts_dm.account_id } { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json, accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id ], stream key: [ account_primary_client_mv_next.account_id, product_types_dm.product_type_id ] }
    ├── MergeExecutor { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json ], stream key: [ account_primary_client_mv_next.account_id ] }
    └── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id, accounts_dm.product_type_id ], stream key: [ accounts_dm.account_id, product_types_dm.product_type_id ] }

Fragment 63513 (Actor 746413,746414)
StreamTableScan { table: account_primary_client_mv_next, columns: [account_id, client_json] } { output: [ account_primary_client_mv_next.account_id, account_primary_client_mv_next.client_json ], stream key: [ account_primary_client_mv_next.account_id ] }
├── Upstream { output: [ account_id, client_json ], stream key: [] }
└── BatchPlanNode { output: [ account_id, client_json ], stream key: [] }

Fragment 63514 (Actor 746393,746394)
StreamLocalityProvider { locality_columns: [accounts_dm.account_id] } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id, accounts_dm.product_type_id ], stream key: [ accounts_dm.account_id, product_types_dm.product_type_id ] }
└── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id, accounts_dm.product_type_id ], stream key: [ product_types_dm.product_type_id, accounts_dm.account_id ] }

Fragment 63515 (Actor 746396,746395)
StreamSyncLogStore { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id, accounts_dm.product_type_id ], stream key: [ product_types_dm.product_type_id, accounts_dm.account_id ] }
└── StreamHashJoin { type: Inner, predicate: product_types_dm.product_type_id = accounts_dm.product_type_id } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.base_currency_code, accounts_dm.opening_date, product_types_dm.product_type_id, accounts_dm.product_type_id ], stream key: [ product_types_dm.product_type_id, accounts_dm.account_id ] }
    ├── MergeExecutor { output: [ product_types_dm.product_type_id ], stream key: [ product_types_dm.product_type_id ] }
    └── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.product_type_id, accounts_dm.base_currency_code, accounts_dm.opening_date ], stream key: [ accounts_dm.product_type_id, accounts_dm.account_id ] }

Fragment 63516 (Actor 746416,746415)
StreamProject { exprs: [product_types_dm.product_type_id] } { output: [ product_types_dm.product_type_id ], stream key: [ product_types_dm.product_type_id ] }
└── StreamFilter { predicate: (product_types_dm.type = 'INVESTMENT':Varchar) } { output: [ product_types_dm.product_type_id, product_types_dm.type ], stream key: [ product_types_dm.product_type_id ] }
    └── StreamTableScan { table: product_types_dm, columns: [product_type_id, type] } { output: [ product_types_dm.product_type_id, product_types_dm.type ], stream key: [ product_types_dm.product_type_id ] }
        ├── Upstream { output: [ product_type_id, type ], stream key: [] }
        └── BatchPlanNode { output: [ product_type_id, type ], stream key: [] }

Fragment 63517 (Actor 746398,746397)
StreamLocalityProvider { locality_columns: [accounts_dm.product_type_id] } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.product_type_id, accounts_dm.base_currency_code, accounts_dm.opening_date ], stream key: [ accounts_dm.product_type_id, accounts_dm.account_id ] }
└── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.product_type_id, accounts_dm.base_currency_code, accounts_dm.opening_date ], stream key: [ accounts_dm.account_id ] }

Fragment 63518 (Actor 746418,746417)
StreamProject { exprs: [accounts_dm.account_id, accounts_dm.number, accounts_dm.product_type_id, accounts_dm.base_currency_code, accounts_dm.opening_date] } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.product_type_id, accounts_dm.base_currency_code, accounts_dm.opening_date ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.product_type_id, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, number, product_type_id, base_currency_code, opening_date, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.number, accounts_dm.product_type_id, accounts_dm.base_currency_code, accounts_dm.opening_date, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, number, product_type_id, base_currency_code, opening_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, number, product_type_id, base_currency_code, opening_date, disabled_at ], stream key: [] }

Fragment 63519 (Actor 746400,746399)
StreamLocalityProvider { locality_columns: [account_to_portfolios_dm.account_id] } { 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: [ 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 63520 (Actor 746422,746421)
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 ] }
└── StreamDynamicFilter { predicate: ($expr2 > now), output_watermarks: [[$expr2]], output: [account_to_portfolios_dm.account_id, $expr2, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date], cleaned_by_watermark: true } { output: [ account_to_portfolios_dm.account_id, $expr2, 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 ] }
    ├── StreamProject { exprs: [account_to_portfolios_dm.account_id, AtTimeZone(Coalesce(account_to_portfolios_dm.effective_end_date, '9999-12-31':Date)::Timestamp, 'UTC':Varchar) as $expr2, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date] } { output: [ account_to_portfolios_dm.account_id, $expr2, 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 ] }
    │   └── StreamDynamicFilter { predicate: ($expr1 <= now), output: [account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_end_date, $expr1, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date], cleaned_by_watermark: true } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_end_date, $expr1, 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 ] }
    │       ├── StreamProject { exprs: [account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_end_date, AtTimeZone(account_to_portfolios_dm.effective_start_date::Timestamp, 'UTC':Varchar) as $expr1, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date] } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_end_date, $expr1, 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.effective_start_date, account_to_portfolios_dm.effective_end_date, account_to_portfolios_dm.portfolio_id, 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, effective_start_date, effective_end_date, portfolio_id, disabled_at] } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date, account_to_portfolios_dm.effective_end_date, account_to_portfolios_dm.portfolio_id, 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, effective_start_date, effective_end_date, portfolio_id, disabled_at ], stream key: [] }
    │       │           └── BatchPlanNode { output: [ account_id, effective_start_date, effective_end_date, portfolio_id, disabled_at ], stream key: [] }
    │       └── MergeExecutor { output: [ now ], stream key: [] }
    └── MergeExecutor { output: [ now ], stream key: [] }

Fragment 63521 (Actor 746401)
StreamNow { output: [ now ], stream key: [] }

Fragment 63522 (Actor 746402)
StreamNow { output: [ now ], stream key: [] }