items_mv · explain — RWM Console

← cluster search objects items_mv explain

materialized view · search.items_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

441 operators

Materialize · search.items_mv

0% idle 2 actors

Project

— — 2 actors

HashAgg Aggregation state — unbounded unless keyed or temporally filtered

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Union

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · party_identifier_edges_mv_next.owner_entity_id = party_refe…

— — 2 actors

HashJoin · Inner · party_identifier_edges_mv_next.owner_entity_id = party_refe… 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 · party_reference_identifier_terms_mv_next

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · party_identifier_edges_mv_next

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 · active_identifier_edges_mv_next.owner_entity_id = olap_refe…

— — 2 actors

HashJoin · Inner · active_identifier_edges_mv_next.owner_entity_id = olap_refe… 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 · olap_reference_identifier_terms_mv

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Filter · active_identifier_edges_mv_next

0% idle 2 actors

StreamScan · active_identifier_edges_mv_next

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 = entity_to_teams_dm.entity…

— — 2 actors

HashJoin · Inner · clients_portfolios_dm.client_id = entity_to_teams_dm.entity… 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 · entity_to_teams_dm

— — 2 actors

Filter · entity_to_teams_dm

0% idle 2 actors

StreamScan · entity_to_teams_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 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 · entity_to_teams_dm.entity_id = portfolios_dm.portfolio_id

— — 2 actors

HashJoin · Inner · entity_to_teams_dm.entity_id = portfolios_dm.portfolio_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 · 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 · entity_to_teams_dm

— — 2 actors

Filter · entity_to_teams_dm

0% idle 2 actors

StreamScan · entity_to_teams_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 · portfolios_dm.status_label_id = labels_dm.label_id

— — 2 actors

HashJoin · Inner · portfolios_dm.status_label_id = labels_dm.label_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 · labels_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 · Inner · portfolios_dm.service_type_id = service_types_dm.service_ty…

— — 2 actors

TemporalJoin · Inner · portfolios_dm.service_type_id = service_types_dm.service_ty…

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

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

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · accounts_to_clients_dm.client_id = entity_to_teams_dm.entit…

— — 2 actors

HashJoin · Inner · accounts_to_clients_dm.client_id = entity_to_teams_dm.entit… 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 · entity_to_teams_dm

— — 2 actors

Filter · entity_to_teams_dm

0% idle 2 actors

StreamScan · entity_to_teams_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Filter · IsNull(accounts_to_clients_dm.disabled_at) AND IsNull(accou…

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore

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

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 · accounts_dm.status_label_id = labels_dm.label_id

— — 2 actors

HashJoin · Inner · accounts_dm.status_label_id = labels_dm.label_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 · labels_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

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · IsNull(accounts_dm.disabled_at) AND Not(IsNull(product_type…

— — 2 actors

Filter · IsNull(accounts_dm.disabled_at) AND Not(IsNull(product_type…

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

TemporalJoin · Inner · accounts_dm.product_type_id = product_types_dm.product_type…

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · product_types_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 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

Project · IsNull(accounts_dm.disabled_at)

— — 2 actors

Filter · IsNull(accounts_dm.disabled_at)

0% idle 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 · entity_to_teams_dm.entity_id = clients_dm.id

— — 2 actors

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

Project · entity_to_teams_dm

— — 2 actors

Filter · entity_to_teams_dm

0% idle 2 actors

StreamScan · entity_to_teams_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_dm.status_label_id = labels_dm.label_id

— — 2 actors

HashJoin · Inner · clients_dm.status_label_id = labels_dm.label_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 · labels_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

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

— — 2 actors

Filter · assets_dm_next

0% idle 2 actors

StreamScan · assets_dm_next

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · assets_dm_next

— — 2 actors

ProjectSet · assets_dm_next

0% idle 2 actors

Project · assets_dm_next

— — 2 actors

Filter · assets_dm_next

0% idle 2 actors

StreamScan · assets_dm_next

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

ProjectSet

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · clients_portfolios_dm.client_id = clients_contacts_dm.clien…

— — 2 actors

HashJoin · Inner · clients_portfolios_dm.client_id = clients_contacts_dm.clien… 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_contacts_dm

— — 2 actors

Filter · clients_contacts_dm

0% idle 2 actors

StreamScan · clients_contacts_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

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

ProjectSet

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

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

ProjectSet

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

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

ProjectSet · portfolios_dm

0% idle 2 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 · party_items_mv_next

— — 2 actors

StreamScan · party_items_mv_next

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

ProjectSet

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

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 · IsNull(accounts_to_clients_dm.disabled_at)

— — 2 actors

ProjectSet · IsNull(accounts_to_clients_dm.disabled_at)

0% idle 2 actors

Project · IsNull(accounts_to_clients_dm.disabled_at)

— — 2 actors

Filter · IsNull(accounts_to_clients_dm.disabled_at)

0% idle 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · clients_contacts_dm

— — 2 actors

ProjectSet · clients_contacts_dm

0% idle 2 actors

Project · clients_contacts_dm

— — 2 actors

Filter · clients_contacts_dm

0% idle 2 actors

StreamScan · clients_contacts_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

ProjectSet

0% idle 2 actors

Project

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

ProjectSet

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · accounts_to_clients_dm.client_id = clients_contacts_dm.clie…

— — 2 actors

HashJoin · Inner · accounts_to_clients_dm.client_id = clients_contacts_dm.clie… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · clients_contacts_dm

— — 2 actors

Filter · clients_contacts_dm

0% idle 2 actors

StreamScan · clients_contacts_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

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

ProjectSet

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

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 · accounts_dm

— — 2 actors

ProjectSet · accounts_dm

0% idle 2 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 25653 (Actor 127853,127852)
StreamMaterialize { columns: [entity_id, entity_type, search_term_en, search_term_ar, filters], stream_key: [entity_id, entity_type], pk_columns: [entity_id, entity_type], pk_conflict: NoCheck }
├── output: [ accounts_dm.account_id, 'account':Varchar, array_agg(distinct Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) filter(Not(IsNull(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))))) AND (Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))) <> '':Varchar) AND (Trim(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar)) AND (null:Varchar <> '':Varchar) AND (Trim(null:Varchar) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar))) ]
├── stream key: [ accounts_dm.account_id, 'account':Varchar ]
└── StreamProject { exprs: [accounts_dm.account_id, 'account':Varchar, array_agg(distinct Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) filter(Not(IsNull(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))))) AND (Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))) <> '':Varchar) AND (Trim(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar)) AND (null:Varchar <> '':Varchar) AND (Trim(null:Varchar) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar)))] }
    ├── output: [ accounts_dm.account_id, 'account':Varchar, array_agg(distinct Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) filter(Not(IsNull(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))))) AND (Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))) <> '':Varchar) AND (Trim(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar)) AND (null:Varchar <> '':Varchar) AND (Trim(null:Varchar) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar))) ]
    ├── stream key: [ accounts_dm.account_id, 'account':Varchar ]
    └── StreamHashAgg { group_key: [accounts_dm.account_id, 'account':Varchar], aggs: [array_agg(distinct Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) filter(Not(IsNull(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))))) AND (Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))) <> '':Varchar) AND (Trim(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar)) AND (null:Varchar <> '':Varchar) AND (Trim(null:Varchar) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar))), count] }
        ├── output: [ accounts_dm.account_id, 'account':Varchar, array_agg(distinct Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) filter(Not(IsNull(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))))) AND (Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))) <> '':Varchar) AND (Trim(Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar)) AND (null:Varchar <> '':Varchar) AND (Trim(null:Varchar) <> '':Varchar)), array_agg(distinct null:Varchar) filter(Not(IsNull(null:Varchar))), count ]
        ├── stream key: [ accounts_dm.account_id, 'account':Varchar ]
        └── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, $src ], stream key: [ accounts_dm.account_id, _rw_projected_row_id, null:Varchar, null:Date, null:Varchar, null:Varchar, null:Int32, null:Int64, null:Int32, null:Int32, null:Date, $src ] }

Fragment 25654 (Actor 127854,127855)
StreamUnion { all: true } { output: [ accounts_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, $src ], stream key: [ accounts_dm.account_id, _rw_projected_row_id, null:Varchar, null:Date, null:Varchar, null:Varchar, null:Int32, null:Int64, null:Int32, null:Int32, null:Date, $src ] }
├── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 0:Int32 ], stream key: [ accounts_dm.account_id, _rw_projected_row_id ] }
├── MergeExecutor
│   ├── output: [ accounts_to_clients_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 1:Int32 ]
│   └── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, _rw_projected_row_id ]
├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 2:Int32 ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ] }
├── MergeExecutor { output: [ clients_dm.id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 3:Int32 ], stream key: [ clients_dm.id, _rw_projected_row_id ] }
├── MergeExecutor { output: [ clients_contacts_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 4:Int32 ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ] }
├── MergeExecutor { output: [ accounts_to_clients_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 5:Int32 ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, _rw_projected_row_id ] }
├── MergeExecutor { output: [ clients_portfolios_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 6:Int32 ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, _rw_projected_row_id ] }
├── MergeExecutor
│   ├── output: [ party_items_mv_next.entity_id, party_items_mv_next.entity_type, party_items_mv_next.val, party_items_mv_next.ar_val, party_items_mv_next.filter_val, party_items_mv_next.active_parties_mv.id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, null:Date, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Int64, party_items_mv_next.null:Int32, party_items_mv_next.null:Int32#1, party_items_mv_next.$src, 7:Int32 ]
│   └── stream key: [ party_items_mv_next.active_parties_mv.id, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Int32, party_items_mv_next.null:Int64, party_items_mv_next.null:Date, party_items_mv_next.null:Int32#1, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.$src ]
├── MergeExecutor { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, portfolios_dm.portfolio_id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 8:Int32 ], stream key: [ portfolios_dm.portfolio_id, _rw_projected_row_id ] }
├── MergeExecutor { output: [ account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 9:Int32 ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, _rw_projected_row_id ] }
├── MergeExecutor
│   ├── output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 10:Int32 ]
│   └── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, _rw_projected_row_id ]
├── MergeExecutor { output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_contacts_dm.type, clients_contacts_dm.value, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 11:Int32 ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ] }
├── MergeExecutor { output: [ assets_dm_next.id, 'asset':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar))), null:Varchar, $expr1, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 12:Int32 ], stream key: [ assets_dm_next.id, _rw_projected_row_id ] }
├── MergeExecutor { output: [ assets_dm_next.id, 'asset':Varchar, null:Varchar, assets_dm_next.name_ar, $expr2, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 13:Int32 ], stream key: [ assets_dm_next.id ] }
├── MergeExecutor { output: [ clients_dm.id, 'client':Varchar, null:Varchar, null:Varchar, $expr3, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 14:Int32 ], stream key: [ clients_dm.id ] }
├── MergeExecutor { output: [ clients_dm.id, 'client':Varchar, null:Varchar, null:Varchar, $expr4, clients_dm.id, clients_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 15:Int32 ], stream key: [ clients_dm.id, clients_dm.status_label_id ] }
├── MergeExecutor { output: [ entity_to_teams_dm.entity_id, 'client':Varchar, null:Varchar, null:Varchar, $expr5, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 16:Int32 ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
├── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr6, accounts_dm.account_id, accounts_dm.product_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 17:Int32 ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }
├── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr7, accounts_dm.account_id, accounts_dm.product_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 18:Int32 ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }
├── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr8, accounts_dm.account_id, accounts_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 19:Int32 ], stream key: [ accounts_dm.account_id, accounts_dm.status_label_id ] }
├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr9, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 20:Int32 ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
├── MergeExecutor { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr10, portfolios_dm.portfolio_id, portfolios_dm.service_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 21:Int32 ], stream key: [ portfolios_dm.portfolio_id, portfolios_dm.service_type_id ] }
├── MergeExecutor { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr11, portfolios_dm.portfolio_id, portfolios_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 22:Int32 ], stream key: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ] }
├── MergeExecutor { output: [ entity_to_teams_dm.entity_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr12, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 23:Int32 ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
├── MergeExecutor { output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr13, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 24:Int32 ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
├── MergeExecutor
│   ├── output: [ active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, olap_reference_identifier_terms_mv.val, olap_reference_identifier_terms_mv.ar_val, null:Varchar, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, active_identifier_edges_mv_next.null:Date, null:Date, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, null:Int32, 25:Int32 ]
│   └── stream key: [ active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type ]
└── MergeExecutor
    ├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, null:Varchar, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.owner_entity_type, null:Varchar, null:Date, null:Date, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, null:Int32, null:Int32, 26:Int32 ]
    └── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.owner_entity_type ]

Fragment 25655 (Actor 127856,127857)
StreamProject { exprs: [accounts_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 0:Int32] } { output: [ accounts_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, accounts_dm.account_id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 0:Int32 ], stream key: [ accounts_dm.account_id, _rw_projected_row_id ] }
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))] } { output: [ _rw_projected_row_id, accounts_dm.account_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))) ], stream key: [ accounts_dm.account_id, _rw_projected_row_id ] }
    └── StreamProject { exprs: [accounts_dm.account_id, accounts_dm.name, accounts_dm.number] } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number ], stream key: [ accounts_dm.account_id ] }
        └── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
            └── StreamTableScan { table: accounts_dm, columns: [account_id, name, number, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
                ├── Upstream { output: [ account_id, name, number, disabled_at ], stream key: [] }
                └── BatchPlanNode { output: [ account_id, name, number, disabled_at ], stream key: [] }

Fragment 25656 (Actor 127861,127860)
StreamProject { exprs: [accounts_to_clients_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 1:Int32] }
├── output: [ accounts_to_clients_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 1:Int32 ]
├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), $5, $6] } { output: [ _rw_projected_row_id, accounts_to_clients_dm.account_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), 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, _rw_projected_row_id ] }
    └── MergeExecutor { output: [ accounts_to_clients_dm.account_id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, 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 ] }

Fragment 25657 (Actor 127859,127858)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, 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 ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.client_id = clients_dm.id } { output: [ accounts_to_clients_dm.account_id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, 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 ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }
    └── MergeExecutor { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file ], stream key: [ clients_dm.id ] }

Fragment 25658 (Actor 127862,127863)
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) } { 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 ], 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] } { 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 ], 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 ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, client_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 25659 (Actor 127865,127864)
StreamProject { exprs: [clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file] } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, display_name, local_display_name, preferred_name, customer_identification_file, closing_date] } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, display_name, local_display_name, preferred_name, customer_identification_file, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, display_name, local_display_name, preferred_name, customer_identification_file, closing_date ], stream key: [] }

Fragment 25660 (Actor 127866,127867)
StreamProject { exprs: [accounts_to_clients_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 2:Int32] }
├── output: [ accounts_to_clients_dm.account_id, 'account':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 2:Int32 ]
├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), $2, $3, $5, $1] } { output: [ _rw_projected_row_id, accounts_to_clients_dm.account_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ] }
    └── MergeExecutor { output: [ accounts_to_clients_dm.account_id, clients_contacts_dm.value, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.client_id, clients_contacts_dm.type ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ] }

Fragment 25661 (Actor 127868,127869)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, clients_contacts_dm.value, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.client_id, clients_contacts_dm.type ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.client_id = clients_contacts_dm.client_id } { output: [ accounts_to_clients_dm.account_id, clients_contacts_dm.value, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.client_id, clients_contacts_dm.type ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ] }
    ├── 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: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }

Fragment 25662 (Actor 127871,127870)
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) } { 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 ], 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] } { 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 ], 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 ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, client_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 25663 (Actor 127872,127873)
StreamProject { exprs: [clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type] } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
└── StreamFilter { predicate: IsNull(clients_contacts_dm.disabled_at) } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type, clients_contacts_dm.disabled_at ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
    └── StreamTableScan { table: clients_contacts_dm, columns: [client_id, value, type, disabled_at] } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type, clients_contacts_dm.disabled_at ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
        ├── Upstream { output: [ client_id, value, type, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ client_id, value, type, disabled_at ], stream key: [] }

Fragment 25664 (Actor 127877,127876)
StreamProject { exprs: [clients_dm.id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 3:Int32] }
├── output: [ clients_dm.id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 3:Int32 ]
├── stream key: [ clients_dm.id, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar)))] } { output: [ _rw_projected_row_id, clients_dm.id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))) ], stream key: [ clients_dm.id, _rw_projected_row_id ] }
    └── StreamProject { exprs: [clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file] } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file ], stream key: [ clients_dm.id ] }
        └── MergeExecutor { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.type ], stream key: [ clients_dm.id ] }

Fragment 25665 (Actor 127875,127874)
StreamProject { exprs: [clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.type] } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.type ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.type, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, display_name, local_display_name, preferred_name, customer_identification_file, type, closing_date] } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.type, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, display_name, local_display_name, preferred_name, customer_identification_file, type, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, display_name, local_display_name, preferred_name, customer_identification_file, type, closing_date ], stream key: [] }

Fragment 25666 (Actor 127881,127880)
StreamProject { exprs: [clients_contacts_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 4:Int32] }
├── output: [ clients_contacts_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 4:Int32 ]
├── stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), $2, $1] } { output: [ _rw_projected_row_id, clients_contacts_dm.client_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), clients_contacts_dm.type, clients_contacts_dm.value ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ] }
    └── StreamProject { exprs: [clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type] } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
        └── StreamFilter { predicate: IsNull(clients_contacts_dm.disabled_at) } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type, clients_contacts_dm.disabled_at ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
            └── StreamTableScan { table: clients_contacts_dm, columns: [client_id, value, type, disabled_at] } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type, clients_contacts_dm.disabled_at ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
                ├── Upstream { output: [ client_id, value, type, disabled_at ], stream key: [] }
                └── BatchPlanNode { output: [ client_id, value, type, disabled_at ], stream key: [] }

Fragment 25667 (Actor 127886,127887)
StreamProject { exprs: [accounts_to_clients_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 5:Int32] }
├── output: [ accounts_to_clients_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 5:Int32 ]
├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), $3, $4] } { output: [ _rw_projected_row_id, accounts_to_clients_dm.client_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), accounts_to_clients_dm.account_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, _rw_projected_row_id ] }
    └── StreamProject { exprs: [accounts_to_clients_dm.client_id, accounts_dm.name, accounts_dm.number, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date] } { output: [ accounts_to_clients_dm.client_id, accounts_dm.name, accounts_dm.number, accounts_to_clients_dm.account_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) } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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 25668 (Actor 127884,127885)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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 25669 (Actor 127888,127889)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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.disabled_at, accounts_to_clients_dm.effective_end_date, 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, accounts_dm.name, accounts_dm.number ], stream key: [ accounts_dm.account_id ] }

Fragment 25670 (Actor 127891,127890)
StreamFilter { predicate: IsNull(accounts_to_clients_dm.disabled_at) } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date, 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 ] }
└── StreamTableScan { table: accounts_to_clients_dm, columns: [account_id, client_id, disabled_at, effective_end_date, effective_start_date] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date, 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 ] }
    ├── Upstream { output: [ account_id, client_id, disabled_at, effective_end_date, effective_start_date ], stream key: [] }
    └── BatchPlanNode { output: [ account_id, client_id, disabled_at, effective_end_date, effective_start_date ], stream key: [] }

Fragment 25671 (Actor 127893,127892)
StreamProject { exprs: [accounts_dm.account_id, accounts_dm.name, accounts_dm.number] } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, name, number, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, name, number, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, name, number, disabled_at ], stream key: [] }

Fragment 25672 (Actor 127895,127894)
StreamProject { exprs: [clients_portfolios_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 6:Int32] }
├── output: [ clients_portfolios_dm.client_id, 'client':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 6:Int32 ]
├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), $3, $4] } { output: [ _rw_projected_row_id, clients_portfolios_dm.client_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), 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, _rw_projected_row_id ] }
    └── MergeExecutor { output: [ clients_portfolios_dm.client_id, portfolios_dm.name, portfolios_dm.number, 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 ] }

Fragment 25673 (Actor 127896,127897)
StreamSyncLogStore { output: [ clients_portfolios_dm.client_id, portfolios_dm.name, portfolios_dm.number, 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 ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.portfolio_id = portfolios_dm.portfolio_id } { output: [ clients_portfolios_dm.client_id, portfolios_dm.name, portfolios_dm.number, 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 ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 25674 (Actor 127899,127898)
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) } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at ], 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] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at ], 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 ], stream key: [] }
        └── BatchPlanNode { output: [ client_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 25675 (Actor 127900,127901)
StreamProject { exprs: [portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number ], stream key: [ portfolios_dm.portfolio_id ] }
└── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) } { output: [ portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
    └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, name, number, disabled_at] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
        ├── Upstream { output: [ portfolio_id, name, number, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ portfolio_id, name, number, disabled_at ], stream key: [] }

Fragment 25676 (Actor 127902,127903)
StreamProject { exprs: [party_items_mv_next.entity_id, party_items_mv_next.entity_type, party_items_mv_next.val, party_items_mv_next.ar_val, party_items_mv_next.filter_val, party_items_mv_next.active_parties_mv.id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, null:Date, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Int64, party_items_mv_next.null:Int32, party_items_mv_next.null:Int32#1, party_items_mv_next.$src, 7:Int32] }
├── output: [ party_items_mv_next.entity_id, party_items_mv_next.entity_type, party_items_mv_next.val, party_items_mv_next.ar_val, party_items_mv_next.filter_val, party_items_mv_next.active_parties_mv.id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, null:Date, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Int64, party_items_mv_next.null:Int32, party_items_mv_next.null:Int32#1, party_items_mv_next.$src, 7:Int32 ]
├── stream key: [ party_items_mv_next.active_parties_mv.id, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Int32, party_items_mv_next.null:Int64, party_items_mv_next.null:Date, party_items_mv_next.null:Int32#1, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.$src ]
└── StreamTableScan { table: party_items_mv_next, columns: [entity_id, entity_type, val, ar_val, filter_val, active_parties_mv.id, _rw_projected_row_id, null:Varchar, null:Int32, null:Int64, null:Date, null:Int32#1, null:Varchar#1, null:Varchar#2, $src] }
    ├── output: [ party_items_mv_next.entity_id, party_items_mv_next.entity_type, party_items_mv_next.val, party_items_mv_next.ar_val, party_items_mv_next.filter_val, party_items_mv_next.active_parties_mv.id, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Int32, party_items_mv_next.null:Int64, party_items_mv_next.null:Date, party_items_mv_next.null:Int32#1, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.$src ]
    ├── stream key: [ party_items_mv_next.active_parties_mv.id, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Int32, party_items_mv_next.null:Int64, party_items_mv_next.null:Date, party_items_mv_next.null:Int32#1, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.$src ]
    ├── Upstream { output: [ entity_id, entity_type, val, ar_val, filter_val, active_parties_mv.id, _rw_projected_row_id, null:Varchar, null:Int32, null:Int64, null:Date, null:Int32#1, null:Varchar#1, null:Varchar#2, $src ], stream key: [] }
    └── BatchPlanNode { output: [ entity_id, entity_type, val, ar_val, filter_val, active_parties_mv.id, _rw_projected_row_id, null:Varchar, null:Int32, null:Int64, null:Date, null:Int32#1, null:Varchar#1, null:Varchar#2, $src ], stream key: [] }

Fragment 25677 (Actor 127905,127904)
StreamProject { exprs: [portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, portfolios_dm.portfolio_id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 8:Int32] }
├── output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, portfolios_dm.portfolio_id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 8:Int32 ]
├── stream key: [ portfolios_dm.portfolio_id, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar)))] } { output: [ _rw_projected_row_id, portfolios_dm.portfolio_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))) ], stream key: [ portfolios_dm.portfolio_id, _rw_projected_row_id ] }
    └── StreamProject { exprs: [portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number ], stream key: [ portfolios_dm.portfolio_id ] }
        └── StreamFilter { predicate: IsNull(portfolios_dm.disabled_at) } { output: [ portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
            └── StreamTableScan { table: portfolios_dm, columns: [portfolio_id, name, number, disabled_at] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.name, portfolios_dm.number, portfolios_dm.disabled_at ], stream key: [ portfolios_dm.portfolio_id ] }
                ├── Upstream { output: [ portfolio_id, name, number, disabled_at ], stream key: [] }
                └── BatchPlanNode { output: [ portfolio_id, name, number, disabled_at ], stream key: [] }

Fragment 25678 (Actor 127908,127909)
StreamProject { exprs: [account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 9:Int32] }
├── output: [ account_to_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, account_to_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 9:Int32 ]
├── stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), $3, $4] } { output: [ _rw_projected_row_id, account_to_portfolios_dm.portfolio_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), account_to_portfolios_dm.account_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, _rw_projected_row_id ] }
    └── MergeExecutor { output: [ account_to_portfolios_dm.portfolio_id, accounts_dm.name, accounts_dm.number, account_to_portfolios_dm.account_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 25679 (Actor 127907,127906)
StreamSyncLogStore { output: [ account_to_portfolios_dm.portfolio_id, accounts_dm.name, accounts_dm.number, account_to_portfolios_dm.account_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.portfolio_id, accounts_dm.name, accounts_dm.number, account_to_portfolios_dm.account_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, accounts_dm.name, accounts_dm.number ], stream key: [ accounts_dm.account_id ] }

Fragment 25680 (Actor 127910,127911)
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) } { 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 ], 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] } { 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 ], 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 ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 25681 (Actor 127912,127913)
StreamProject { exprs: [accounts_dm.account_id, accounts_dm.name, accounts_dm.number] } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number ], stream key: [ accounts_dm.account_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    └── StreamTableScan { table: accounts_dm, columns: [account_id, name, number, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.name, accounts_dm.number, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
        ├── Upstream { output: [ account_id, name, number, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, name, number, disabled_at ], stream key: [] }

Fragment 25682 (Actor 127916,127917)
StreamProject { exprs: [clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 10:Int32] }
├── output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 10:Int32 ]
├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), $5, $6] } { output: [ _rw_projected_row_id, clients_portfolios_dm.portfolio_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar))), clients_portfolios_dm.client_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, _rw_projected_row_id ] }
    └── MergeExecutor { output: [ clients_portfolios_dm.portfolio_id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_portfolios_dm.client_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 25683 (Actor 127915,127914)
StreamSyncLogStore { output: [ clients_portfolios_dm.portfolio_id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_portfolios_dm.client_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.portfolio_id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_portfolios_dm.client_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, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file ], stream key: [ clients_dm.id ] }

Fragment 25684 (Actor 127919,127918)
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) } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at ], 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] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at ], 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 ], stream key: [] }
        └── BatchPlanNode { output: [ client_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 25685 (Actor 127920,127921)
StreamProject { exprs: [clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file] } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file ], stream key: [ clients_dm.id ] }
└── StreamFilter { predicate: IsNull(clients_dm.closing_date) } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
    └── StreamTableScan { table: clients_dm, columns: [id, display_name, local_display_name, preferred_name, customer_identification_file, closing_date] } { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.closing_date ], stream key: [ clients_dm.id ] }
        ├── Upstream { output: [ id, display_name, local_display_name, preferred_name, customer_identification_file, closing_date ], stream key: [] }
        └── BatchPlanNode { output: [ id, display_name, local_display_name, preferred_name, customer_identification_file, closing_date ], stream key: [] }

Fragment 25686 (Actor 127925,127924)
StreamProject { exprs: [clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_contacts_dm.type, clients_contacts_dm.value, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 11:Int32] }
├── output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), null:Varchar, null:Varchar, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_contacts_dm.type, clients_contacts_dm.value, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 11:Int32 ]
├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), $2, $3, $5, $1] } { output: [ _rw_projected_row_id, clients_portfolios_dm.portfolio_id, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar))), clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ] }
    └── MergeExecutor { output: [ clients_portfolios_dm.portfolio_id, clients_contacts_dm.value, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.client_id, clients_contacts_dm.type ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ] }

Fragment 25687 (Actor 127922,127923)
StreamSyncLogStore { output: [ clients_portfolios_dm.portfolio_id, clients_contacts_dm.value, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.client_id, clients_contacts_dm.type ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.client_id = clients_contacts_dm.client_id } { output: [ clients_portfolios_dm.portfolio_id, clients_contacts_dm.value, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.client_id, clients_contacts_dm.type ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ] }
    ├── 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_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }

Fragment 25688 (Actor 127926,127927)
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) } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at ], 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] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_portfolios_dm.disabled_at ], 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 ], stream key: [] }
        └── BatchPlanNode { output: [ client_id, portfolio_id, effective_start_date, disabled_at ], stream key: [] }

Fragment 25689 (Actor 127928,127929)
StreamProject { exprs: [clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type] } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
└── StreamFilter { predicate: IsNull(clients_contacts_dm.disabled_at) } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type, clients_contacts_dm.disabled_at ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
    └── StreamTableScan { table: clients_contacts_dm, columns: [client_id, value, type, disabled_at] } { output: [ clients_contacts_dm.client_id, clients_contacts_dm.value, clients_contacts_dm.type, clients_contacts_dm.disabled_at ], stream key: [ clients_contacts_dm.client_id, clients_contacts_dm.type, clients_contacts_dm.value ] }
        ├── Upstream { output: [ client_id, value, type, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ client_id, value, type, disabled_at ], stream key: [] }

Fragment 25690 (Actor 127930,127931)
StreamProject { exprs: [assets_dm_next.id, 'asset':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar))), null:Varchar, Lower(assets_dm_next.type) as $expr1, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 12:Int32] }
├── output: [ assets_dm_next.id, 'asset':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar))), null:Varchar, $expr1, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, 12:Int32 ]
├── stream key: [ assets_dm_next.id, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, $2, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar)))] }
    ├── output: [ _rw_projected_row_id, assets_dm_next.id, assets_dm_next.type, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar))) ]
    ├── stream key: [ assets_dm_next.id, _rw_projected_row_id ]
    └── StreamProject { exprs: [assets_dm_next.id, assets_dm_next.name_en, assets_dm_next.type, assets_dm_next.ticker, assets_dm_next.isin, assets_dm_next.cusip, assets_dm_next.sedol, assets_dm_next.ric, assets_dm_next.figi] } { output: [ assets_dm_next.id, assets_dm_next.name_en, assets_dm_next.type, assets_dm_next.ticker, assets_dm_next.isin, assets_dm_next.cusip, assets_dm_next.sedol, assets_dm_next.ric, assets_dm_next.figi ], stream key: [ assets_dm_next.id ] }
        └── StreamFilter { predicate: IsNull(assets_dm_next.disabled_at) } { output: [ assets_dm_next.id, assets_dm_next.name_en, assets_dm_next.type, assets_dm_next.ticker, assets_dm_next.isin, assets_dm_next.cusip, assets_dm_next.sedol, assets_dm_next.ric, assets_dm_next.figi, assets_dm_next.disabled_at ], stream key: [ assets_dm_next.id ] }
            └── StreamTableScan { table: assets_dm_next, columns: [id, name_en, type, ticker, isin, cusip, sedol, ric, figi, disabled_at] } { output: [ assets_dm_next.id, assets_dm_next.name_en, assets_dm_next.type, assets_dm_next.ticker, assets_dm_next.isin, assets_dm_next.cusip, assets_dm_next.sedol, assets_dm_next.ric, assets_dm_next.figi, assets_dm_next.disabled_at ], stream key: [ assets_dm_next.id ] }
                ├── Upstream { output: [ id, name_en, type, ticker, isin, cusip, sedol, ric, figi, disabled_at ], stream key: [] }
                └── BatchPlanNode { output: [ id, name_en, type, ticker, isin, cusip, sedol, ric, figi, disabled_at ], stream key: [] }

Fragment 25691 (Actor 127932,127933)
StreamProject { exprs: [assets_dm_next.id, 'asset':Varchar, null:Varchar, assets_dm_next.name_ar, Lower(assets_dm_next.type) as $expr2, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 13:Int32] } { output: [ assets_dm_next.id, 'asset':Varchar, null:Varchar, assets_dm_next.name_ar, $expr2, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 13:Int32 ], stream key: [ assets_dm_next.id ] }
└── StreamFilter { predicate: IsNull(assets_dm_next.disabled_at) } { output: [ assets_dm_next.id, assets_dm_next.name_ar, assets_dm_next.type, assets_dm_next.disabled_at ], stream key: [ assets_dm_next.id ] }
    └── StreamTableScan { table: assets_dm_next, columns: [id, name_ar, type, disabled_at] } { output: [ assets_dm_next.id, assets_dm_next.name_ar, assets_dm_next.type, assets_dm_next.disabled_at ], stream key: [ assets_dm_next.id ] }
        ├── Upstream { output: [ id, name_ar, type, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ id, name_ar, type, disabled_at ], stream key: [] }

Fragment 25692 (Actor 127878,127879)
StreamProject { exprs: [clients_dm.id, 'client':Varchar, null:Varchar, null:Varchar, ConcatOp('client_type:':Varchar, Lower(clients_dm.type)) as $expr3, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 14:Int32] } { output: [ clients_dm.id, 'client':Varchar, null:Varchar, null:Varchar, $expr3, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 14:Int32 ], stream key: [ clients_dm.id ] }
└── MergeExecutor { output: [ clients_dm.id, clients_dm.display_name, clients_dm.local_display_name, clients_dm.preferred_name, clients_dm.customer_identification_file, clients_dm.type ], stream key: [ clients_dm.id ] }

Fragment 25693 (Actor 127934,127935)
StreamProject { exprs: [clients_dm.id, 'client':Varchar, null:Varchar, null:Varchar, ConcatOp('status:':Varchar, Lower(labels_dm.name_en)) as $expr4, clients_dm.id, clients_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 15:Int32] } { output: [ clients_dm.id, 'client':Varchar, null:Varchar, null:Varchar, $expr4, clients_dm.id, clients_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 15:Int32 ], stream key: [ clients_dm.id, clients_dm.status_label_id ] }
└── MergeExecutor { output: [ clients_dm.id, labels_dm.name_en, clients_dm.status_label_id, labels_dm.label_id ], stream key: [ clients_dm.id, clients_dm.status_label_id ] }

Fragment 25694 (Actor 127936,127937)
StreamSyncLogStore { output: [ clients_dm.id, labels_dm.name_en, clients_dm.status_label_id, labels_dm.label_id ], stream key: [ clients_dm.id, clients_dm.status_label_id ] }
└── StreamHashJoin { type: Inner, predicate: clients_dm.status_label_id = labels_dm.label_id } { output: [ clients_dm.id, labels_dm.name_en, clients_dm.status_label_id, labels_dm.label_id ], stream key: [ clients_dm.id, clients_dm.status_label_id ] }
    ├── MergeExecutor { output: [ clients_dm.id, clients_dm.status_label_id ], stream key: [ clients_dm.id ] }
    └── MergeExecutor { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }

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

Fragment 25696 (Actor 127941,127940)
StreamTableScan { table: labels_dm, columns: [label_id, name_en] } { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }
├── Upstream { output: [ label_id, name_en ], stream key: [] }
└── BatchPlanNode { output: [ label_id, name_en ], stream key: [] }

Fragment 25697 (Actor 127944,127945)
StreamProject { exprs: [entity_to_teams_dm.entity_id, 'client':Varchar, null:Varchar, null:Varchar, ConcatOp('team:':Varchar, entity_to_teams_dm.team_id) as $expr5, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 16:Int32] }
├── output: [ entity_to_teams_dm.entity_id, 'client':Varchar, null:Varchar, null:Varchar, $expr5, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 16:Int32 ]
├── stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ]
└── MergeExecutor { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, clients_dm.id ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 25698 (Actor 127942,127943)
StreamSyncLogStore { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, clients_dm.id ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: entity_to_teams_dm.entity_id = clients_dm.id } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, clients_dm.id ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    └── MergeExecutor { output: [ clients_dm.id ], stream key: [ clients_dm.id ] }

Fragment 25699 (Actor 127946,127947)
StreamProject { exprs: [entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date] } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
└── StreamFilter { predicate: (entity_to_teams_dm.entity_type = 'CLIENT':Varchar) AND IsNull(entity_to_teams_dm.disabled_at) AND IsNull(entity_to_teams_dm.effective_end_date) } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, entity_to_teams_dm.effective_end_date, entity_to_teams_dm.disabled_at ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    └── StreamTableScan { table: entity_to_teams_dm, columns: [team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at] } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, entity_to_teams_dm.effective_end_date, entity_to_teams_dm.disabled_at ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
        ├── Upstream { output: [ team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at ], stream key: [] }

Fragment 25700 (Actor 127949,127948)
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 25701 (Actor 127604,127605)
StreamProject { exprs: [accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, ConcatOp('account_type:':Varchar, Lower(product_types_dm.type)) as $expr6, accounts_dm.account_id, accounts_dm.product_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 17:Int32] } { output: [ accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr6, accounts_dm.account_id, accounts_dm.product_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 17:Int32 ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at, product_types_dm.type, product_types_dm.name_en, accounts_dm.product_type_id, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }
    └── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.disabled_at, product_types_dm.type, product_types_dm.name_en, accounts_dm.product_type_id, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }

Fragment 25702 (Actor 127603,127602)
StreamTemporalJoin { type: Inner, append_only: false, predicate: accounts_dm.product_type_id = product_types_dm.product_type_id, nested_loop: false } { output: [ accounts_dm.account_id, accounts_dm.disabled_at, product_types_dm.type, product_types_dm.name_en, accounts_dm.product_type_id, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }
├── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.product_type_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
└── MergeExecutor { output: [ product_types_dm.product_type_id, product_types_dm.type, product_types_dm.name_en ], stream key: [ product_types_dm.product_type_id ] }

Fragment 25703 (Actor 127950,127951)
StreamFilter { predicate: IsNull(accounts_dm.disabled_at) } { output: [ accounts_dm.account_id, accounts_dm.product_type_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
└── StreamTableScan { table: accounts_dm, columns: [account_id, product_type_id, disabled_at] } { output: [ accounts_dm.account_id, accounts_dm.product_type_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }
    ├── Upstream { output: [ account_id, product_type_id, disabled_at ], stream key: [] }
    └── BatchPlanNode { output: [ account_id, product_type_id, disabled_at ], stream key: [] }

Fragment 25704 (Actor 127601,127600)
StreamTableScan { table: product_types_dm, columns: [product_type_id, type, name_en] } { output: [ product_types_dm.product_type_id, product_types_dm.type, product_types_dm.name_en ], stream key: [ product_types_dm.product_type_id ] }
├── Upstream { output: [ product_type_id, type, name_en ], stream key: [] }
└── BatchPlanNode { output: [ product_type_id, type, name_en ], stream key: [] }

Fragment 25705 (Actor 127598,127599)
StreamProject { exprs: [accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, ConcatOp('product:':Varchar, Lower(Trim(product_types_dm.name_en))) as $expr7, accounts_dm.account_id, accounts_dm.product_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 18:Int32] } { output: [ accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr7, accounts_dm.account_id, accounts_dm.product_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 18:Int32 ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }
└── StreamFilter { predicate: IsNull(accounts_dm.disabled_at) AND Not(IsNull(product_types_dm.name_en)) AND (Trim(product_types_dm.name_en) <> '':Varchar) } { output: [ accounts_dm.account_id, accounts_dm.disabled_at, product_types_dm.type, product_types_dm.name_en, accounts_dm.product_type_id, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }
    └── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.disabled_at, product_types_dm.type, product_types_dm.name_en, accounts_dm.product_type_id, product_types_dm.product_type_id ], stream key: [ accounts_dm.account_id, accounts_dm.product_type_id ] }

Fragment 25706 (Actor 127954,127955)
StreamProject { exprs: [accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, ConcatOp('status:':Varchar, Lower(labels_dm.name_en)) as $expr8, accounts_dm.account_id, accounts_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 19:Int32] } { output: [ accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr8, accounts_dm.account_id, accounts_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 19:Int32 ], stream key: [ accounts_dm.account_id, accounts_dm.status_label_id ] }
└── MergeExecutor { output: [ accounts_dm.account_id, labels_dm.name_en, accounts_dm.status_label_id, labels_dm.label_id ], stream key: [ accounts_dm.account_id, accounts_dm.status_label_id ] }

Fragment 25707 (Actor 127952,127953)
StreamSyncLogStore { output: [ accounts_dm.account_id, labels_dm.name_en, accounts_dm.status_label_id, labels_dm.label_id ], stream key: [ accounts_dm.account_id, accounts_dm.status_label_id ] }
└── StreamHashJoin { type: Inner, predicate: accounts_dm.status_label_id = labels_dm.label_id } { output: [ accounts_dm.account_id, labels_dm.name_en, accounts_dm.status_label_id, labels_dm.label_id ], stream key: [ accounts_dm.account_id, accounts_dm.status_label_id ] }
    ├── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.status_label_id ], stream key: [ accounts_dm.account_id ] }
    └── MergeExecutor { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }

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

Fragment 25709 (Actor 127958,127959)
StreamTableScan { table: labels_dm, columns: [label_id, name_en] } { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }
├── Upstream { output: [ label_id, name_en ], stream key: [] }
└── BatchPlanNode { output: [ label_id, name_en ], stream key: [] }

Fragment 25710 (Actor 127961,127960)
StreamProject { exprs: [accounts_to_clients_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, ConcatOp('team:':Varchar, entity_to_teams_dm.team_id) as $expr9, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 20:Int32] }
├── output: [ accounts_to_clients_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr9, accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 20:Int32 ]
├── stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ]
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, entity_to_teams_dm.team_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_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, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 25711 (Actor 127963,127962)
StreamSyncLogStore { output: [ accounts_to_clients_dm.account_id, entity_to_teams_dm.team_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_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, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: accounts_to_clients_dm.client_id = entity_to_teams_dm.entity_id } { output: [ accounts_to_clients_dm.account_id, entity_to_teams_dm.team_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_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, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 25712 (Actor 127882,127883)
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.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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.disabled_at, accounts_to_clients_dm.effective_end_date, accounts_dm.name, accounts_dm.number, 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 25713 (Actor 127964,127965)
StreamProject { exprs: [entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date] } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
└── StreamFilter { predicate: (entity_to_teams_dm.entity_type = 'CLIENT':Varchar) AND IsNull(entity_to_teams_dm.disabled_at) AND IsNull(entity_to_teams_dm.effective_end_date) } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, entity_to_teams_dm.effective_end_date, entity_to_teams_dm.disabled_at ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    └── StreamTableScan { table: entity_to_teams_dm, columns: [team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at] } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, entity_to_teams_dm.effective_end_date, entity_to_teams_dm.disabled_at ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
        ├── Upstream { output: [ team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at ], stream key: [] }

Fragment 25714 (Actor 127462,127463)
StreamProject { exprs: [portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, ConcatOp('portfolio_type:':Varchar, Lower(service_types_dm.type)) as $expr10, portfolios_dm.portfolio_id, portfolios_dm.service_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 21:Int32] } { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr10, portfolios_dm.portfolio_id, portfolios_dm.service_type_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 21:Int32 ], stream key: [ portfolios_dm.portfolio_id, portfolios_dm.service_type_id ] }
└── StreamTemporalJoin { type: Inner, append_only: false, predicate: portfolios_dm.service_type_id = service_types_dm.service_type_id, nested_loop: false } { output: [ portfolios_dm.portfolio_id, service_types_dm.type, portfolios_dm.service_type_id, service_types_dm.service_type_id ], stream key: [ portfolios_dm.portfolio_id, portfolios_dm.service_type_id ] }
    ├── MergeExecutor { output: [ portfolios_dm.portfolio_id, portfolios_dm.service_type_id ], stream key: [ portfolios_dm.portfolio_id ] }
    └── MergeExecutor { output: [ service_types_dm.service_type_id, service_types_dm.type ], stream key: [ service_types_dm.service_type_id ] }

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

Fragment 25716 (Actor 127465,127464)
StreamTableScan { table: service_types_dm, columns: [service_type_id, type] } { output: [ service_types_dm.service_type_id, service_types_dm.type ], stream key: [ service_types_dm.service_type_id ] }
├── Upstream { output: [ service_type_id, type ], stream key: [] }
└── BatchPlanNode { output: [ service_type_id, type ], stream key: [] }

Fragment 25717 (Actor 127969,127968)
StreamProject { exprs: [portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, ConcatOp('status:':Varchar, Lower(labels_dm.name_en)) as $expr11, portfolios_dm.portfolio_id, portfolios_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 22:Int32] } { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr11, portfolios_dm.portfolio_id, portfolios_dm.status_label_id, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 22:Int32 ], stream key: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ] }
└── MergeExecutor { output: [ portfolios_dm.portfolio_id, labels_dm.name_en, portfolios_dm.status_label_id, labels_dm.label_id ], stream key: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ] }

Fragment 25718 (Actor 127970,127971)
StreamSyncLogStore { output: [ portfolios_dm.portfolio_id, labels_dm.name_en, portfolios_dm.status_label_id, labels_dm.label_id ], stream key: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ] }
└── StreamHashJoin { type: Inner, predicate: portfolios_dm.status_label_id = labels_dm.label_id } { output: [ portfolios_dm.portfolio_id, labels_dm.name_en, portfolios_dm.status_label_id, labels_dm.label_id ], stream key: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ] }
    ├── MergeExecutor { output: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ], stream key: [ portfolios_dm.portfolio_id ] }
    └── MergeExecutor { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }

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

Fragment 25720 (Actor 127974,127975)
StreamTableScan { table: labels_dm, columns: [label_id, name_en] } { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }
├── Upstream { output: [ label_id, name_en ], stream key: [] }
└── BatchPlanNode { output: [ label_id, name_en ], stream key: [] }

Fragment 25721 (Actor 127977,127976)
StreamProject { exprs: [entity_to_teams_dm.entity_id, 'portfolio':Varchar, null:Varchar, null:Varchar, ConcatOp('team:':Varchar, entity_to_teams_dm.team_id) as $expr12, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 23:Int32] }
├── output: [ entity_to_teams_dm.entity_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr12, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 23:Int32 ]
├── stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ]
└── MergeExecutor { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 25722 (Actor 127978,127979)
StreamSyncLogStore { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: entity_to_teams_dm.entity_id = portfolios_dm.portfolio_id } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, portfolios_dm.portfolio_id ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    ├── MergeExecutor { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    └── MergeExecutor { output: [ portfolios_dm.portfolio_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 25723 (Actor 127980,127981)
StreamProject { exprs: [entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date] } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
└── StreamFilter { predicate: (entity_to_teams_dm.entity_type = 'PORTFOLIO':Varchar) AND IsNull(entity_to_teams_dm.disabled_at) AND IsNull(entity_to_teams_dm.effective_end_date) } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, entity_to_teams_dm.effective_end_date, entity_to_teams_dm.disabled_at ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    └── StreamTableScan { table: entity_to_teams_dm, columns: [team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at] } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, entity_to_teams_dm.effective_end_date, entity_to_teams_dm.disabled_at ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
        ├── Upstream { output: [ team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at ], stream key: [] }

Fragment 25724 (Actor 127983,127982)
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 25725 (Actor 127986,127987)
StreamProject { exprs: [clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, ConcatOp('team:':Varchar, entity_to_teams_dm.team_id) as $expr13, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 24:Int32] }
├── output: [ clients_portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr13, clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, 24:Int32 ]
├── stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ]
└── MergeExecutor { output: [ clients_portfolios_dm.portfolio_id, entity_to_teams_dm.team_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 25726 (Actor 127984,127985)
StreamSyncLogStore { output: [ clients_portfolios_dm.portfolio_id, entity_to_teams_dm.team_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
└── StreamHashJoin { type: Inner, predicate: clients_portfolios_dm.client_id = entity_to_teams_dm.entity_id } { output: [ clients_portfolios_dm.portfolio_id, entity_to_teams_dm.team_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_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: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 25727 (Actor 127988,127989)
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 25728 (Actor 127991,127990)
StreamProject { exprs: [entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date] } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
└── StreamFilter { predicate: (entity_to_teams_dm.entity_type = 'CLIENT':Varchar) AND IsNull(entity_to_teams_dm.disabled_at) AND IsNull(entity_to_teams_dm.effective_end_date) } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, entity_to_teams_dm.effective_end_date, entity_to_teams_dm.disabled_at ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
    └── StreamTableScan { table: entity_to_teams_dm, columns: [team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at] } { output: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date, entity_to_teams_dm.effective_end_date, entity_to_teams_dm.disabled_at ], stream key: [ entity_to_teams_dm.team_id, entity_to_teams_dm.entity_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }
        ├── Upstream { output: [ team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ team_id, entity_id, entity_type, effective_start_date, effective_end_date, disabled_at ], stream key: [] }

Fragment 25729 (Actor 127994,127995)
StreamProject { exprs: [active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, olap_reference_identifier_terms_mv.val, olap_reference_identifier_terms_mv.ar_val, null:Varchar, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, active_identifier_edges_mv_next.null:Date, null:Date, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, null:Int32, 25:Int32] }
├── output: [ active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, olap_reference_identifier_terms_mv.val, olap_reference_identifier_terms_mv.ar_val, null:Varchar, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, active_identifier_edges_mv_next.null:Date, null:Date, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, null:Int32, 25:Int32 ]
├── stream key: [ active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type ]
└── MergeExecutor
    ├── output: [ active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, olap_reference_identifier_terms_mv.val, olap_reference_identifier_terms_mv.ar_val, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]
    └── stream key: [ active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type ]

Fragment 25730 (Actor 127993,127992)
StreamSyncLogStore
├── output: [ active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, olap_reference_identifier_terms_mv.val, olap_reference_identifier_terms_mv.ar_val, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]
├── stream key: [ active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type ]
└── StreamHashJoin { type: Inner, predicate: active_identifier_edges_mv_next.owner_entity_id = olap_reference_identifier_terms_mv.owner_entity_id AND active_identifier_edges_mv_next.owner_entity_type = olap_reference_identifier_terms_mv.owner_entity_type }
    ├── output: [ active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, olap_reference_identifier_terms_mv.val, olap_reference_identifier_terms_mv.ar_val, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]
    ├── stream key: [ active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type ]
    ├── MergeExecutor { output: [ active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src ], stream key: [ active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src ] }
    └── MergeExecutor { output: [ olap_reference_identifier_terms_mv.owner_entity_id, olap_reference_identifier_terms_mv.owner_entity_type, olap_reference_identifier_terms_mv.val, olap_reference_identifier_terms_mv.ar_val, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ], stream key: [ olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ] }

Fragment 25731 (Actor 127997,127996)
StreamFilter { predicate: In(active_identifier_edges_mv_next.target_entity_type, 'account':Varchar, 'client':Varchar, 'portfolio':Varchar) }
├── output: [ active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src ]
├── stream key: [ active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src ]
└── StreamTableScan { table: active_identifier_edges_mv_next, columns: [target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, accounts_dm.account_id, null:Varchar, null:Date, null:Int32, $src] }
    ├── output: [ active_identifier_edges_mv_next.target_entity_id, active_identifier_edges_mv_next.target_entity_type, active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type, active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src ]
    ├── stream key: [ active_identifier_edges_mv_next.accounts_dm.account_id, active_identifier_edges_mv_next.null:Varchar, active_identifier_edges_mv_next.null:Date, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.$src ]
    ├── Upstream { output: [ target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, accounts_dm.account_id, null:Varchar, null:Date, null:Int32, $src ], stream key: [] }
    └── BatchPlanNode { output: [ target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, accounts_dm.account_id, null:Varchar, null:Date, null:Int32, $src ], stream key: [] }

Fragment 25732 (Actor 127998,127999)
StreamTableScan { table: olap_reference_identifier_terms_mv, columns: [owner_entity_id, owner_entity_type, val, ar_val, reference_identifiers.id, _rw_projected_row_id] } { output: [ olap_reference_identifier_terms_mv.owner_entity_id, olap_reference_identifier_terms_mv.owner_entity_type, olap_reference_identifier_terms_mv.val, olap_reference_identifier_terms_mv.ar_val, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ], stream key: [ olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ] }
├── Upstream { output: [ owner_entity_id, owner_entity_type, val, ar_val, reference_identifiers.id, _rw_projected_row_id ], stream key: [] }
└── BatchPlanNode { output: [ owner_entity_id, owner_entity_type, val, ar_val, reference_identifiers.id, _rw_projected_row_id ], stream key: [] }

Fragment 25733 (Actor 128002,128003)
StreamProject { exprs: [party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, null:Varchar, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.owner_entity_type, null:Varchar, null:Date, null:Date, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, null:Int32, null:Int32, 26:Int32] }
├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, null:Varchar, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.owner_entity_type, null:Varchar, null:Date, null:Date, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, null:Int32, null:Int32, 26:Int32 ]
├── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.owner_entity_type ]
└── MergeExecutor
    ├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ]
    └── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.owner_entity_type ]

Fragment 25734 (Actor 128001,128000)
StreamSyncLogStore
├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ]
├── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.owner_entity_type ]
└── StreamHashJoin { type: Inner, predicate: party_identifier_edges_mv_next.owner_entity_id = party_reference_identifier_terms_mv_next.owner_entity_id AND party_identifier_edges_mv_next.owner_entity_type = party_reference_identifier_terms_mv_next.owner_entity_type }
    ├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ]
    ├── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.owner_entity_type ]
    ├── MergeExecutor { output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src ], stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src ] }
    └── MergeExecutor { output: [ party_reference_identifier_terms_mv_next.owner_entity_id, party_reference_identifier_terms_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ] }

Fragment 25735 (Actor 128005,128004)
StreamTableScan { table: party_identifier_edges_mv_next, columns: [target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, party_holder_edges_mv_next.$src] } { output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src ], stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src ] }
├── Upstream { output: [ target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, party_holder_edges_mv_next.$src ], stream key: [] }
└── BatchPlanNode { output: [ target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, party_holder_edges_mv_next.$src ], stream key: [] }

Fragment 25736 (Actor 128006,128007)
StreamTableScan { table: party_reference_identifier_terms_mv_next, columns: [owner_entity_id, owner_entity_type, val, ar_val, reference_identifiers_next.id, _rw_projected_row_id] } { output: [ party_reference_identifier_terms_mv_next.owner_entity_id, party_reference_identifier_terms_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ] }
├── Upstream { output: [ owner_entity_id, owner_entity_type, val, ar_val, reference_identifiers_next.id, _rw_projected_row_id ], stream key: [] }
└── BatchPlanNode { output: [ owner_entity_id, owner_entity_type, val, ar_val, reference_identifiers_next.id, _rw_projected_row_id ], stream key: [] }

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