portfolio_last_transaction_mv · explain

← cluster opportunity objects portfolio_last_transaction_mv explain

materialized view · opportunity.portfolio_last_transaction_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

56 operators

Materialize · opportunity.portfolio_last_transaction_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 · transactions_dm.account_id = account_to_portfolios_dm.accou…

— — 2 actors

HashJoin · Inner · transactions_dm.account_id = account_to_portfolios_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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · portfolios_dm.portfolio_id = account_to_portfolios_dm.portf…

— — 2 actors

HashJoin · Inner · portfolios_dm.portfolio_id = account_to_portfolios_dm.portf… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · account_to_portfolios_dm

— — 2 actors

Filter · account_to_portfolios_dm

0% idle 2 actors

StreamScan · account_to_portfolios_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

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

SyncLogStore · Inner · transactions_dm.account_id = open_accounts_mv.account_id

— — 2 actors

HashJoin · Inner · transactions_dm.account_id = open_accounts_mv.account_id Stateful hash join (4 state tables) — consider a temporal join for dimension lookups

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

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

Project · transactions_dm

— — 2 actors

Filter · transactions_dm

0% idle 2 actors

StreamScan · transactions_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 12762 (Actor 157557,157556)
StreamMaterialize { columns: [portfolio_id, last_transaction_date, activity_name], stream_key: [portfolio_id], pk_columns: [portfolio_id], pk_conflict: NoCheck }
├── output: [ account_to_portfolios_dm.portfolio_id, max(max(transactions_dm.transaction_valuation_date)), 'GET_PORTFOLIO_DAYS_SINCE_LAST_TRANSACTION':Varchar ]
├── stream key: [ account_to_portfolios_dm.portfolio_id ]
└── StreamProject { exprs: [account_to_portfolios_dm.portfolio_id, max(max(transactions_dm.transaction_valuation_date)), 'GET_PORTFOLIO_DAYS_SINCE_LAST_TRANSACTION':Varchar] }
    ├── output: [ account_to_portfolios_dm.portfolio_id, max(max(transactions_dm.transaction_valuation_date)), 'GET_PORTFOLIO_DAYS_SINCE_LAST_TRANSACTION':Varchar ]
    ├── stream key: [ account_to_portfolios_dm.portfolio_id ]
    └── StreamHashAgg { group_key: [account_to_portfolios_dm.portfolio_id], aggs: [max(max(transactions_dm.transaction_valuation_date)), count] }
        ├── output: [ account_to_portfolios_dm.portfolio_id, max(max(transactions_dm.transaction_valuation_date)), count ]
        ├── stream key: [ account_to_portfolios_dm.portfolio_id ]
        └── StreamLocalityProvider { locality_columns: [account_to_portfolios_dm.portfolio_id] }
            ├── output:
            │   ┌── account_to_portfolios_dm.portfolio_id
            │   ├── max(transactions_dm.transaction_valuation_date)
            │   ├── transactions_dm.account_id
            │   ├── open_accounts_mv.account_id
            │   ├── account_to_portfolios_dm.account_id
            │   ├── portfolios_dm.portfolio_id
            │   └── account_to_portfolios_dm.effective_start_date
            ├── stream key:
            │   ┌── account_to_portfolios_dm.portfolio_id
            │   ├── transactions_dm.account_id
            │   ├── open_accounts_mv.account_id
            │   ├── portfolios_dm.portfolio_id
            │   └── account_to_portfolios_dm.effective_start_date
            └── MergeExecutor
                ├── output:
                │   ┌── account_to_portfolios_dm.portfolio_id
                │   ├── max(transactions_dm.transaction_valuation_date)
                │   ├── transactions_dm.account_id
                │   ├── open_accounts_mv.account_id
                │   ├── account_to_portfolios_dm.account_id
                │   ├── portfolios_dm.portfolio_id
                │   └── account_to_portfolios_dm.effective_start_date
                └── stream key: [ transactions_dm.account_id, open_accounts_mv.account_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]

Fragment 12763 (Actor 157748,157749)
StreamSyncLogStore
├── output:
│   ┌── account_to_portfolios_dm.portfolio_id
│   ├── max(transactions_dm.transaction_valuation_date)
│   ├── transactions_dm.account_id
│   ├── open_accounts_mv.account_id
│   ├── account_to_portfolios_dm.account_id
│   ├── portfolios_dm.portfolio_id
│   └── account_to_portfolios_dm.effective_start_date
├── stream key: [ transactions_dm.account_id, open_accounts_mv.account_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: transactions_dm.account_id = account_to_portfolios_dm.account_id AND open_accounts_mv.account_id = account_to_portfolios_dm.account_id }
    ├── output:
    │   ┌── account_to_portfolios_dm.portfolio_id
    │   ├── max(transactions_dm.transaction_valuation_date)
    │   ├── transactions_dm.account_id
    │   ├── open_accounts_mv.account_id
    │   ├── account_to_portfolios_dm.account_id
    │   ├── portfolios_dm.portfolio_id
    │   └── account_to_portfolios_dm.effective_start_date
    ├── stream key: [ transactions_dm.account_id, open_accounts_mv.account_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor
    │   ├── output: [ transactions_dm.account_id, max(transactions_dm.transaction_valuation_date), open_accounts_mv.account_id ]
    │   └── stream key: [ transactions_dm.account_id, open_accounts_mv.account_id ]
    └── MergeExecutor
        ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
        └── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.account_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]

Fragment 12764 (Actor 157751,157750)
StreamLocalityProvider { locality_columns: [transactions_dm.account_id, open_accounts_mv.account_id] }
├── output: [ transactions_dm.account_id, max(transactions_dm.transaction_valuation_date), open_accounts_mv.account_id ]
├── stream key: [ transactions_dm.account_id, open_accounts_mv.account_id ]
└── MergeExecutor
    ├── output: [ transactions_dm.account_id, max(transactions_dm.transaction_valuation_date), open_accounts_mv.account_id ]
    └── stream key: [ transactions_dm.account_id ]

Fragment 12765 (Actor 157753,157752)
StreamSyncLogStore
├── output: [ transactions_dm.account_id, max(transactions_dm.transaction_valuation_date), open_accounts_mv.account_id ]
├── stream key: [ transactions_dm.account_id ]
└── StreamHashJoin { type: Inner, predicate: transactions_dm.account_id = open_accounts_mv.account_id }
    ├── output: [ transactions_dm.account_id, max(transactions_dm.transaction_valuation_date), open_accounts_mv.account_id ]
    ├── stream key: [ transactions_dm.account_id ]
    ├── MergeExecutor { output: [ transactions_dm.account_id, max(transactions_dm.transaction_valuation_date) ], stream key: [ transactions_dm.account_id ] }
    └── MergeExecutor { output: [ open_accounts_mv.account_id ], stream key: [ open_accounts_mv.account_id ] }

Fragment 12766 (Actor 157754,157755)
StreamProject { exprs: [transactions_dm.account_id, max(transactions_dm.transaction_valuation_date)] }
├── output: [ transactions_dm.account_id, max(transactions_dm.transaction_valuation_date) ]
├── stream key: [ transactions_dm.account_id ]
└── StreamHashAgg { group_key: [transactions_dm.account_id], aggs: [max(transactions_dm.transaction_valuation_date), count] }
    ├── output: [ transactions_dm.account_id, max(transactions_dm.transaction_valuation_date), count ]
    ├── stream key: [ transactions_dm.account_id ]
    └── StreamLocalityProvider { locality_columns: [transactions_dm.account_id] }
        ├── output: [ transactions_dm.account_id, transactions_dm.transaction_valuation_date, transactions_dm.transaction_id ]
        ├── stream key: [ transactions_dm.account_id, transactions_dm.transaction_id ]
        └── MergeExecutor
            ├── output: [ transactions_dm.account_id, transactions_dm.transaction_valuation_date, transactions_dm.transaction_id ]
            └── stream key: [ transactions_dm.transaction_id ]

Fragment 12767 (Actor 157757,157756)
StreamProject { exprs: [transactions_dm.account_id, transactions_dm.transaction_valuation_date, transactions_dm.transaction_id] }
├── output: [ transactions_dm.account_id, transactions_dm.transaction_valuation_date, transactions_dm.transaction_id ]
├── stream key: [ transactions_dm.transaction_id ]
└── StreamFilter { predicate: IsNull(transactions_dm.disabled_at) }
    ├── output: [ transactions_dm.account_id, transactions_dm.transaction_valuation_date, transactions_dm.transaction_id, transactions_dm.disabled_at ]
    ├── stream key: [ transactions_dm.transaction_id ]
    └── StreamTableScan { table: transactions_dm, columns: [account_id, transaction_valuation_date, transaction_id, disabled_at] }
        ├── output: [ transactions_dm.account_id, transactions_dm.transaction_valuation_date, transactions_dm.transaction_id, transactions_dm.disabled_at ]
        ├── stream key: [ transactions_dm.transaction_id ]
        ├── Upstream { output: [ account_id, transaction_valuation_date, transaction_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, transaction_valuation_date, transaction_id, disabled_at ], stream key: [] }

Fragment 12768 (Actor 157713,157712)
StreamTableScan { table: open_accounts_mv, columns: [account_id] } { output: [ open_accounts_mv.account_id ], stream key: [ open_accounts_mv.account_id ] }
├── Upstream { output: [ account_id ], stream key: [] }
└── BatchPlanNode { output: [ account_id ], stream key: [] }

Fragment 12769 (Actor 157758,157759)
StreamLocalityProvider { locality_columns: [account_to_portfolios_dm.account_id, account_to_portfolios_dm.account_id] }
├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.account_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── MergeExecutor
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    └── stream key: [ portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]

Fragment 12770 (Actor 157761,157760)
StreamSyncLogStore
├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
└── StreamHashJoin { type: Inner, predicate: portfolios_dm.portfolio_id = account_to_portfolios_dm.portfolio_id }
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    ├── stream key: [ portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
    ├── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }
    └── MergeExecutor
        ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
        └── stream key: [ account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]

Fragment 12771 (Actor 157765,157764)
StreamProject { exprs: [portfolios_dm.portfolio_id] } { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) AND Not(IsTrue(portfolios_dm.m_is_stub)) }
    ├── output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at, portfolios_dm.m_is_stub ]
    ├── stream key: [ portfolios_dm.portfolio_id ]
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, disabled_at, m_is_stub] }
        ├── output: [ portfolios_dm.portfolio_id, portfolios_dm.disabled_at, portfolios_dm.m_is_stub ]
        ├── stream key: [ portfolios_dm.portfolio_id ]
        ├── Upstream { output: [ portfolio_id, disabled_at, m_is_stub ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, disabled_at, m_is_stub ], stream key: [] }

Fragment 12772 (Actor 157763,157762)
StreamLocalityProvider { locality_columns: [account_to_portfolios_dm.portfolio_id] }
├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ]
└── MergeExecutor
    ├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    └── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]

Fragment 12773 (Actor 157715,157714)
StreamProject { exprs: [account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date] }
├── output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
└── StreamFilter { predicate: IsNotNull(account_to_portfolios_dm.account_id) AND IsNull(account_to_portfolios_dm.disabled_at) AND IsNull(account_to_portfolios_dm.effective_end_date) }
    ├── output:
    │   ┌── account_to_portfolios_dm.account_id
    │   ├── account_to_portfolios_dm.portfolio_id
    │   ├── account_to_portfolios_dm.effective_start_date
    │   ├── account_to_portfolios_dm.disabled_at
    │   └── account_to_portfolios_dm.effective_end_date
    ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
    └── StreamTableScan { table: account_to_portfolios_dm, columns: [account_id, portfolio_id, effective_start_date, disabled_at, effective_end_date] }
        ├── output:
        │   ┌── account_to_portfolios_dm.account_id
        │   ├── account_to_portfolios_dm.portfolio_id
        │   ├── account_to_portfolios_dm.effective_start_date
        │   ├── account_to_portfolios_dm.disabled_at
        │   └── account_to_portfolios_dm.effective_end_date
        ├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ]
        ├── Upstream { output: [ account_id, portfolio_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, portfolio_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

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