RWM Console cluster: risingwave-ai-hub-int.ai-hub-rwm.svc.cluster.local

← cluster alpheya_agent objects party_position_by_asset_mv explain
Overview Objects Graph History
materialized view · alpheya_agent.party_position_by_asset_mv profiled over 5s
seconds (1–30)

Job is idle — throughput ~0; structure shown.

Stateful hash join (4 state tables) — consider a temporal join for dimension lookupsAggregation state — unbounded unless keyed or temporally filtered
25 operators
Materialize · alpheya_agent.party_position_by_asset_mv
0% idle 2 actors
Project
— — 2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · position_by_asset_mv_next.account_group_id = party_to_accou…
— — 2 actors
HashJoin · Inner · position_by_asset_mv_next.account_group_id = party_to_accou… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
StreamScan · party_to_account_groups_mv
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Filter · position_by_asset_mv_next
0% idle 2 actors
StreamScan · position_by_asset_mv_next
0% idle 2 actors
BatchPlan
— — 2 actors
Merge
— — 2 actors
Heat = the operator's output-buffer backpressure over the sampling window. Click a node to fold its subtree.
Materialize · alpheya_agent.party_position_by_asset_mv Materialize alpheya_agent.party_pos… idle · 2 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · position_by_asset_mv_next.account_group_id = party_to_accou… SyncLogStore Inner · position_by_ass… — · 2 actors HashJoin · Inner · position_by_asset_mv_next.account_group_id = party_to_accou… HashJoin Inner · position_by_ass… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · party_to_account_groups_mv StreamScan party_to_account_groups… idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Filter · position_by_asset_mv_next Filter position_by_asset_mv_ne… idle · 2 actors StreamScan · position_by_asset_mv_next StreamScan position_by_asset_mv_ne… idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors
Streaming operator plan from EXPLAIN ANALYZE. Node heat = backpressure. Drag to pan, scroll to zoom.
Fragments (DESCRIBE FRAGMENTS) — click to expand 
Fragment 20187 (Actor 167110,167111)
StreamMaterialize { columns: [party_id, account_group_type, as_of_date, asset_id, market_value_system, system_currency_code], stream_key: [as_of_date, asset_id, party_id, account_group_type], pk_columns: [as_of_date, asset_id, party_id, account_group_type], pk_conflict: NoCheck }
├── output: [ party_to_account_groups_mv.party_id, party_to_account_groups_mv.type, position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, sum(position_by_asset_mv_next.market_value_system_currency), 'USD':Varchar ]
├── stream key: [ position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, party_to_account_groups_mv.party_id, party_to_account_groups_mv.type ]
└── StreamProject { exprs: [party_to_account_groups_mv.party_id, party_to_account_groups_mv.type, position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, sum(position_by_asset_mv_next.market_value_system_currency), 'USD':Varchar] }
    ├── output: [ party_to_account_groups_mv.party_id, party_to_account_groups_mv.type, position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, sum(position_by_asset_mv_next.market_value_system_currency), 'USD':Varchar ]
    ├── stream key: [ position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, party_to_account_groups_mv.party_id, party_to_account_groups_mv.type ]
    └── StreamHashAgg { group_key: [position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, party_to_account_groups_mv.party_id, party_to_account_groups_mv.type], aggs: [sum(position_by_asset_mv_next.market_value_system_currency), count] }
        ├── output: [ position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, party_to_account_groups_mv.party_id, party_to_account_groups_mv.type, sum(position_by_asset_mv_next.market_value_system_currency), count ]
        ├── stream key: [ position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, party_to_account_groups_mv.party_id, party_to_account_groups_mv.type ]
        └── StreamLocalityProvider { locality_columns: [position_by_asset_mv_next.dim_balance_date, position_by_asset_mv_next.asset_id, party_to_account_groups_mv.party_id, party_to_account_groups_mv.type] }
            ├── output:
            │   ┌── position_by_asset_mv_next.dim_balance_date
            │   ├── position_by_asset_mv_next.asset_id
            │   ├── position_by_asset_mv_next.market_value_system_currency
            │   ├── party_to_account_groups_mv.party_id
            │   ├── party_to_account_groups_mv.type
            │   ├── position_by_asset_mv_next.account_group_id
            │   ├── position_by_asset_mv_next.position_type
            │   ├── position_by_asset_mv_next.currency_code
            │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
            │   ├── position_by_asset_mv_next.asset_currency
            │   ├── position_by_asset_mv_next.source_entity_type
            │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
            │   ├── position_by_asset_mv_next.flag
            │   ├── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
            │   ├── party_to_account_groups_mv.account_group_id
            │   ├── party_to_account_groups_mv.parties.id
            │   ├── party_to_account_groups_mv.null:Varchar
            │   └── party_to_account_groups_mv.$src
            ├── stream key:
            │   ┌── position_by_asset_mv_next.dim_balance_date
            │   ├── position_by_asset_mv_next.asset_id
            │   ├── party_to_account_groups_mv.party_id
            │   ├── party_to_account_groups_mv.type
            │   ├── position_by_asset_mv_next.account_group_id
            │   ├── position_by_asset_mv_next.position_type
            │   ├── position_by_asset_mv_next.currency_code
            │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
            │   ├── position_by_asset_mv_next.asset_currency
            │   ├── position_by_asset_mv_next.source_entity_type
            │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
            │   ├── position_by_asset_mv_next.flag
            │   ├── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
            │   ├── party_to_account_groups_mv.parties.id
            │   ├── party_to_account_groups_mv.null:Varchar
            │   └── party_to_account_groups_mv.$src
            └── MergeExecutor
                ├── output:
                │   ┌── position_by_asset_mv_next.dim_balance_date
                │   ├── position_by_asset_mv_next.asset_id
                │   ├── position_by_asset_mv_next.market_value_system_currency
                │   ├── party_to_account_groups_mv.party_id
                │   ├── party_to_account_groups_mv.type
                │   ├── position_by_asset_mv_next.account_group_id
                │   ├── position_by_asset_mv_next.position_type
                │   ├── position_by_asset_mv_next.currency_code
                │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
                │   ├── position_by_asset_mv_next.asset_currency
                │   ├── position_by_asset_mv_next.source_entity_type
                │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
                │   ├── position_by_asset_mv_next.flag
                │   ├── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
                │   ├── party_to_account_groups_mv.account_group_id
                │   ├── party_to_account_groups_mv.parties.id
                │   ├── party_to_account_groups_mv.null:Varchar
                │   └── party_to_account_groups_mv.$src
                └── stream key:
                    ┌── position_by_asset_mv_next.account_group_id
                    ├── position_by_asset_mv_next.dim_balance_date
                    ├── position_by_asset_mv_next.position_type
                    ├── position_by_asset_mv_next.currency_code
                    ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
                    ├── position_by_asset_mv_next.asset_currency
                    ├── position_by_asset_mv_next.source_entity_type
                    ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
                    ├── position_by_asset_mv_next.asset_id
                    ├── position_by_asset_mv_next.flag
                    ├── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
                    ├── party_to_account_groups_mv.parties.id
                    ├── party_to_account_groups_mv.null:Varchar
                    └── party_to_account_groups_mv.$src

Fragment 20188 (Actor 167105,167104)
StreamSyncLogStore
├── output:
│   ┌── position_by_asset_mv_next.dim_balance_date
│   ├── position_by_asset_mv_next.asset_id
│   ├── position_by_asset_mv_next.market_value_system_currency
│   ├── party_to_account_groups_mv.party_id
│   ├── party_to_account_groups_mv.type
│   ├── position_by_asset_mv_next.account_group_id
│   ├── position_by_asset_mv_next.position_type
│   ├── position_by_asset_mv_next.currency_code
│   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
│   ├── position_by_asset_mv_next.asset_currency
│   ├── position_by_asset_mv_next.source_entity_type
│   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
│   ├── position_by_asset_mv_next.flag
│   ├── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
│   ├── party_to_account_groups_mv.account_group_id
│   ├── party_to_account_groups_mv.parties.id
│   ├── party_to_account_groups_mv.null:Varchar
│   └── party_to_account_groups_mv.$src
├── stream key:
│   ┌── position_by_asset_mv_next.account_group_id
│   ├── position_by_asset_mv_next.dim_balance_date
│   ├── position_by_asset_mv_next.position_type
│   ├── position_by_asset_mv_next.currency_code
│   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
│   ├── position_by_asset_mv_next.asset_currency
│   ├── position_by_asset_mv_next.source_entity_type
│   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
│   ├── position_by_asset_mv_next.asset_id
│   ├── position_by_asset_mv_next.flag
│   ├── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
│   ├── party_to_account_groups_mv.parties.id
│   ├── party_to_account_groups_mv.null:Varchar
│   └── party_to_account_groups_mv.$src
└── StreamHashJoin { type: Inner, predicate: position_by_asset_mv_next.account_group_id = party_to_account_groups_mv.account_group_id }
    ├── output:
    │   ┌── position_by_asset_mv_next.dim_balance_date
    │   ├── position_by_asset_mv_next.asset_id
    │   ├── position_by_asset_mv_next.market_value_system_currency
    │   ├── party_to_account_groups_mv.party_id
    │   ├── party_to_account_groups_mv.type
    │   ├── position_by_asset_mv_next.account_group_id
    │   ├── position_by_asset_mv_next.position_type
    │   ├── position_by_asset_mv_next.currency_code
    │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
    │   ├── position_by_asset_mv_next.asset_currency
    │   ├── position_by_asset_mv_next.source_entity_type
    │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
    │   ├── position_by_asset_mv_next.flag
    │   ├── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
    │   ├── party_to_account_groups_mv.account_group_id
    │   ├── party_to_account_groups_mv.parties.id
    │   ├── party_to_account_groups_mv.null:Varchar
    │   └── party_to_account_groups_mv.$src
    ├── stream key:
    │   ┌── position_by_asset_mv_next.account_group_id
    │   ├── position_by_asset_mv_next.dim_balance_date
    │   ├── position_by_asset_mv_next.position_type
    │   ├── position_by_asset_mv_next.currency_code
    │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
    │   ├── position_by_asset_mv_next.asset_currency
    │   ├── position_by_asset_mv_next.source_entity_type
    │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
    │   ├── position_by_asset_mv_next.asset_id
    │   ├── position_by_asset_mv_next.flag
    │   ├── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
    │   ├── party_to_account_groups_mv.parties.id
    │   ├── party_to_account_groups_mv.null:Varchar
    │   └── party_to_account_groups_mv.$src
    ├── MergeExecutor
    │   ├── output:
    │   │   ┌── position_by_asset_mv_next.account_group_id
    │   │   ├── position_by_asset_mv_next.dim_balance_date
    │   │   ├── position_by_asset_mv_next.asset_id
    │   │   ├── position_by_asset_mv_next.market_value_system_currency
    │   │   ├── position_by_asset_mv_next.position_type
    │   │   ├── position_by_asset_mv_next.currency_code
    │   │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
    │   │   ├── position_by_asset_mv_next.asset_currency
    │   │   ├── position_by_asset_mv_next.source_entity_type
    │   │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
    │   │   ├── position_by_asset_mv_next.flag
    │   │   └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
    │   └── stream key:
    │       ┌── position_by_asset_mv_next.account_group_id
    │       ├── position_by_asset_mv_next.dim_balance_date
    │       ├── position_by_asset_mv_next.position_type
    │       ├── position_by_asset_mv_next.currency_code
    │       ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
    │       ├── position_by_asset_mv_next.asset_currency
    │       ├── position_by_asset_mv_next.source_entity_type
    │       ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
    │       ├── position_by_asset_mv_next.asset_id
    │       ├── position_by_asset_mv_next.flag
    │       └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
    └── MergeExecutor
        ├── output: [ party_to_account_groups_mv.party_id, party_to_account_groups_mv.account_group_id, party_to_account_groups_mv.type, party_to_account_groups_mv.parties.id, party_to_account_groups_mv.null:Varchar, party_to_account_groups_mv.$src ]
        └── stream key: [ party_to_account_groups_mv.account_group_id, party_to_account_groups_mv.parties.id, party_to_account_groups_mv.null:Varchar, party_to_account_groups_mv.$src ]

Fragment 20189 (Actor 167109,167108)
StreamLocalityProvider { locality_columns: [position_by_asset_mv_next.account_group_id] }
├── output:
│   ┌── position_by_asset_mv_next.account_group_id
│   ├── position_by_asset_mv_next.dim_balance_date
│   ├── position_by_asset_mv_next.asset_id
│   ├── position_by_asset_mv_next.market_value_system_currency
│   ├── position_by_asset_mv_next.position_type
│   ├── position_by_asset_mv_next.currency_code
│   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
│   ├── position_by_asset_mv_next.asset_currency
│   ├── position_by_asset_mv_next.source_entity_type
│   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
│   ├── position_by_asset_mv_next.flag
│   └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
├── stream key:
│   ┌── position_by_asset_mv_next.account_group_id
│   ├── position_by_asset_mv_next.dim_balance_date
│   ├── position_by_asset_mv_next.position_type
│   ├── position_by_asset_mv_next.currency_code
│   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
│   ├── position_by_asset_mv_next.asset_currency
│   ├── position_by_asset_mv_next.source_entity_type
│   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
│   ├── position_by_asset_mv_next.asset_id
│   ├── position_by_asset_mv_next.flag
│   └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
└── MergeExecutor
    ├── output:
    │   ┌── position_by_asset_mv_next.account_group_id
    │   ├── position_by_asset_mv_next.dim_balance_date
    │   ├── position_by_asset_mv_next.asset_id
    │   ├── position_by_asset_mv_next.market_value_system_currency
    │   ├── position_by_asset_mv_next.position_type
    │   ├── position_by_asset_mv_next.currency_code
    │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
    │   ├── position_by_asset_mv_next.asset_currency
    │   ├── position_by_asset_mv_next.source_entity_type
    │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
    │   ├── position_by_asset_mv_next.flag
    │   └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
    └── stream key:
        ┌── position_by_asset_mv_next.account_group_id
        ├── position_by_asset_mv_next.dim_balance_date
        ├── position_by_asset_mv_next.position_type
        ├── position_by_asset_mv_next.currency_code
        ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
        ├── position_by_asset_mv_next.asset_currency
        ├── position_by_asset_mv_next.source_entity_type
        ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
        ├── position_by_asset_mv_next.asset_id
        ├── position_by_asset_mv_next.flag
        └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type

Fragment 20190 (Actor 167106,167107)
StreamFilter { predicate: (position_by_asset_mv_next.source_entity_type = 'party':Varchar) AND (position_by_asset_mv_next.position_type = 'POSITION':Varchar) }
├── output:
│   ┌── position_by_asset_mv_next.account_group_id
│   ├── position_by_asset_mv_next.dim_balance_date
│   ├── position_by_asset_mv_next.asset_id
│   ├── position_by_asset_mv_next.market_value_system_currency
│   ├── position_by_asset_mv_next.position_type
│   ├── position_by_asset_mv_next.currency_code
│   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
│   ├── position_by_asset_mv_next.asset_currency
│   ├── position_by_asset_mv_next.source_entity_type
│   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
│   ├── position_by_asset_mv_next.flag
│   └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
├── stream key:
│   ┌── position_by_asset_mv_next.account_group_id
│   ├── position_by_asset_mv_next.dim_balance_date
│   ├── position_by_asset_mv_next.position_type
│   ├── position_by_asset_mv_next.currency_code
│   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
│   ├── position_by_asset_mv_next.asset_currency
│   ├── position_by_asset_mv_next.source_entity_type
│   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
│   ├── position_by_asset_mv_next.asset_id
│   ├── position_by_asset_mv_next.flag
│   └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
└── StreamTableScan { table: position_by_asset_mv_next, columns: [account_group_id, dim_balance_date, asset_id, market_value_system_currency, position_type, currency_code, position_values_mv_next.holding_currency_expanded, asset_currency, source_entity_type, position_values_mv_next.position_type_expanded, flag, position_summary_mv_next.source_entity_type] }
    ├── output:
    │   ┌── position_by_asset_mv_next.account_group_id
    │   ├── position_by_asset_mv_next.dim_balance_date
    │   ├── position_by_asset_mv_next.asset_id
    │   ├── position_by_asset_mv_next.market_value_system_currency
    │   ├── position_by_asset_mv_next.position_type
    │   ├── position_by_asset_mv_next.currency_code
    │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
    │   ├── position_by_asset_mv_next.asset_currency
    │   ├── position_by_asset_mv_next.source_entity_type
    │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
    │   ├── position_by_asset_mv_next.flag
    │   └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
    ├── stream key:
    │   ┌── position_by_asset_mv_next.account_group_id
    │   ├── position_by_asset_mv_next.dim_balance_date
    │   ├── position_by_asset_mv_next.position_type
    │   ├── position_by_asset_mv_next.currency_code
    │   ├── position_by_asset_mv_next.position_values_mv_next.holding_currency_expanded
    │   ├── position_by_asset_mv_next.asset_currency
    │   ├── position_by_asset_mv_next.source_entity_type
    │   ├── position_by_asset_mv_next.position_values_mv_next.position_type_expanded
    │   ├── position_by_asset_mv_next.asset_id
    │   ├── position_by_asset_mv_next.flag
    │   └── position_by_asset_mv_next.position_summary_mv_next.source_entity_type
    ├── Upstream { output: [ account_group_id, dim_balance_date, asset_id, market_value_system_currency, position_type, currency_code, position_values_mv_next.holding_currency_expanded, asset_currency, source_entity_type, position_values_mv_next.position_type_expanded, flag, position_summary_mv_next.source_entity_type ], stream key: [] }
    └── BatchPlanNode { output: [ account_group_id, dim_balance_date, asset_id, market_value_system_currency, position_type, currency_code, position_values_mv_next.holding_currency_expanded, asset_currency, source_entity_type, position_values_mv_next.position_type_expanded, flag, position_summary_mv_next.source_entity_type ], stream key: [] }

Fragment 20191 (Actor 167112,167113)
StreamLocalityProvider { locality_columns: [party_to_account_groups_mv.account_group_id] }
├── output: [ party_to_account_groups_mv.party_id, party_to_account_groups_mv.account_group_id, party_to_account_groups_mv.type, party_to_account_groups_mv.parties.id, party_to_account_groups_mv.null:Varchar, party_to_account_groups_mv.$src ]
├── stream key: [ party_to_account_groups_mv.account_group_id, party_to_account_groups_mv.parties.id, party_to_account_groups_mv.null:Varchar, party_to_account_groups_mv.$src ]
└── MergeExecutor
    ├── output: [ party_to_account_groups_mv.party_id, party_to_account_groups_mv.account_group_id, party_to_account_groups_mv.type, party_to_account_groups_mv.parties.id, party_to_account_groups_mv.null:Varchar, party_to_account_groups_mv.$src ]
    └── stream key: [ party_to_account_groups_mv.parties.id, party_to_account_groups_mv.null:Varchar, party_to_account_groups_mv.$src ]

Fragment 20192 (Actor 167103,167102)
StreamTableScan { table: party_to_account_groups_mv, columns: [party_id, account_group_id, type, parties.id, null:Varchar, $src] }
├── output: [ party_to_account_groups_mv.party_id, party_to_account_groups_mv.account_group_id, party_to_account_groups_mv.type, party_to_account_groups_mv.parties.id, party_to_account_groups_mv.null:Varchar, party_to_account_groups_mv.$src ]
├── stream key: [ party_to_account_groups_mv.parties.id, party_to_account_groups_mv.null:Varchar, party_to_account_groups_mv.$src ]
├── Upstream { output: [ party_id, account_group_id, type, parties.id, null:Varchar, $src ], stream key: [] }
└── BatchPlanNode { output: [ party_id, account_group_id, type, parties.id, null:Varchar, $src ], stream key: [] }