Job is idle — throughput ~0; structure shown.
Fragment 62476 (Actor 742477,742478)
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 62477 (Actor 742484,742483)
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 62478 (Actor 742512,742513)
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 62479 (Actor 742514,742515)
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 62480 (Actor 742528,742529)
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 62481 (Actor 742714,742713)
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 62482 (Actor 742531,742532)
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 62483 (Actor 742540,742541)
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 62484 (Actor 742548,742547)
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 62485 (Actor 742733,742734)
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 62486 (Actor 742549,742550)
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 62487 (Actor 742673,742674)
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: [] }