active_identifier_edges_mv · explain

← cluster search objects active_identifier_edges_mv explain

Overview Objects Graph History

materialized view · search.active_identifier_edges_mv profiled over 5s

Job is idle — throughput ~0; structure shown.

Stateful hash join (4 state tables) — consider a temporal join for dimension lookups

Heat tree Operator graph

257 operators

Materialize · search.active_identifier_edges_mv

0% idle 2 actors

Union

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · party_holder_edges_mv

— — 2 actors

StreamScan · party_holder_edges_mv

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · account_to_portfolios_dm.account_id = accounts_dm.account_id

— — 2 actors

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

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

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

— — 2 actors

HashJoin · Inner · account_to_portfolios_dm.portfolio_id = 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

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

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

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · clients_portfolios_dm.client_id = clients_dm.id

— — 2 actors

HashJoin · Inner · clients_portfolios_dm.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

SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli…

— — 2 actors

HashJoin · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli… 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

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

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli…

— — 2 actors

HashJoin · Inner · clients_portfolios_dm.portfolio_id = portfolios_dm.portfoli… 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

SyncLogStore · Inner · clients_portfolios_dm.client_id = clients_dm.id

— — 2 actors

HashJoin · Inner · clients_portfolios_dm.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

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

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_id

— — 2 actors

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

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

SyncLogStore · Inner · accounts_to_clients_dm.client_id = clients_dm.id

— — 2 actors

HashJoin · Inner · accounts_to_clients_dm.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

Project · accounts_to_clients_dm

— — 2 actors

Filter · accounts_to_clients_dm

0% idle 2 actors

StreamScan · accounts_to_clients_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

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

— — 2 actors

HashJoin · Inner · account_to_portfolios_dm.portfolio_id = 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

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 · account_to_portfolios_dm.account_id = accounts_dm.account_id

— — 2 actors

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

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

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

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · accounts_to_clients_dm.client_id = clients_dm.id

— — 2 actors

HashJoin · Inner · accounts_to_clients_dm.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

SyncLogStore · Inner · accounts_to_clients_dm.account_id = accounts_dm.account_id

— — 2 actors

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

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

Project · accounts_to_clients_dm

— — 2 actors

Filter · accounts_to_clients_dm

0% idle 2 actors

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

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

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

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 52334 (Actor 737669,737668)
StreamMaterialize { columns: [target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, null:Int32(hidden), null:Int32#1(hidden), accounts_dm.account_id(hidden), null:Varchar(hidden), null:Varchar#1(hidden), null:Varchar#2(hidden), null:Date(hidden), $src(hidden)], stream_key: [accounts_dm.account_id, null:Varchar, null:Date, null:Varchar#1, null:Varchar#2, null:Int32, null:Int32#1, $src], pk_columns: [accounts_dm.account_id, null:Varchar, null:Date, null:Varchar#1, null:Varchar#2, null:Int32, null:Int32#1, $src], pk_conflict: NoCheck }
├── output: [ accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, null:Int32, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, $src ]
├── stream key: [ accounts_dm.account_id, null:Varchar, null:Date, null:Varchar, null:Varchar, null:Int32, null:Int32, $src ]
└── StreamUnion { all: true } { output: [ accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, null:Int32, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, $src ], stream key: [ accounts_dm.account_id, null:Varchar, null:Date, null:Varchar, null:Varchar, null:Int32, null:Int32, $src ] }
    ├── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, null:Int32, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, 0:Int32 ], stream key: [ accounts_dm.account_id ] }
    ├── MergeExecutor { output: [ clients_dm.id, 'client':Varchar, clients_dm.id, 'client':Varchar, null:Int32, null:Int32, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, 1:Int32 ], stream key: [ clients_dm.id ] }
    ├── MergeExecutor { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, portfolios_dm.portfolio_id, null:Varchar, null:Varchar, null:Varchar, null:Date, 2:Int32 ], stream key: [ portfolios_dm.portfolio_id ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, 'account':Varchar, accounts_to_clients_dm.client_id, 'client':Varchar, null:Int32, null:Int32, accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, 3:Int32 ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ account_to_portfolios_dm.account_id, 'account':Varchar, account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, 4:Int32 ], stream key: [ account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.client_id, 'client':Varchar, accounts_to_clients_dm.account_id, 'account':Varchar, null:Int32, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, 5:Int32 ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_portfolios_dm.client_id, 'client':Varchar, clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, 6:Int32 ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, clients_portfolios_dm.client_id, 'client':Varchar, null:Int32, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, 7:Int32 ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, account_to_portfolios_dm.account_id, 'account':Varchar, null:Int32, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, 8:Int32 ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
    └── MergeExecutor
        ├── output: [ party_holder_edges_mv.party_id, 'party':Varchar, party_holder_edges_mv.entity_id, party_holder_edges_mv.entity_type, party_holder_edges_mv.$src, party_holder_edges_mv.$src#1, party_holder_edges_mv.accounts_dm.account_id, party_holder_edges_mv.customer_relationships.id, party_holder_edges_mv.active_parties_mv.id, party_holder_edges_mv.party_involvements_dm.id, null:Date, 9:Int32 ]
        └── stream key: [ party_holder_edges_mv.accounts_dm.account_id, party_holder_edges_mv.customer_relationships.id, party_holder_edges_mv.active_parties_mv.id, party_holder_edges_mv.party_involvements_dm.id, party_holder_edges_mv.$src, party_holder_edges_mv.$src#1 ]

Fragment 52335 (Actor 737763,737764)
StreamProject { exprs: [accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, null:Int32, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, 0:Int32] } { output: [ accounts_dm.account_id, 'account':Varchar, accounts_dm.account_id, 'account':Varchar, null:Int32, null:Int32, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, 0:Int32 ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 52336 (Actor 737765,737766)
StreamProject { exprs: [clients_dm.id, 'client':Varchar, clients_dm.id, 'client':Varchar, null:Int32, null:Int32, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, 1:Int32] } { output: [ clients_dm.id, 'client':Varchar, clients_dm.id, 'client':Varchar, null:Int32, null:Int32, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, 1:Int32 ], 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 52337 (Actor 737679,737678)
StreamProject { exprs: [portfolios_dm.portfolio_id, 'portfolio':Varchar, portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, portfolios_dm.portfolio_id, null:Varchar, null:Varchar, null:Varchar, null:Date, 2:Int32] } { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, portfolios_dm.portfolio_id, null:Varchar, null:Varchar, null:Varchar, null:Date, 2:Int32 ], 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 52338 (Actor 737670,737671)
StreamProject { exprs: [accounts_to_clients_dm.account_id, 'account':Varchar, accounts_to_clients_dm.client_id, 'client':Varchar, null:Int32, null:Int32, accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, 3:Int32] }
├── output: [ accounts_to_clients_dm.account_id, 'account':Varchar, accounts_to_clients_dm.client_id, 'client':Varchar, null:Int32, null:Int32, accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, 3:Int32 ]
├── stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ]
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 52339 (Actor 737672,737673)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.client_id = clients_dm.id } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 52340 (Actor 737674,737675)
StreamLocalityProvider { locality_columns: [accounts_to_clients_dm.client_id] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 52341 (Actor 737676,737677)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.account_id = accounts_dm.account_id } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 52342 (Actor 737681,737680)
StreamLocalityProvider { locality_columns: [accounts_to_clients_dm.account_id] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 52343 (Actor 737767,737768)
StreamProject { exprs: [accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamFilter { predicate: IsNull(accounts_to_clients_dm.disabled_at) AND IsNull(accounts_to_clients_dm.effective_end_date) } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    └── StreamTableScan { table: accounts_to_clients_dm, columns: [account_id, client_id, effective_start_date, disabled_at, effective_end_date] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
        ├── Upstream { output: [ account_id, client_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, client_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

Fragment 52344 (Actor 737769,737770)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 52345 (Actor 737772,737771)
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 52346 (Actor 737686,737687)
StreamProject { exprs: [account_to_portfolios_dm.account_id, 'account':Varchar, account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, 4:Int32] }
├── output: [ account_to_portfolios_dm.account_id, 'account':Varchar, account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, 4:Int32 ]
├── 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, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 52347 (Actor 737685,737684)
StreamSyncLogStore { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], 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, accounts_dm.account_id ], stream key: [ account_to_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 ] }

Fragment 52348 (Actor 737688,737689)
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, accounts_dm.account_id ], 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, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 52349 (Actor 737692,737693)
StreamSyncLogStore { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.account_id = accounts_dm.account_id } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_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, 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 ] }
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 52350 (Actor 737694,737695)
StreamLocalityProvider { locality_columns: [account_to_portfolios_dm.account_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.account_id, account_to_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, 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 52351 (Actor 737774,737773)
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: 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: [] }

Fragment 52352 (Actor 737776,737775)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 52353 (Actor 737777,737778)
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 52354 (Actor 737700,737701)
StreamProject { exprs: [accounts_to_clients_dm.client_id, 'client':Varchar, accounts_to_clients_dm.account_id, 'account':Varchar, null:Int32, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, 5:Int32] }
├── output: [ accounts_to_clients_dm.client_id, 'client':Varchar, accounts_to_clients_dm.account_id, 'account':Varchar, null:Int32, null:Int32, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, 5:Int32 ]
├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ]
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 52355 (Actor 737703,737702)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.account_id = accounts_dm.account_id } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_dm.account_id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 52356 (Actor 737705,737704)
StreamLocalityProvider { locality_columns: [accounts_to_clients_dm.account_id] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 52357 (Actor 737707,737706)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.client_id = clients_dm.id } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_dm.id ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 52358 (Actor 737709,737708)
StreamLocalityProvider { locality_columns: [accounts_to_clients_dm.client_id] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 52359 (Actor 737779,737780)
StreamProject { exprs: [accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
└── StreamFilter { predicate: IsNull(accounts_to_clients_dm.disabled_at) AND IsNull(accounts_to_clients_dm.effective_end_date) } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    └── StreamTableScan { table: accounts_to_clients_dm, columns: [account_id, client_id, effective_start_date, disabled_at, effective_end_date] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
        ├── Upstream { output: [ account_id, client_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, client_id, effective_start_date, disabled_at, effective_end_date ], stream key: [] }

Fragment 52360 (Actor 737782,737781)
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 52361 (Actor 737783,737784)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 52362 (Actor 737715,737714)
StreamProject { exprs: [clients_portfolios_dm.client_id, 'client':Varchar, clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, 6:Int32] }
├── output: [ clients_portfolios_dm.client_id, 'client':Varchar, clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Int32, null:Int32, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, 6:Int32 ]
├── stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ]
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }

Fragment 52363 (Actor 737713,737712)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 52364 (Actor 737716,737717)
StreamLocalityProvider { locality_columns: [clients_portfolios_dm.portfolio_id] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 52365 (Actor 737719,737718)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.client_id = clients_dm.id } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── 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 ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 52366 (Actor 737723,737724)
StreamLocalityProvider { locality_columns: [clients_portfolios_dm.client_id] } { 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 ] }
└── 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 52367 (Actor 737786,737785)
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 ] }
└── 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: [] }

Fragment 52368 (Actor 737787,737788)
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 52369 (Actor 737790,737789)
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 52370 (Actor 737734,737733)
StreamProject { exprs: [clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, clients_portfolios_dm.client_id, 'client':Varchar, null:Int32, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, 7:Int32] }
├── output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, clients_portfolios_dm.client_id, 'client':Varchar, null:Int32, null:Int32, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, 7:Int32 ]
├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ]
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 52371 (Actor 737735,737736)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.client_id = clients_dm.id } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_dm.id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 52372 (Actor 737738,737737)
StreamLocalityProvider { locality_columns: [clients_portfolios_dm.client_id] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }

Fragment 52373 (Actor 737739,737740)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 52374 (Actor 737741,737742)
StreamLocalityProvider { locality_columns: [clients_portfolios_dm.portfolio_id] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
└── 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 52375 (Actor 737792,737791)
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 ] }
└── 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: [] }

Fragment 52376 (Actor 737793,737794)
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 52377 (Actor 737795,737796)
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 52378 (Actor 737744,737743)
StreamProject { exprs: [account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, account_to_portfolios_dm.account_id, 'account':Varchar, null:Int32, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, 8:Int32] }
├── output: [ account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, account_to_portfolios_dm.account_id, 'account':Varchar, null:Int32, null:Int32, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, 8:Int32 ]
├── stream key: [ account_to_portfolios_dm.account_id, account_to_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, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 52379 (Actor 737746,737745)
StreamSyncLogStore { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.account_id = accounts_dm.account_id } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, accounts_dm.account_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_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, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }

Fragment 52380 (Actor 737748,737747)
StreamLocalityProvider { locality_columns: [account_to_portfolios_dm.account_id] } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.account_id, account_to_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, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 52381 (Actor 737749,737750)
StreamSyncLogStore { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.account_id, account_to_portfolios_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: account_to_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, portfolios_dm.portfolio_id ], 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.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 ] }

Fragment 52382 (Actor 737752,737751)
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 52383 (Actor 737797,737798)
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: 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: [] }

Fragment 52384 (Actor 737799,737800)
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 52385 (Actor 737755,737756)
StreamProject { exprs: [accounts_dm.account_id] } { output: [ accounts_dm.account_id ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, disabled_at ], stream key: [] }

Fragment 52386 (Actor 737758,737757)
StreamProject { exprs: [party_holder_edges_mv.party_id, 'party':Varchar, party_holder_edges_mv.entity_id, party_holder_edges_mv.entity_type, party_holder_edges_mv.$src, party_holder_edges_mv.$src#1, party_holder_edges_mv.accounts_dm.account_id, party_holder_edges_mv.customer_relationships.id, party_holder_edges_mv.active_parties_mv.id, party_holder_edges_mv.party_involvements_dm.id, null:Date, 9:Int32] }
├── output: [ party_holder_edges_mv.party_id, 'party':Varchar, party_holder_edges_mv.entity_id, party_holder_edges_mv.entity_type, party_holder_edges_mv.$src, party_holder_edges_mv.$src#1, party_holder_edges_mv.accounts_dm.account_id, party_holder_edges_mv.customer_relationships.id, party_holder_edges_mv.active_parties_mv.id, party_holder_edges_mv.party_involvements_dm.id, null:Date, 9:Int32 ]
├── stream key: [ party_holder_edges_mv.accounts_dm.account_id, party_holder_edges_mv.customer_relationships.id, party_holder_edges_mv.active_parties_mv.id, party_holder_edges_mv.party_involvements_dm.id, party_holder_edges_mv.$src, party_holder_edges_mv.$src#1 ]
└── StreamTableScan { table: party_holder_edges_mv, columns: [party_id, entity_id, entity_type, accounts_dm.account_id, customer_relationships.id, active_parties_mv.id, party_involvements_dm.id, $src, $src#1] }
    ├── output: [ party_holder_edges_mv.party_id, party_holder_edges_mv.entity_id, party_holder_edges_mv.entity_type, party_holder_edges_mv.accounts_dm.account_id, party_holder_edges_mv.customer_relationships.id, party_holder_edges_mv.active_parties_mv.id, party_holder_edges_mv.party_involvements_dm.id, party_holder_edges_mv.$src, party_holder_edges_mv.$src#1 ]
    ├── stream key: [ party_holder_edges_mv.accounts_dm.account_id, party_holder_edges_mv.customer_relationships.id, party_holder_edges_mv.active_parties_mv.id, party_holder_edges_mv.party_involvements_dm.id, party_holder_edges_mv.$src, party_holder_edges_mv.$src#1 ]
    ├── Upstream { output: [ party_id, entity_id, entity_type, accounts_dm.account_id, customer_relationships.id, active_parties_mv.id, party_involvements_dm.id, $src, $src#1 ], stream key: [] }
    └── BatchPlanNode { output: [ party_id, entity_id, entity_type, accounts_dm.account_id, customer_relationships.id, active_parties_mv.id, party_involvements_dm.id, $src, $src#1 ], stream key: [] }

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