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

← cluster adib_rm objects sdk_portfolio_chart_mv explain
Overview Objects Graph History
materialized view · adib_rm.sdk_portfolio_chart_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
166 operators
Materialize · adib_rm.sdk_portfolio_chart_mv
0% idle 2 actors
Union
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project
— — 2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · (portfolio_to_account_groups_mv.type = 'all':Varchar)
— — 2 actors
Filter · (portfolio_to_account_groups_mv.type = 'all':Varchar)
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · LeftOuter · position_summary_mv_next.account_group_id = pnl_summary_mv_…
— — 2 actors
HashJoin · LeftOuter · position_summary_mv_next.account_group_id = pnl_summary_mv_… 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 · pnl_summary_mv_next
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · portfolio_to_account_groups_mv.account_group_id = position_…
— — 2 actors
HashJoin · Inner · portfolio_to_account_groups_mv.account_group_id = position_… 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
Filter · position_summary_mv_next
0% idle 2 actors
StreamScan · position_summary_mv_next
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · portfolio_to_account_groups_mv.portfolio_id = portfolios_dm…
— — 2 actors
HashJoin · Inner · portfolio_to_account_groups_mv.portfolio_id = portfolios_dm… 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
Filter · portfolio_to_account_groups_mv
0% idle 2 actors
StreamScan · portfolio_to_account_groups_mv
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project
— — 2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · ($expr13 >= $expr14), output_watermarks: [[$expr13]], outpu…
— — 2 actors
DynamicFilter · ($expr13 >= $expr14), output_watermarks: [[$expr13]], outpu… 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
Project
— — 1 actor
Now
0% 2/s 1 actor
Project · ($expr13 >= $expr14), output_watermarks: [[$expr13]], outpu…
— — 2 actors
Filter · ($expr13 >= $expr14), output_watermarks: [[$expr13]], outpu…
1% idle 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project
— — 2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · ($expr10 >= $expr11), output_watermarks: [[$expr10]], outpu…
— — 2 actors
DynamicFilter · ($expr10 >= $expr11), output_watermarks: [[$expr10]], outpu… Dynamic filter — verify it pairs with a temporal condition to clean state
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project
— — 1 actor
Now
0% 2/s 1 actor
Project · ($expr10 >= $expr11), output_watermarks: [[$expr10]], outpu…
— — 2 actors
Filter · ($expr10 >= $expr11), output_watermarks: [[$expr10]], outpu…
0% idle 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project
— — 2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: …
— — 2 actors
DynamicFilter · ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: … 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
Project
— — 1 actor
Now
0% 2/s 1 actor
Project · ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: …
— — 2 actors
Filter · ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: …
1% idle 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project
— — 2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: …
— — 2 actors
DynamicFilter · ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: … 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
Project
— — 1 actor
Now
0% 2/s 1 actor
Project · ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: …
— — 2 actors
Filter · ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: …
1% idle 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project
— — 2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: …
— — 2 actors
DynamicFilter · ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: … 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
Project
— — 1 actor
Now
0% 2/s 1 actor
Project · ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: …
— — 2 actors
Filter · ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: …
1% idle 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.sdk_portfolio_chart_mv Materialize adib_rm.sdk_portfolio_c… idle · 2 actors Union Union idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · (portfolio_to_account_groups_mv.type = 'all':Varchar) Project (portfolio_to_account_g… — · 2 actors Filter · (portfolio_to_account_groups_mv.type = 'all':Varchar) Filter (portfolio_to_account_g… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · LeftOuter · position_summary_mv_next.account_group_id = pnl_summary_mv_… SyncLogStore LeftOuter · position_su… — · 2 actors HashJoin · LeftOuter · position_summary_mv_next.account_group_id = pnl_summary_mv_… HashJoin LeftOuter · position_su… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · pnl_summary_mv_next StreamScan pnl_summary_mv_next idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · portfolio_to_account_groups_mv.account_group_id = position_… SyncLogStore Inner · portfolio_to_ac… — · 2 actors HashJoin · Inner · portfolio_to_account_groups_mv.account_group_id = position_… HashJoin Inner · portfolio_to_ac… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Filter · position_summary_mv_next Filter position_summary_mv_next idle · 2 actors StreamScan · position_summary_mv_next StreamScan position_summary_mv_next idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore SyncLogStore — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · portfolio_to_account_groups_mv.portfolio_id = portfolios_dm… SyncLogStore Inner · portfolio_to_ac… — · 2 actors HashJoin · Inner · portfolio_to_account_groups_mv.portfolio_id = portfolios_dm… HashJoin Inner · portfolio_to_ac… 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 Filter · portfolio_to_account_groups_mv Filter portfolio_to_account_gr… idle · 2 actors StreamScan · portfolio_to_account_groups_mv StreamScan portfolio_to_account_gr… idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · ($expr13 >= $expr14), output_watermarks: [[$expr13]], outpu… Project ($expr13 >= $expr14), o… — · 2 actors DynamicFilter · ($expr13 >= $expr14), output_watermarks: [[$expr13]], outpu… DynamicFilter ($expr13 >= $expr14), o… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange 2/s · 0 actors Project Project — · 1 actor Now Now 2/s · 1 actor Project · ($expr13 >= $expr14), output_watermarks: [[$expr13]], outpu… Project ($expr13 >= $expr14), o… — · 2 actors Filter · ($expr13 >= $expr14), output_watermarks: [[$expr13]], outpu… Filter ($expr13 >= $expr14), o… idle · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · ($expr10 >= $expr11), output_watermarks: [[$expr10]], outpu… Project ($expr10 >= $expr11), o… — · 2 actors DynamicFilter · ($expr10 >= $expr11), output_watermarks: [[$expr10]], outpu… DynamicFilter ($expr10 >= $expr11), o… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 1 actor Now Now 2/s · 1 actor Project · ($expr10 >= $expr11), output_watermarks: [[$expr10]], outpu… Project ($expr10 >= $expr11), o… — · 2 actors Filter · ($expr10 >= $expr11), output_watermarks: [[$expr10]], outpu… Filter ($expr10 >= $expr11), o… idle · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: … Project ($expr7 >= $expr8), out… — · 2 actors DynamicFilter · ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: … DynamicFilter ($expr7 >= $expr8), out… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange 2/s · 0 actors Project Project — · 1 actor Now Now 2/s · 1 actor Project · ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: … Project ($expr7 >= $expr8), out… — · 2 actors Filter · ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: … Filter ($expr7 >= $expr8), out… idle · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: … Project ($expr4 >= $expr5), out… — · 2 actors DynamicFilter · ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: … DynamicFilter ($expr4 >= $expr5), out… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange 2/s · 0 actors Project Project — · 1 actor Now Now 2/s · 1 actor Project · ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: … Project ($expr4 >= $expr5), out… — · 2 actors Filter · ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: … Filter ($expr4 >= $expr5), out… idle · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: … Project ($expr1 >= $expr2), out… — · 2 actors DynamicFilter · ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: … DynamicFilter ($expr1 >= $expr2), out… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange 2/s · 0 actors Project Project — · 1 actor Now Now 2/s · 1 actor Project · ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: … Project ($expr1 >= $expr2), out… — · 2 actors Filter · ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: … Filter ($expr1 >= $expr2), out… idle · 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 24473 (Actor 114774,114775)
StreamMaterialize { columns: [portfolio_id, historic_range, series, portfolio_to_account_groups_mv.portfolio_id(hidden), portfolios_dm.base_currency_code(hidden), $src(hidden)], stream_key: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $src], pk_columns: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $src], pk_conflict: NoCheck }
├── output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1W':Varchar, jsonb_agg($expr3 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $src ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $src ]
└── StreamUnion { all: true } { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1W':Varchar, jsonb_agg($expr3 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $src ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $src ] }
    ├── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1W':Varchar, jsonb_agg($expr3 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 0:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    ├── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1M':Varchar, jsonb_agg($expr6 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 1:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    ├── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_3M':Varchar, jsonb_agg($expr9 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 2:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    ├── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_YTD':Varchar, jsonb_agg($expr12 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 3:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    ├── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1Y':Varchar, jsonb_agg($expr15 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 4:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    └── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_ALL':Varchar, jsonb_agg($expr16 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 5:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }

Fragment 24474 (Actor 114777,114776)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1W':Varchar, jsonb_agg($expr3 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 0:Int32] }
├── output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1W':Varchar, jsonb_agg($expr3 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 0:Int32 ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ]
└── StreamHashAgg { group_key: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code], aggs: [jsonb_agg($expr3 order_by(position_summary_mv_next.dim_balance_date ASC)), count] }
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, jsonb_agg($expr3 order_by(position_summary_mv_next.dim_balance_date ASC)), count ]
    ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ]
    └── MergeExecutor
        ├── output:
        │   ┌── portfolio_to_account_groups_mv.portfolio_id
        │   ├── portfolios_dm.base_currency_code
        │   ├── $expr3
        │   ├── position_summary_mv_next.dim_balance_date
        │   ├── position_summary_mv_next.source_entity_type
        │   ├── position_summary_mv_next.position_type
        │   ├── portfolio_to_account_groups_mv.account_group_id
        │   ├── pnl_summary_mv_next.position_summary_mv_next.source_entity_type
        │   ├── pnl_summary_mv_next.flow_summary_mv.dim_transaction_date
        │   ├── position_summary_mv_next.account_group_id
        │   └── position_summary_mv_next.currency_code
        └── stream key:
            ┌── portfolio_to_account_groups_mv.portfolio_id
            ├── position_summary_mv_next.source_entity_type
            ├── position_summary_mv_next.dim_balance_date
            ├── position_summary_mv_next.position_type
            ├── portfolio_to_account_groups_mv.account_group_id
            ├── portfolios_dm.base_currency_code
            ├── pnl_summary_mv_next.position_summary_mv_next.source_entity_type
            ├── pnl_summary_mv_next.flow_summary_mv.dim_transaction_date
            ├── position_summary_mv_next.account_group_id
            └── position_summary_mv_next.currency_code

Fragment 24475 (Actor 114799,114798)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, JsonbBuildObject('date':Varchar, position_summary_mv_next.dim_balance_date::Varchar, 'marketValue':Varchar, JsonbBuildObject('amount':Varchar, position_summary_mv_next.market_value::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'unrealizedGainLoss':Varchar, JsonbBuildObject('value':Varchar, JsonbBuildObject('amount':Varchar, pnl_summary_mv_next.unrealized_gain_loss::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'percentage':Varchar, Case((IsNull(pnl_summary_mv_next.total_average_cost) OR (pnl_summary_mv_next.total_average_cost = 0:Decimal)), null:Varchar, (pnl_summary_mv_next.unrealized_gain_loss / pnl_summary_mv_next.total_average_cost)::Varchar))) as $expr3, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr3, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── StreamDynamicFilter { predicate: ($expr1 >= $expr2), output_watermarks: [[$expr1]], output: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr1, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code], cleaned_by_watermark: true }
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr1, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, AtTimeZone(position_summary_mv_next.dim_balance_date::Timestamp, 'UTC':Varchar) as $expr1, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
    │   ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr1, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   └── StreamFilter { predicate: (portfolio_to_account_groups_mv.type = 'all':Varchar) }
    │       ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │       ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │       └── MergeExecutor
    │           ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │           └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    └── MergeExecutor { output: [ $expr2 ], stream key: [] }

Fragment 24476 (Actor 114795,114794)
StreamSyncLogStore
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── MergeExecutor
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]

Fragment 24477 (Actor 114788,114789)
StreamSyncLogStore
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── MergeExecutor
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]

Fragment 24478 (Actor 114780,114781)
StreamSyncLogStore
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── MergeExecutor
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]

Fragment 24479 (Actor 114801,114800)
StreamSyncLogStore
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── MergeExecutor
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]

Fragment 24480 (Actor 114778,114779)
StreamSyncLogStore
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── MergeExecutor
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]

Fragment 24481 (Actor 114797,114796)
StreamSyncLogStore
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── StreamHashJoin { type: LeftOuter, predicate: position_summary_mv_next.account_group_id = pnl_summary_mv_next.account_group_id AND position_summary_mv_next.position_type = pnl_summary_mv_next.position_type AND position_summary_mv_next.currency_code = pnl_summary_mv_next.currency_code AND position_summary_mv_next.dim_balance_date = pnl_summary_mv_next.dim_balance_date }
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }
    └── MergeExecutor { output: [ pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.position_type, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ], stream key: [ pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ] }

Fragment 24482 (Actor 114812,114813)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }

Fragment 24483 (Actor 114809,114808)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }

Fragment 24484 (Actor 114806,114807)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }

Fragment 24485 (Actor 114804,114805)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }

Fragment 24486 (Actor 114802,114803)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }

Fragment 24487 (Actor 114811,114810)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }
└── StreamHashJoin { type: Inner, predicate: portfolio_to_account_groups_mv.account_group_id = position_summary_mv_next.account_group_id AND portfolios_dm.base_currency_code = position_summary_mv_next.currency_code } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.source_entity_type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code ] }
    ├── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
    └── MergeExecutor { output: [ position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, position_summary_mv_next.source_entity_type ], stream key: [ position_summary_mv_next.account_group_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.currency_code, position_summary_mv_next.position_type ] }

Fragment 24488 (Actor 114819,114818)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }

Fragment 24489 (Actor 114817,114816)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }

Fragment 24490 (Actor 114821,114820)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }

Fragment 24491 (Actor 114822,114823)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }

Fragment 24492 (Actor 114824,114825)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
└── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }

Fragment 24493 (Actor 114814,114815)
StreamSyncLogStore { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
└── StreamHashJoin { type: Inner, predicate: portfolio_to_account_groups_mv.portfolio_id = portfolios_dm.portfolio_id } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, portfolios_dm.portfolio_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
    ├── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id, portfolios_dm.base_currency_code ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 24494 (Actor 114842,114841)
StreamFilter { predicate: (portfolio_to_account_groups_mv.type = 'all':Varchar) } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
└── StreamTableScan { table: portfolio_to_account_groups_mv, columns: [portfolio_id, account_group_id, type] } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id, portfolio_to_account_groups_mv.type ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
    ├── Upstream { output: [ portfolio_id, account_group_id, type ], stream key: [] }
    └── BatchPlanNode { output: [ portfolio_id, account_group_id, type ], stream key: [] }

Fragment 24495 (Actor 114844,114843)
StreamProject { exprs: [portfolios_dm.portfolio_id, portfolios_dm.base_currency_code] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.base_currency_code ], stream key: [ portfolios_dm.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_dm.closing_date) } { output: [ portfolios_dm.portfolio_id, portfolios_dm.base_currency_code, portfolios_dm.closing_date ], stream key: [ portfolios_dm.portfolio_id ] }
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, base_currency_code, closing_date] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.base_currency_code, portfolios_dm.closing_date ], stream key: [ portfolios_dm.portfolio_id ] }
        ├── Upstream { output: [ portfolio_id, base_currency_code, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, base_currency_code, closing_date ], stream key: [] }

Fragment 24496 (Actor 114845,114846)
StreamFilter { predicate: (position_summary_mv_next.position_type = 'POSITION':Varchar) } { output: [ position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, position_summary_mv_next.source_entity_type ], stream key: [ position_summary_mv_next.account_group_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.currency_code, position_summary_mv_next.position_type ] }
└── StreamTableScan { table: position_summary_mv_next, columns: [account_group_id, dim_balance_date, position_type, currency_code, market_value, source_entity_type] } { output: [ position_summary_mv_next.account_group_id, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, position_summary_mv_next.currency_code, position_summary_mv_next.market_value, position_summary_mv_next.source_entity_type ], stream key: [ position_summary_mv_next.account_group_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.currency_code, position_summary_mv_next.position_type ] }
    ├── Upstream { output: [ account_group_id, dim_balance_date, position_type, currency_code, market_value, source_entity_type ], stream key: [] }
    └── BatchPlanNode { output: [ account_group_id, dim_balance_date, position_type, currency_code, market_value, source_entity_type ], stream key: [] }

Fragment 24497 (Actor 114847,114848)
StreamTableScan { table: pnl_summary_mv_next, columns: [account_group_id, dim_balance_date, position_type, currency_code, total_average_cost, unrealized_gain_loss, position_summary_mv_next.source_entity_type, flow_summary_mv.dim_transaction_date] } { output: [ pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.position_type, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ], stream key: [ pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ] }
├── Upstream { output: [ account_group_id, dim_balance_date, position_type, currency_code, total_average_cost, unrealized_gain_loss, position_summary_mv_next.source_entity_type, flow_summary_mv.dim_transaction_date ], stream key: [] }
└── BatchPlanNode { output: [ account_group_id, dim_balance_date, position_type, currency_code, total_average_cost, unrealized_gain_loss, position_summary_mv_next.source_entity_type, flow_summary_mv.dim_transaction_date ], stream key: [] }

Fragment 24498 (Actor 114826)
StreamProject { exprs: [SubtractWithTimeZone(now, '7 days':Interval, 'UTC':Varchar) as $expr2], output_watermarks: [[$expr2]] } { output: [ $expr2 ], stream key: [] }
└── StreamNow { output: [ now ], stream key: [] }

Fragment 24499 (Actor 114827,114828)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1M':Varchar, jsonb_agg($expr6 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 1:Int32] } { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1M':Varchar, jsonb_agg($expr6 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 1:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
└── StreamHashAgg { group_key: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code], aggs: [jsonb_agg($expr6 order_by(position_summary_mv_next.dim_balance_date ASC)), count] } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, jsonb_agg($expr6 order_by(position_summary_mv_next.dim_balance_date ASC)), count ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    └── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr6, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ] }

Fragment 24500 (Actor 114782,114783)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, JsonbBuildObject('date':Varchar, position_summary_mv_next.dim_balance_date::Varchar, 'marketValue':Varchar, JsonbBuildObject('amount':Varchar, position_summary_mv_next.market_value::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'unrealizedGainLoss':Varchar, JsonbBuildObject('value':Varchar, JsonbBuildObject('amount':Varchar, pnl_summary_mv_next.unrealized_gain_loss::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'percentage':Varchar, Case((IsNull(pnl_summary_mv_next.total_average_cost) OR (pnl_summary_mv_next.total_average_cost = 0:Decimal)), null:Varchar, (pnl_summary_mv_next.unrealized_gain_loss / pnl_summary_mv_next.total_average_cost)::Varchar))) as $expr6, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr6, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── StreamDynamicFilter { predicate: ($expr4 >= $expr5), output_watermarks: [[$expr4]], output: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr4, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code], cleaned_by_watermark: true }
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr4, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, AtTimeZone(position_summary_mv_next.dim_balance_date::Timestamp, 'UTC':Varchar) as $expr4, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
    │   ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr4, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   └── StreamFilter { predicate: (portfolio_to_account_groups_mv.type = 'all':Varchar) }
    │       ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │       ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │       └── MergeExecutor
    │           ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │           └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    └── MergeExecutor { output: [ $expr5 ], stream key: [] }

Fragment 24501 (Actor 114829)
StreamProject { exprs: [SubtractWithTimeZone(now, '30 days':Interval, 'UTC':Varchar) as $expr5], output_watermarks: [[$expr5]] } { output: [ $expr5 ], stream key: [] }
└── StreamNow { output: [ now ], stream key: [] }

Fragment 24502 (Actor 114830,114831)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_3M':Varchar, jsonb_agg($expr9 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 2:Int32] } { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_3M':Varchar, jsonb_agg($expr9 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 2:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
└── StreamHashAgg { group_key: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code], aggs: [jsonb_agg($expr9 order_by(position_summary_mv_next.dim_balance_date ASC)), count] } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, jsonb_agg($expr9 order_by(position_summary_mv_next.dim_balance_date ASC)), count ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    └── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr9, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ] }

Fragment 24503 (Actor 114792,114793)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, JsonbBuildObject('date':Varchar, position_summary_mv_next.dim_balance_date::Varchar, 'marketValue':Varchar, JsonbBuildObject('amount':Varchar, position_summary_mv_next.market_value::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'unrealizedGainLoss':Varchar, JsonbBuildObject('value':Varchar, JsonbBuildObject('amount':Varchar, pnl_summary_mv_next.unrealized_gain_loss::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'percentage':Varchar, Case((IsNull(pnl_summary_mv_next.total_average_cost) OR (pnl_summary_mv_next.total_average_cost = 0:Decimal)), null:Varchar, (pnl_summary_mv_next.unrealized_gain_loss / pnl_summary_mv_next.total_average_cost)::Varchar))) as $expr9, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr9, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── StreamDynamicFilter { predicate: ($expr7 >= $expr8), output_watermarks: [[$expr7]], output: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr7, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code], cleaned_by_watermark: true }
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr7, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, AtTimeZone(position_summary_mv_next.dim_balance_date::Timestamp, 'UTC':Varchar) as $expr7, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
    │   ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr7, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   └── StreamFilter { predicate: (portfolio_to_account_groups_mv.type = 'all':Varchar) }
    │       ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │       ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │       └── MergeExecutor
    │           ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │           └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    └── MergeExecutor { output: [ $expr8 ], stream key: [] }

Fragment 24504 (Actor 114832)
StreamProject { exprs: [SubtractWithTimeZone(now, '90 days':Interval, 'UTC':Varchar) as $expr8], output_watermarks: [[$expr8]] } { output: [ $expr8 ], stream key: [] }
└── StreamNow { output: [ now ], stream key: [] }

Fragment 24505 (Actor 114834,114833)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_YTD':Varchar, jsonb_agg($expr12 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 3:Int32] } { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_YTD':Varchar, jsonb_agg($expr12 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 3:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
└── StreamHashAgg { group_key: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code], aggs: [jsonb_agg($expr12 order_by(position_summary_mv_next.dim_balance_date ASC)), count] } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, jsonb_agg($expr12 order_by(position_summary_mv_next.dim_balance_date ASC)), count ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    └── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr12, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ] }

Fragment 24506 (Actor 114790,114791)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, JsonbBuildObject('date':Varchar, position_summary_mv_next.dim_balance_date::Varchar, 'marketValue':Varchar, JsonbBuildObject('amount':Varchar, position_summary_mv_next.market_value::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'unrealizedGainLoss':Varchar, JsonbBuildObject('value':Varchar, JsonbBuildObject('amount':Varchar, pnl_summary_mv_next.unrealized_gain_loss::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'percentage':Varchar, Case((IsNull(pnl_summary_mv_next.total_average_cost) OR (pnl_summary_mv_next.total_average_cost = 0:Decimal)), null:Varchar, (pnl_summary_mv_next.unrealized_gain_loss / pnl_summary_mv_next.total_average_cost)::Varchar))) as $expr12, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr12, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── StreamDynamicFilter { predicate: ($expr10 >= $expr11), output_watermarks: [[$expr10]], output: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr10, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code], cleaned_by_watermark: true }
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr10, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, AtTimeZone(position_summary_mv_next.dim_balance_date::Timestamp, 'UTC':Varchar) as $expr10, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
    │   ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr10, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   └── StreamFilter { predicate: (portfolio_to_account_groups_mv.type = 'all':Varchar) }
    │       ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │       ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │       └── MergeExecutor
    │           ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │           └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    └── MergeExecutor { output: [ $expr11 ], stream key: [] }

Fragment 24507 (Actor 114835)
StreamProject { exprs: [DateTrunc('YEAR':Varchar, now, 'UTC':Varchar) as $expr11], output_watermarks: [[$expr11]] } { output: [ $expr11 ], stream key: [] }
└── StreamNow { output: [ now ], stream key: [] }

Fragment 24508 (Actor 114836,114837)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1Y':Varchar, jsonb_agg($expr15 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 4:Int32] } { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_1Y':Varchar, jsonb_agg($expr15 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 4:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
└── StreamHashAgg { group_key: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code], aggs: [jsonb_agg($expr15 order_by(position_summary_mv_next.dim_balance_date ASC)), count] } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, jsonb_agg($expr15 order_by(position_summary_mv_next.dim_balance_date ASC)), count ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    └── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr15, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ] }

Fragment 24509 (Actor 114787,114786)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, JsonbBuildObject('date':Varchar, position_summary_mv_next.dim_balance_date::Varchar, 'marketValue':Varchar, JsonbBuildObject('amount':Varchar, position_summary_mv_next.market_value::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'unrealizedGainLoss':Varchar, JsonbBuildObject('value':Varchar, JsonbBuildObject('amount':Varchar, pnl_summary_mv_next.unrealized_gain_loss::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'percentage':Varchar, Case((IsNull(pnl_summary_mv_next.total_average_cost) OR (pnl_summary_mv_next.total_average_cost = 0:Decimal)), null:Varchar, (pnl_summary_mv_next.unrealized_gain_loss / pnl_summary_mv_next.total_average_cost)::Varchar))) as $expr15, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr15, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── StreamDynamicFilter { predicate: ($expr13 >= $expr14), output_watermarks: [[$expr13]], output: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr13, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code], cleaned_by_watermark: true }
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr13, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    ├── StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, AtTimeZone(position_summary_mv_next.dim_balance_date::Timestamp, 'UTC':Varchar) as $expr13, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
    │   ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, $expr13, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │   └── StreamFilter { predicate: (portfolio_to_account_groups_mv.type = 'all':Varchar) }
    │       ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │       ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    │       └── MergeExecutor
    │           ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    │           └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    └── MergeExecutor { output: [ $expr14 ], stream key: [] }

Fragment 24510 (Actor 114838)
StreamProject { exprs: [SubtractWithTimeZone(now, '365 days':Interval, 'UTC':Varchar) as $expr14], output_watermarks: [[$expr14]] } { output: [ $expr14 ], stream key: [] }
└── StreamNow { output: [ now ], stream key: [] }

Fragment 24511 (Actor 114840,114839)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_ALL':Varchar, jsonb_agg($expr16 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 5:Int32] } { output: [ portfolio_to_account_groups_mv.portfolio_id, 'HISTORIC_RANGE_ALL':Varchar, jsonb_agg($expr16 order_by(position_summary_mv_next.dim_balance_date ASC)), portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, 5:Int32 ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
└── StreamHashAgg { group_key: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code], aggs: [jsonb_agg($expr16 order_by(position_summary_mv_next.dim_balance_date ASC)), count] } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, jsonb_agg($expr16 order_by(position_summary_mv_next.dim_balance_date ASC)), count ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code ] }
    └── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr16, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ], stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ] }

Fragment 24512 (Actor 114785,114784)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, JsonbBuildObject('date':Varchar, position_summary_mv_next.dim_balance_date::Varchar, 'marketValue':Varchar, JsonbBuildObject('amount':Varchar, position_summary_mv_next.market_value::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'unrealizedGainLoss':Varchar, JsonbBuildObject('value':Varchar, JsonbBuildObject('amount':Varchar, pnl_summary_mv_next.unrealized_gain_loss::Varchar, 'currencyCode':Varchar, portfolios_dm.base_currency_code), 'percentage':Varchar, Case((IsNull(pnl_summary_mv_next.total_average_cost) OR (pnl_summary_mv_next.total_average_cost = 0:Decimal)), null:Varchar, (pnl_summary_mv_next.unrealized_gain_loss / pnl_summary_mv_next.total_average_cost)::Varchar))) as $expr16, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code] }
├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolios_dm.base_currency_code, $expr16, position_summary_mv_next.dim_balance_date, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
└── StreamFilter { predicate: (portfolio_to_account_groups_mv.type = 'all':Varchar) }
    ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
    ├── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]
    └── MergeExecutor
        ├── output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.type, portfolios_dm.base_currency_code, position_summary_mv_next.dim_balance_date, position_summary_mv_next.market_value, pnl_summary_mv_next.total_average_cost, pnl_summary_mv_next.unrealized_gain_loss, position_summary_mv_next.source_entity_type, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code, pnl_summary_mv_next.account_group_id, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.dim_balance_date, pnl_summary_mv_next.currency_code, pnl_summary_mv_next.position_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date ]
        └── stream key: [ portfolio_to_account_groups_mv.portfolio_id, position_summary_mv_next.source_entity_type, position_summary_mv_next.dim_balance_date, position_summary_mv_next.position_type, portfolio_to_account_groups_mv.account_group_id, portfolios_dm.base_currency_code, pnl_summary_mv_next.position_summary_mv_next.source_entity_type, pnl_summary_mv_next.flow_summary_mv.dim_transaction_date, position_summary_mv_next.account_group_id, position_summary_mv_next.currency_code ]