client_portfolios_mv · explain

← cluster adib_rm objects client_portfolios_mv explain

materialized view · adib_rm.client_portfolios_mv profiled over 5s

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

Heat tree Operator graph

90 operators

Materialize · adib_rm.client_portfolios_mv

0% idle 2 actors

Project

— — 2 actors

HashAgg Aggregation state — unbounded unless keyed or temporally filtered

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · LeftOuter · portfolio_to_account_groups_mv.account_group_id = pnl_snaps…

— — 2 actors

HashJoin · LeftOuter · portfolio_to_account_groups_mv.account_group_id = pnl_snaps… 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 · pnl_snapshot_mv_next

0% idle 2 actors

StreamScan · pnl_snapshot_mv_next

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · LeftOuter · portfolio_to_account_groups_mv.account_group_id = intraday_…

— — 2 actors

HashJoin · LeftOuter · portfolio_to_account_groups_mv.account_group_id = intraday_… 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 · intraday_position_summary_mv_next

0% idle 2 actors

StreamScan · intraday_position_summary_mv_next

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · LeftOuter · portfolio_to_account_groups_mv.account_group_id = position_…

— — 2 actors

HashJoin · LeftOuter · 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

Project · position_snapshot_mv

— — 2 actors

Filter · position_snapshot_mv

0% idle 2 actors

StreamScan · position_snapshot_mv

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · LeftOuter · portfolios_plain_mv.portfolio_id = portfolio_to_account_gro…

— — 2 actors

HashJoin · LeftOuter · portfolios_plain_mv.portfolio_id = portfolio_to_account_gro… 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 · portfolio_to_account_groups_mv

— — 2 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

TemporalJoin · Inner · portfolios_plain_mv.service_type_id = service_types_dm.serv…

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · service_types_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = portfolios_plain_mv.po…

— — 2 actors

HashJoin · Inner · clients_portfolios_dm.portfolio_id = portfolios_plain_mv.po… 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_plain_mv

— — 2 actors

Filter · portfolios_plain_mv

0% idle 2 actors

StreamScan · portfolios_plain_mv

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · clients_dm.id = clients_portfolios_dm.client_id

— — 2 actors

HashJoin · Inner · clients_dm.id = clients_portfolios_dm.client_id Stateful hash join (4 state tables) — consider a temporal join for dimension lookups

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · clients_portfolios_dm

— — 2 actors

DynamicFilter · clients_portfolios_dm Dynamic filter — verify it pairs with a temporal condition to clean state

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% 2/s 0 actors

Now

0% 2/s 1 actor

Project · clients_portfolios_dm

— — 2 actors

Filter · clients_portfolios_dm

0% idle 2 actors

StreamScan · clients_portfolios_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · clients_dm

— — 2 actors

Filter · clients_dm

0% idle 2 actors

StreamScan · clients_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Heat = the operator's output-buffer backpressure over the sampling window. Click a node to fold its subtree.

Streaming operator plan from EXPLAIN ANALYZE. Node heat = backpressure. Drag to pan, scroll to zoom.

Fragments (DESCRIBE FRAGMENTS) — click to expand

Fragment 24623 (Actor 116311,116310)
StreamMaterialize { columns: [client_id, portfolios], stream_key: [client_id], pk_columns: [client_id], pk_conflict: NoCheck }
├── output: [ clients_dm.id, jsonb_agg($expr2 order_by(portfolios_plain_mv.portfolio_id ASC)) ]
├── stream key: [ clients_dm.id ]
└── StreamProject { exprs: [clients_dm.id, jsonb_agg($expr2 order_by(portfolios_plain_mv.portfolio_id ASC))] }
    ├── output: [ clients_dm.id, jsonb_agg($expr2 order_by(portfolios_plain_mv.portfolio_id ASC)) ]
    ├── stream key: [ clients_dm.id ]
    └── StreamHashAgg { group_key: [clients_dm.id], aggs: [jsonb_agg($expr2 order_by(portfolios_plain_mv.portfolio_id ASC)), count] }
        ├── output: [ clients_dm.id, jsonb_agg($expr2 order_by(portfolios_plain_mv.portfolio_id ASC)), count ]
        ├── stream key: [ clients_dm.id ]
        └── MergeExecutor
            ├── output:
            │   ┌── clients_dm.id
            │   ├── $expr2
            │   ├── portfolios_plain_mv.portfolio_id
            │   ├── clients_portfolios_dm.portfolio_id
            │   ├── clients_portfolios_dm.effective_start_date
            │   ├── portfolios_plain_mv.service_type_id
            │   ├── position_snapshot_mv.holding_values_latest_mv_next.type_expanded
            │   ├── position_snapshot_mv.flag
            │   ├── portfolio_to_account_groups_mv.account_group_id
            │   ├── portfolios_plain_mv.base_currency_code
            │   ├── intraday_position_summary_mv_next.position_type
            │   └── pnl_snapshot_mv_next.position_type
            └── stream key:
                ┌── clients_dm.id
                ├── clients_portfolios_dm.portfolio_id
                ├── clients_portfolios_dm.effective_start_date
                ├── portfolios_plain_mv.service_type_id
                ├── portfolios_plain_mv.portfolio_id
                ├── position_snapshot_mv.holding_values_latest_mv_next.type_expanded
                ├── position_snapshot_mv.flag
                ├── portfolio_to_account_groups_mv.account_group_id
                ├── portfolios_plain_mv.base_currency_code
                ├── intraday_position_summary_mv_next.position_type
                └── pnl_snapshot_mv_next.position_type

Fragment 24624 (Actor 116312,116313)
StreamProject
└─exprs:
  ┌─clients_dm.id
  ├─JsonbBuildObject('id':Varchar, portfolios_plain_mv.portfolio_id, 'name':Varchar, portfolios_plain_mv.name, 'number':Varchar, portfolios_plain_mv.number, 'serviceTypeId':Varchar, portfolios_plain_mv.service_type_id, 'serviceType':Varchar, service_types_dm.type, 'marketValue':Varchar, JsonbBuildObject('amount':Varchar, Coalesce(intraday_position_summary_mv_next.market_value, position_snapshot_mv.market_value)::Varchar, 'currencyCode':Varchar, portfolios_plain_mv.base_currency_code), 'marketValueSystemCurrency':Varchar, JsonbBuildObject('amount':Varchar, Case(Not(IsNull(intraday_position_summary_mv_next.market_value)), intraday_position_summary_mv_next.market_value_system_currency::Varchar, position_snapshot_mv.market_value_system_currency::Varchar), 'currencyCode':Varchar, clients_dm.base_currency_code), 'unrealizedGainLoss':Varchar, JsonbBuildObject('value':Varchar, JsonbBuildObject('amount':Varchar, Case(Not(IsNull(intraday_position_summary_mv_next.market_value)), (intraday_position_summary_mv_next.market_value - intraday_position_summary_mv_next.total_average_cost)::Varchar, pnl_snapshot_mv_next.unrealized_gain_loss::Varchar), 'currencyCode':Varchar, portfolios_plain_mv.base_currency_code), 'percentage':Varchar, Case(Not(IsNull(intraday_position_summary_mv_next.market_value)), Case((IsNull(intraday_position_summary_mv_next.total_average_cost) OR (intraday_position_summary_mv_next.total_average_cost = 0:Decimal)), null:Varchar, ((intraday_position_summary_mv_next.market_value - intraday_position_summary_mv_next.total_average_cost) / intraday_position_summary_mv_next.total_average_cost)::Varchar), Case((IsNull(pnl_snapshot_mv_next.total_average_cost) OR (pnl_snapshot_mv_next.total_average_cost = 0:Decimal)), null:Varchar, (pnl_snapshot_mv_next.unrealized_gain_loss / pnl_snapshot_mv_next.total_average_cost)::Varchar)))) as $expr2
  ├─portfolios_plain_mv.portfolio_id
  ├─clients_portfolios_dm.portfolio_id
  ├─clients_portfolios_dm.effective_start_date
  ├─portfolios_plain_mv.service_type_id
  ├─position_snapshot_mv.holding_values_latest_mv_next.type_expanded
  ├─position_snapshot_mv.flag
  ├─portfolio_to_account_groups_mv.account_group_id
  ├─portfolios_plain_mv.base_currency_code
  ├─intraday_position_summary_mv_next.position_type
  └─pnl_snapshot_mv_next.position_type
├── output: [ clients_dm.id, $expr2, portfolios_plain_mv.portfolio_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code, intraday_position_summary_mv_next.position_type, pnl_snapshot_mv_next.position_type ]
├── stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code, intraday_position_summary_mv_next.position_type, pnl_snapshot_mv_next.position_type ]
└── MergeExecutor { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, pnl_snapshot_mv_next.unrealized_gain_loss, pnl_snapshot_mv_next.total_average_cost, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, intraday_position_summary_mv_next.position_type, pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.position_type, pnl_snapshot_mv_next.currency_code ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code, intraday_position_summary_mv_next.position_type, pnl_snapshot_mv_next.position_type ] }

Fragment 24625 (Actor 116319,116318)
StreamSyncLogStore { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, pnl_snapshot_mv_next.unrealized_gain_loss, pnl_snapshot_mv_next.total_average_cost, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, intraday_position_summary_mv_next.position_type, pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.position_type, pnl_snapshot_mv_next.currency_code ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code, intraday_position_summary_mv_next.position_type, pnl_snapshot_mv_next.position_type ] }
└── StreamHashJoin { type: LeftOuter, predicate: portfolio_to_account_groups_mv.account_group_id = pnl_snapshot_mv_next.account_group_id AND portfolios_plain_mv.base_currency_code = pnl_snapshot_mv_next.currency_code } { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, pnl_snapshot_mv_next.unrealized_gain_loss, pnl_snapshot_mv_next.total_average_cost, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, intraday_position_summary_mv_next.position_type, pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.position_type, pnl_snapshot_mv_next.currency_code ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code, intraday_position_summary_mv_next.position_type, pnl_snapshot_mv_next.position_type ] }
    ├── MergeExecutor { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.position_type ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code, intraday_position_summary_mv_next.position_type ] }
    └── MergeExecutor { output: [ pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.currency_code, pnl_snapshot_mv_next.unrealized_gain_loss, pnl_snapshot_mv_next.total_average_cost, pnl_snapshot_mv_next.position_type ], stream key: [ pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.position_type, pnl_snapshot_mv_next.currency_code ] }

Fragment 24626 (Actor 116314,116315)
StreamSyncLogStore { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.position_type ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code, intraday_position_summary_mv_next.position_type ] }
└── StreamHashJoin { type: LeftOuter, predicate: portfolio_to_account_groups_mv.account_group_id = intraday_position_summary_mv_next.account_group_id AND portfolios_plain_mv.base_currency_code = intraday_position_summary_mv_next.currency_code } { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.position_type ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code, intraday_position_summary_mv_next.position_type ] }
    ├── MergeExecutor { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code ] }
    └── MergeExecutor { output: [ intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, intraday_position_summary_mv_next.position_type ], stream key: [ intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.position_type ] }

Fragment 24627 (Actor 116316,116317)
StreamSyncLogStore { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code ] }
└── StreamHashJoin { type: LeftOuter, predicate: portfolio_to_account_groups_mv.account_group_id = position_snapshot_mv.account_group_id AND portfolios_plain_mv.base_currency_code = position_snapshot_mv.currency_code } { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, portfolio_to_account_groups_mv.account_group_id, portfolios_plain_mv.base_currency_code ] }
    ├── MergeExecutor { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolio_to_account_groups_mv.portfolio_id ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id ] }
    └── MergeExecutor { output: [ position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ], stream key: [ position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ] }

Fragment 24628 (Actor 116320,116321)
StreamSyncLogStore { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolio_to_account_groups_mv.portfolio_id ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id ] }
└── StreamHashJoin { type: LeftOuter, predicate: portfolios_plain_mv.portfolio_id = portfolio_to_account_groups_mv.portfolio_id } { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, portfolio_to_account_groups_mv.account_group_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolio_to_account_groups_mv.portfolio_id ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id, portfolios_plain_mv.portfolio_id ] }
    ├── MergeExecutor { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, service_types_dm.service_type_id ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id ] }
    └── MergeExecutor { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }

Fragment 24629 (Actor 116240,116239)
StreamTemporalJoin { type: Inner, append_only: false, predicate: portfolios_plain_mv.service_type_id = service_types_dm.service_type_id, nested_loop: false } { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, service_types_dm.type, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, service_types_dm.service_type_id ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_plain_mv.service_type_id ] }
├── MergeExecutor { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── MergeExecutor { output: [ service_types_dm.service_type_id, service_types_dm.type ], stream key: [ service_types_dm.service_type_id ] }

Fragment 24630 (Actor 116323,116322)
StreamSyncLogStore { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.portfolio_id = portfolios_plain_mv.portfolio_id } { output: [ clients_dm.id, clients_dm.base_currency_code, portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_dm.id, clients_dm.base_currency_code, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id ], stream key: [ portfolios_plain_mv.portfolio_id ] }

Fragment 24631 (Actor 116325,116324)
StreamSyncLogStore { output: [ clients_dm.id, clients_dm.base_currency_code, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_dm.id = clients_portfolios_dm.client_id } { output: [ clients_dm.id, clients_dm.base_currency_code, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_dm.id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_dm.id, clients_dm.base_currency_code ], stream key: [ clients_dm.id ] }
    └── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 24632 (Actor 116327,116326)
StreamProject { exprs: [clients_dm.id, clients_dm.base_currency_code] } { output: [ clients_dm.id, clients_dm.base_currency_code ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.disabled_at) } { output: [ clients_dm.id, clients_dm.base_currency_code, clients_dm.disabled_at ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, base_currency_code, disabled_at] } { output: [ clients_dm.id, clients_dm.base_currency_code, clients_dm.disabled_at ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, base_currency_code, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ id, base_currency_code, disabled_at ], stream key: [] }

Fragment 24633 (Actor 116328,116329)
StreamProject { exprs: [clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamDynamicFilter { predicate: ($expr1 <= now), output: [clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, $expr1, clients_portfolios_dm.effective_start_date], cleaned_by_watermark: true } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, $expr1, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── StreamProject { exprs: [clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, AtTimeZone(clients_portfolios_dm.effective_start_date::Timestamp, 'UTC':Varchar) as $expr1, clients_portfolios_dm.effective_start_date] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, $expr1, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    │   └── StreamFilter { predicate: IsNull(clients_portfolios_dm.disabled_at) AND IsNull(clients_portfolios_dm.effective_end_date) } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at, clients_portfolios_dm.effective_end_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    │       └── StreamTableScan { table: clients_portfolios_dm, columns: [client_id, portfolio_id, effective_start_date, disabled_at, effective_end_date] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at, clients_portfolios_dm.effective_end_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    │           ├── Upstream { output: [ client_id, portfolio_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }
    │           └── BatchPlanNode { output: [ client_id, portfolio_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }
    └── MergeExecutor { output: [ now ], stream key: [] }

Fragment 24634 (Actor 116330)
StreamNow { output: [ now ], stream key: [] }

Fragment 24635 (Actor 116331,116332)
StreamProject { exprs: [portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id] } { output: [ portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id ], stream key: [ portfolios_plain_mv.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_plain_mv.closing_date) } { output: [ portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, portfolios_plain_mv.closing_date ], stream key: [ portfolios_plain_mv.portfolio_id ] }
    └── StreamTableScan { table: portfolios_plain_mv, columns: [portfolio_id, name, number, base_currency_code, service_type_id, closing_date] } { output: [ portfolios_plain_mv.portfolio_id, portfolios_plain_mv.name, portfolios_plain_mv.number, portfolios_plain_mv.base_currency_code, portfolios_plain_mv.service_type_id, portfolios_plain_mv.closing_date ], stream key: [ portfolios_plain_mv.portfolio_id ] }
        ├── Upstream { output: [ portfolio_id, name, number, base_currency_code, service_type_id, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, name, number, base_currency_code, service_type_id, closing_date ], stream key: [] }

Fragment 24636 (Actor 116242,116241)
StreamTableScan { table: service_types_dm, columns: [service_type_id, type] } { output: [ service_types_dm.service_type_id, service_types_dm.type ], stream key: [ service_types_dm.service_type_id ] }
├── Upstream { output: [ service_type_id, type ], stream key: [] }
└── BatchPlanNode { output: [ service_type_id, type ], stream key: [] }

Fragment 24637 (Actor 116334,116333)
StreamProject { exprs: [portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id] } { output: [ portfolio_to_account_groups_mv.portfolio_id, portfolio_to_account_groups_mv.account_group_id ], stream key: [ portfolio_to_account_groups_mv.portfolio_id ] }
└── 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 24638 (Actor 116336,116335)
StreamProject { exprs: [position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag] } { output: [ position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ], stream key: [ position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ] }
└── StreamFilter { predicate: (position_snapshot_mv.position_type = 'POSITION':Varchar) } { output: [ position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, position_snapshot_mv.position_type ], stream key: [ position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ] }
    └── StreamTableScan { table: position_snapshot_mv, columns: [account_group_id, currency_code, market_value, market_value_system_currency, holding_values_latest_mv_next.type_expanded, flag, position_type] } { output: [ position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.market_value, position_snapshot_mv.market_value_system_currency, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag, position_snapshot_mv.position_type ], stream key: [ position_snapshot_mv.account_group_id, position_snapshot_mv.currency_code, position_snapshot_mv.holding_values_latest_mv_next.type_expanded, position_snapshot_mv.flag ] }
        ├── Upstream { output: [ account_group_id, currency_code, market_value, market_value_system_currency, holding_values_latest_mv_next.type_expanded, flag, position_type ], stream key: [] }
        └── BatchPlanNode { output: [ account_group_id, currency_code, market_value, market_value_system_currency, holding_values_latest_mv_next.type_expanded, flag, position_type ], stream key: [] }

Fragment 24639 (Actor 116338,116337)
StreamFilter { predicate: (intraday_position_summary_mv_next.position_type = 'POSITION':Varchar) } { output: [ intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, intraday_position_summary_mv_next.position_type ], stream key: [ intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.position_type ] }
└── StreamTableScan { table: intraday_position_summary_mv_next, columns: [account_group_id, currency_code, market_value, total_average_cost, market_value_system_currency, position_type] } { output: [ intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.market_value, intraday_position_summary_mv_next.total_average_cost, intraday_position_summary_mv_next.market_value_system_currency, intraday_position_summary_mv_next.position_type ], stream key: [ intraday_position_summary_mv_next.account_group_id, intraday_position_summary_mv_next.currency_code, intraday_position_summary_mv_next.position_type ] }
    ├── Upstream { output: [ account_group_id, currency_code, market_value, total_average_cost, market_value_system_currency, position_type ], stream key: [] }
    └── BatchPlanNode { output: [ account_group_id, currency_code, market_value, total_average_cost, market_value_system_currency, position_type ], stream key: [] }

Fragment 24640 (Actor 116339,116340)
StreamFilter { predicate: (pnl_snapshot_mv_next.position_type = 'POSITION':Varchar) } { output: [ pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.currency_code, pnl_snapshot_mv_next.unrealized_gain_loss, pnl_snapshot_mv_next.total_average_cost, pnl_snapshot_mv_next.position_type ], stream key: [ pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.position_type, pnl_snapshot_mv_next.currency_code ] }
└── StreamTableScan { table: pnl_snapshot_mv_next, columns: [account_group_id, currency_code, unrealized_gain_loss, total_average_cost, position_type] } { output: [ pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.currency_code, pnl_snapshot_mv_next.unrealized_gain_loss, pnl_snapshot_mv_next.total_average_cost, pnl_snapshot_mv_next.position_type ], stream key: [ pnl_snapshot_mv_next.account_group_id, pnl_snapshot_mv_next.position_type, pnl_snapshot_mv_next.currency_code ] }
    ├── Upstream { output: [ account_group_id, currency_code, unrealized_gain_loss, total_average_cost, position_type ], stream key: [] }
    └── BatchPlanNode { output: [ account_group_id, currency_code, unrealized_gain_loss, total_average_cost, position_type ], stream key: [] }

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