advisor_kpi_counts_mv · explain

← cluster alinma_bff objects advisor_kpi_counts_mv explain

materialized view · alinma_bff.advisor_kpi_counts_mv profiled over 5s

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

Heat tree Operator graph

121 operators

Materialize · alinma_bff.advisor_kpi_counts_mv

0% idle 2 actors

Project

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · LeftOuter · advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_acco…

— — 2 actors

HashJoin · LeftOuter · advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_acco… 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

— — 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 · advisor_kpi_user_accounts_mv.account_id = accounts_dm.accou…

— — 2 actors

HashJoin · Inner · advisor_kpi_user_accounts_mv.account_id = accounts_dm.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

Project · accounts_dm

— — 2 actors

Filter · accounts_dm

0% idle 2 actors

StreamScan · accounts_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

NoOp

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · advisor_kpi_user_accounts_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

SyncLogStore · LeftOuter · advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_port…

— — 2 actors

HashJoin · LeftOuter · advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_port… 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

— — 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 · advisor_kpi_user_portfolios_mv.portfolio_id = portfolios_dm…

— — 2 actors

HashJoin · Inner · advisor_kpi_user_portfolios_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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

NoOp

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · advisor_kpi_user_portfolios_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

SyncLogStore · LeftOuter · advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_clie…

— — 2 actors

HashJoin · LeftOuter · advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_clie… 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

— — 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 · advisor_kpi_user_clients_mv.client_id = clients_dm.id

— — 2 actors

HashJoin · Inner · advisor_kpi_user_clients_mv.client_id = clients_dm.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_dm

— — 2 actors

Filter · clients_dm

0% idle 2 actors

StreamScan · clients_dm

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

NoOp

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · advisor_kpi_user_clients_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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Union

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 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 54951 (Actor 740730,740729)
StreamMaterialize { columns: [user_id, clients_count, portfolios_count, accounts_count], stream_key: [user_id], pk_columns: [user_id], pk_conflict: NoCheck }
├── output: [ advisor_kpi_user_clients_mv.user_id, $expr4, $expr5, $expr6 ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
└── StreamProject { exprs: [advisor_kpi_user_clients_mv.user_id, Coalesce($expr1, 0:Int32) as $expr4, Coalesce($expr2, 0:Int32) as $expr5, Coalesce($expr3, 0:Int32) as $expr6] }
    ├── output: [ advisor_kpi_user_clients_mv.user_id, $expr4, $expr5, $expr6 ]
    ├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
    └── MergeExecutor
        ├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, $expr2, $expr3, advisor_kpi_user_accounts_mv.user_id ]
        └── stream key: [ advisor_kpi_user_clients_mv.user_id ]

Fragment 54952 (Actor 740727,740728)
StreamSyncLogStore
├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, $expr2, $expr3, advisor_kpi_user_accounts_mv.user_id ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
└── StreamHashJoin { type: LeftOuter, predicate: advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_accounts_mv.user_id }
    ├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, $expr2, $expr3, advisor_kpi_user_accounts_mv.user_id ]
    ├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
    ├── MergeExecutor
    │   ├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, $expr2, advisor_kpi_user_portfolios_mv.user_id ]
    │   └── stream key: [ advisor_kpi_user_clients_mv.user_id ]
    └── MergeExecutor { output: [ advisor_kpi_user_accounts_mv.user_id, $expr3 ], stream key: [ advisor_kpi_user_accounts_mv.user_id ] }

Fragment 54953 (Actor 740733,740734)
StreamLocalityProvider { locality_columns: [advisor_kpi_user_clients_mv.user_id] }
├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, $expr2, advisor_kpi_user_portfolios_mv.user_id ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, $expr2, advisor_kpi_user_portfolios_mv.user_id ]
    └── stream key: [ advisor_kpi_user_clients_mv.user_id ]

Fragment 54954 (Actor 740731,740732)
StreamSyncLogStore
├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, $expr2, advisor_kpi_user_portfolios_mv.user_id ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
└── StreamHashJoin { type: LeftOuter, predicate: advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_portfolios_mv.user_id }
    ├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, $expr2, advisor_kpi_user_portfolios_mv.user_id ]
    ├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
    ├── MergeExecutor { output: [ advisor_kpi_user_clients_mv.user_id, $expr1, advisor_kpi_user_clients_mv.user_id ], stream key: [ advisor_kpi_user_clients_mv.user_id ] }
    └── MergeExecutor { output: [ advisor_kpi_user_portfolios_mv.user_id, $expr2 ], stream key: [ advisor_kpi_user_portfolios_mv.user_id ] }

Fragment 54955 (Actor 740737,740738)
StreamLocalityProvider { locality_columns: [advisor_kpi_user_clients_mv.user_id] }
├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, advisor_kpi_user_clients_mv.user_id ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
└── MergeExecutor { output: [ advisor_kpi_user_clients_mv.user_id, $expr1, advisor_kpi_user_clients_mv.user_id ], stream key: [ advisor_kpi_user_clients_mv.user_id ] }

Fragment 54956 (Actor 740736,740735)
StreamSyncLogStore { output: [ advisor_kpi_user_clients_mv.user_id, $expr1, advisor_kpi_user_clients_mv.user_id ], stream key: [ advisor_kpi_user_clients_mv.user_id ] }
└── StreamHashJoin { type: LeftOuter, predicate: advisor_kpi_user_clients_mv.user_id = advisor_kpi_user_clients_mv.user_id }
    ├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1, advisor_kpi_user_clients_mv.user_id ]
    ├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
    ├── MergeExecutor { output: [ advisor_kpi_user_clients_mv.user_id ], stream key: [ advisor_kpi_user_clients_mv.user_id ] }
    └── MergeExecutor { output: [ advisor_kpi_user_clients_mv.user_id, $expr1 ], stream key: [ advisor_kpi_user_clients_mv.user_id ] }

Fragment 54957 (Actor 740740,740739)
StreamProject { exprs: [advisor_kpi_user_clients_mv.user_id] } { output: [ advisor_kpi_user_clients_mv.user_id ], stream key: [ advisor_kpi_user_clients_mv.user_id ] }
└── StreamHashAgg { group_key: [advisor_kpi_user_clients_mv.user_id], aggs: [count] }
    ├── output: [ advisor_kpi_user_clients_mv.user_id, count ]
    ├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
    └── StreamLocalityProvider { locality_columns: [advisor_kpi_user_clients_mv.user_id] }
        ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, $src ]
        ├── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, $src ]
        └── MergeExecutor
            ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, $src ]
            └── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, $src ]

Fragment 54958 (Actor 740741,740742)
StreamUnion { all: true }
├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, $src ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, $src ]
├── MergeExecutor
│   ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, 0:Int32 ]
│   └── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
├── MergeExecutor
│   ├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id, 1:Int32 ]
│   └── stream key: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id, 2:Int32 ]
    └── stream key: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]

Fragment 54959 (Actor 740776,740775)
StreamProject { exprs: [advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, 0:Int32] }
├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, 0:Int32 ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
    └── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]

Fragment 54960 (Actor 740774,740773)
StreamTableScan { table: advisor_kpi_user_clients_mv, columns: [user_id, client_id] }
├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
├── Upstream { output: [ user_id, client_id ], stream key: [] }
└── BatchPlanNode { output: [ user_id, client_id ], stream key: [] }

Fragment 54961 (Actor 740779,740780)
StreamProject { exprs: [advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id, 1:Int32] }
├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id, 1:Int32 ]
├── stream key: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
    └── stream key: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]

Fragment 54962 (Actor 740782,740781)
StreamTableScan { table: advisor_kpi_user_portfolios_mv, columns: [user_id, portfolio_id] }
├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
├── stream key: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
├── Upstream { output: [ user_id, portfolio_id ], stream key: [] }
└── BatchPlanNode { output: [ user_id, portfolio_id ], stream key: [] }

Fragment 54963 (Actor 740743,740744)
StreamProject { exprs: [advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id, 2:Int32] }
├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id, 2:Int32 ]
├── stream key: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
    └── stream key: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]

Fragment 54964 (Actor 740747,740748)
StreamTableScan { table: advisor_kpi_user_accounts_mv, columns: [user_id, account_id] }
├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
├── stream key: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
├── Upstream { output: [ user_id, account_id ], stream key: [] }
└── BatchPlanNode { output: [ user_id, account_id ], stream key: [] }

Fragment 54965 (Actor 740749,740750)
StreamProject { exprs: [advisor_kpi_user_clients_mv.user_id, count::Int32 as $expr1] }
├── output: [ advisor_kpi_user_clients_mv.user_id, $expr1 ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
└── StreamHashAgg { group_key: [advisor_kpi_user_clients_mv.user_id], aggs: [count] }
    ├── output: [ advisor_kpi_user_clients_mv.user_id, count ]
    ├── stream key: [ advisor_kpi_user_clients_mv.user_id ]
    └── StreamLocalityProvider { locality_columns: [advisor_kpi_user_clients_mv.user_id] }
        ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, clients_dm.id ]
        ├── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
        └── MergeExecutor
            ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, clients_dm.id ]
            └── stream key: [ advisor_kpi_user_clients_mv.client_id, advisor_kpi_user_clients_mv.user_id ]

Fragment 54966 (Actor 740751,740752)
StreamSyncLogStore
├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, clients_dm.id ]
├── stream key: [ advisor_kpi_user_clients_mv.client_id, advisor_kpi_user_clients_mv.user_id ]
└── StreamHashJoin { type: Inner, predicate: advisor_kpi_user_clients_mv.client_id = clients_dm.id }
    ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id, clients_dm.id ]
    ├── stream key: [ advisor_kpi_user_clients_mv.client_id, advisor_kpi_user_clients_mv.user_id ]
    ├── MergeExecutor
    │   ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
    │   └── stream key: [ advisor_kpi_user_clients_mv.client_id, advisor_kpi_user_clients_mv.user_id ]
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 54967 (Actor 740753,740754)
StreamLocalityProvider { locality_columns: [advisor_kpi_user_clients_mv.client_id] }
├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
├── stream key: [ advisor_kpi_user_clients_mv.client_id, advisor_kpi_user_clients_mv.user_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
    └── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]

Fragment 54968 (Actor 740778,740777)
StreamNoOp
├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
├── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]
    └── stream key: [ advisor_kpi_user_clients_mv.user_id, advisor_kpi_user_clients_mv.client_id ]

Fragment 54969 (Actor 740786,740785)
StreamProject { exprs: [clients_dm.id] } { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, closing_date] } { output: [ clients_dm.id, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, closing_date ], stream key: [] }

Fragment 54970 (Actor 740755,740756)
StreamProject { exprs: [advisor_kpi_user_portfolios_mv.user_id, count::Int32 as $expr2] }
├── output: [ advisor_kpi_user_portfolios_mv.user_id, $expr2 ]
├── stream key: [ advisor_kpi_user_portfolios_mv.user_id ]
└── StreamHashAgg { group_key: [advisor_kpi_user_portfolios_mv.user_id], aggs: [count] }
    ├── output: [ advisor_kpi_user_portfolios_mv.user_id, count ]
    ├── stream key: [ advisor_kpi_user_portfolios_mv.user_id ]
    └── StreamLocalityProvider { locality_columns: [advisor_kpi_user_portfolios_mv.user_id] }
        ├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id, portfolios_dm.portfolio_id ]
        ├── stream key: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
        └── MergeExecutor
            ├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id, portfolios_dm.portfolio_id ]
            └── stream key: [ advisor_kpi_user_portfolios_mv.portfolio_id, advisor_kpi_user_portfolios_mv.user_id ]

Fragment 54971 (Actor 740758,740757)
StreamSyncLogStore
├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id, portfolios_dm.portfolio_id ]
├── stream key: [ advisor_kpi_user_portfolios_mv.portfolio_id, advisor_kpi_user_portfolios_mv.user_id ]
└── StreamHashJoin { type: Inner, predicate: advisor_kpi_user_portfolios_mv.portfolio_id = portfolios_dm.portfolio_id }
    ├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id, portfolios_dm.portfolio_id ]
    ├── stream key: [ advisor_kpi_user_portfolios_mv.portfolio_id, advisor_kpi_user_portfolios_mv.user_id ]
    ├── MergeExecutor
    │   ├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
    │   └── stream key: [ advisor_kpi_user_portfolios_mv.portfolio_id, advisor_kpi_user_portfolios_mv.user_id ]
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 54972 (Actor 740759,740760)
StreamLocalityProvider { locality_columns: [advisor_kpi_user_portfolios_mv.portfolio_id] }
├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
├── stream key: [ advisor_kpi_user_portfolios_mv.portfolio_id, advisor_kpi_user_portfolios_mv.user_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
    └── stream key: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]

Fragment 54973 (Actor 740784,740783)
StreamNoOp
├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
├── stream key: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]
    └── stream key: [ advisor_kpi_user_portfolios_mv.user_id, advisor_kpi_user_portfolios_mv.portfolio_id ]

Fragment 54974 (Actor 740788,740787)
StreamProject { exprs: [portfolios_dm.portfolio_id] } { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) }
    ├── output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ]
    ├── stream key: [ portfolios_dm.portfolio_id ]
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, disabled_at] }
        ├── output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at ]
        ├── stream key: [ portfolios_dm.portfolio_id ]
        ├── Upstream { output: [ portfolio_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, disabled_at ], stream key: [] }

Fragment 54975 (Actor 740762,740761)
StreamProject { exprs: [advisor_kpi_user_accounts_mv.user_id, count::Int32 as $expr3] }
├── output: [ advisor_kpi_user_accounts_mv.user_id, $expr3 ]
├── stream key: [ advisor_kpi_user_accounts_mv.user_id ]
└── StreamHashAgg { group_key: [advisor_kpi_user_accounts_mv.user_id], aggs: [count] }
    ├── output: [ advisor_kpi_user_accounts_mv.user_id, count ]
    ├── stream key: [ advisor_kpi_user_accounts_mv.user_id ]
    └── StreamLocalityProvider { locality_columns: [advisor_kpi_user_accounts_mv.user_id] }
        ├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id, accounts_dm.account_id ]
        ├── stream key: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
        └── MergeExecutor
            ├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id, accounts_dm.account_id ]
            └── stream key: [ advisor_kpi_user_accounts_mv.account_id, advisor_kpi_user_accounts_mv.user_id ]

Fragment 54976 (Actor 740763,740764)
StreamSyncLogStore
├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id, accounts_dm.account_id ]
├── stream key: [ advisor_kpi_user_accounts_mv.account_id, advisor_kpi_user_accounts_mv.user_id ]
└── StreamHashJoin { type: Inner, predicate: advisor_kpi_user_accounts_mv.account_id = accounts_dm.account_id }
    ├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id, accounts_dm.account_id ]
    ├── stream key: [ advisor_kpi_user_accounts_mv.account_id, advisor_kpi_user_accounts_mv.user_id ]
    ├── MergeExecutor
    │   ├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
    │   └── stream key: [ advisor_kpi_user_accounts_mv.account_id, advisor_kpi_user_accounts_mv.user_id ]
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 54977 (Actor 740765,740766)
StreamLocalityProvider { locality_columns: [advisor_kpi_user_accounts_mv.account_id] }
├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
├── stream key: [ advisor_kpi_user_accounts_mv.account_id, advisor_kpi_user_accounts_mv.user_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
    └── stream key: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]

Fragment 54978 (Actor 740746,740745)
StreamNoOp
├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
├── stream key: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
└── MergeExecutor
    ├── output: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]
    └── stream key: [ advisor_kpi_user_accounts_mv.user_id, advisor_kpi_user_accounts_mv.account_id ]

Fragment 54979 (Actor 740767,740768)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.closing_date) AND IsNull(accounts_dm.disabled_at) }
    ├── output: [ accounts_dm.account_id, accounts_dm.closing_date, accounts_dm.disabled_at ]
    ├── stream key: [ accounts_dm.account_id ]
    └── StreamTableScan { table: accounts_dm, columns: [account_id, closing_date, disabled_at] }
        ├── output: [ accounts_dm.account_id, accounts_dm.closing_date, accounts_dm.disabled_at ]
        ├── stream key: [ accounts_dm.account_id ]
        ├── Upstream { output: [ account_id, closing_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, closing_date, disabled_at ], stream key: [] }

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