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

514 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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Union

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · party_reference_identifier_terms_mv

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

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

LocalityProvider

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

LocalityProvider

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

LocalityProvider

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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · clients_portfolios_dm

— — 2 actors

Filter · clients_portfolios_dm

0% idle 2 actors

StreamScan · clients_portfolios_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · Inner · 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

LocalityProvider

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

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

LocalityProvider

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

LocalityProvider

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

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

LocalityProvider

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

LocalityProvider

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

LocalityProvider

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

LocalityProvider

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

LocalityProvider

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

LocalityProvider

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

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

LocalityProvider

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

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

— — 2 actors

Filter · assets_dm

0% idle 2 actors

StreamScan · assets_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · assets_dm

— — 2 actors

ProjectSet · assets_dm

0% idle 2 actors

Project · assets_dm

— — 2 actors

Filter · assets_dm

0% idle 2 actors

StreamScan · assets_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_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

LocalityProvider

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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · clients_portfolios_dm

— — 2 actors

Filter · clients_portfolios_dm

0% idle 2 actors

StreamScan · clients_portfolios_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · clients_portfolios_dm

— — 2 actors

Filter · clients_portfolios_dm

0% idle 2 actors

StreamScan · clients_portfolios_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · account_to_portfolios_dm

— — 2 actors

Filter · account_to_portfolios_dm

0% idle 2 actors

StreamScan · account_to_portfolios_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · portfolios_dm

— — 2 actors

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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · clients_portfolios_dm

— — 2 actors

Filter · clients_portfolios_dm

0% idle 2 actors

StreamScan · clients_portfolios_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

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

LocalityProvider

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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · accounts_to_clients_dm

— — 2 actors

Filter · accounts_to_clients_dm

0% idle 2 actors

StreamScan · accounts_to_clients_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

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

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · accounts_to_clients_dm

— — 2 actors

Filter · accounts_to_clients_dm

0% idle 2 actors

StreamScan · accounts_to_clients_dm

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · 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 18256 (Actor 159098,159097)
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 ]
        └── StreamLocalityProvider { locality_columns: [accounts_dm.account_id, 'account':Varchar] } { 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, 'account':Varchar, 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, $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 18257 (Actor 159102,159101)
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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.portfolio_id, clients_portfolios_dm.client_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.portfolio_id, clients_portfolios_dm.client_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:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, party_items_mv_next.null:Int32#1, 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.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.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.id, _rw_projected_row_id ] }
├── MergeExecutor { output: [ assets_dm.id, 'asset':Varchar, null:Varchar, assets_dm.name_ar, $expr2, assets_dm.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.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.status_label_id, clients_dm.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.status_label_id, clients_dm.id ] }
├── MergeExecutor { output: [ entity_to_teams_dm.entity_id, 'client':Varchar, null:Varchar, null:Varchar, $expr5, entity_to_teams_dm.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_id ] }
├── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr7, accounts_dm.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_id ] }
├── MergeExecutor { output: [ accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr8, accounts_dm.status_label_id, accounts_dm.account_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.status_label_id, accounts_dm.account_id ] }
├── MergeExecutor { output: [ accounts_to_clients_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, $expr9, accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.service_type_id, portfolios_dm.portfolio_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.service_type_id, portfolios_dm.portfolio_id ] }
├── MergeExecutor { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr11, portfolios_dm.status_label_id, portfolios_dm.portfolio_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.status_label_id, portfolios_dm.portfolio_id ] }
├── MergeExecutor { output: [ entity_to_teams_dm.entity_id, 'portfolio':Varchar, null:Varchar, null:Varchar, $expr12, entity_to_teams_dm.entity_id, entity_to_teams_dm.team_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.entity_id, 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, '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.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, 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.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, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]
└── MergeExecutor
    ├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv.val, party_reference_identifier_terms_mv.ar_val, null:Varchar, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, null:Varchar, null:Date, null:Date, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._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.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ]

Fragment 18258 (Actor 159104,159103)
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 18259 (Actor 159324,159323)
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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }

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

Fragment 18261 (Actor 159327,159328)
StreamLocalityProvider { locality_columns: [accounts_to_clients_dm.client_id] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 18262 (Actor 159330,159329)
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 18263 (Actor 159332,159331)
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 18264 (Actor 159334,159333)
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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value ] }

Fragment 18265 (Actor 159335,159336)
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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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 18266 (Actor 159337,159338)
StreamLocalityProvider { locality_columns: [accounts_to_clients_dm.client_id] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ], stream key: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.effective_start_date ] }

Fragment 18267 (Actor 159340,159339)
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 18268 (Actor 159342,159341)
StreamLocalityProvider { locality_columns: [clients_contacts_dm.client_id] } { 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 ] }
└── 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 18269 (Actor 159343,159344)
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 18270 (Actor 159350,159349)
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 18271 (Actor 159348,159347)
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 18272 (Actor 159352,159351)
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 18273 (Actor 159355,159356)
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 18274 (Actor 159358,159357)
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 18275 (Actor 159354,159353)
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 18276 (Actor 159361,159362)
StreamLocalityProvider { locality_columns: [accounts_to_clients_dm.account_id] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.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_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 ] }

Fragment 18277 (Actor 159363,159364)
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 18278 (Actor 159365,159366)
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 18279 (Actor 159370,159369)
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.portfolio_id, clients_portfolios_dm.client_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.portfolio_id, clients_portfolios_dm.client_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.portfolio_id, clients_portfolios_dm.client_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.portfolio_id, clients_portfolios_dm.client_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.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }

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

Fragment 18281 (Actor 159371,159372)
StreamLocalityProvider { locality_columns: [clients_portfolios_dm.portfolio_id] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.portfolio_id, clients_portfolios_dm.client_id, clients_portfolios_dm.effective_start_date ] }
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 18282 (Actor 159374,159373)
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 18283 (Actor 159375,159376)
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 18284 (Actor 159378,159377)
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:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, party_items_mv_next.null:Int32#1, 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:Varchar#1, null:Varchar#2, null:Date, null:Int32#1, $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:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, party_items_mv_next.null:Int32#1, 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:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, party_items_mv_next.null:Int32#1, 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:Varchar#1, null:Varchar#2, null:Date, null:Int32#1, $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:Varchar#1, null:Varchar#2, null:Date, null:Int32#1, $src ], stream key: [] }

Fragment 18285 (Actor 159379,159380)
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 18286 (Actor 159383,159384)
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 18287 (Actor 159382,159381)
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 18288 (Actor 159386,159385)
StreamLocalityProvider { locality_columns: [account_to_portfolios_dm.account_id] } { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }
└── MergeExecutor { output: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ], stream key: [ account_to_portfolios_dm.account_id, account_to_portfolios_dm.portfolio_id, account_to_portfolios_dm.effective_start_date ] }

Fragment 18289 (Actor 159388,159387)
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 18290 (Actor 159390,159389)
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 18291 (Actor 159391,159392)
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 18292 (Actor 159393,159394)
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 18293 (Actor 159395,159396)
StreamLocalityProvider { locality_columns: [clients_portfolios_dm.client_id] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 18294 (Actor 159397,159398)
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 18295 (Actor 159399,159400)
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 18296 (Actor 159404,159403)
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 18297 (Actor 159401,159402)
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 18298 (Actor 159405,159406)
StreamLocalityProvider { locality_columns: [clients_portfolios_dm.client_id] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 18299 (Actor 159408,159407)
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 18300 (Actor 159409,159410)
StreamLocalityProvider { locality_columns: [clients_contacts_dm.client_id] } { 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 ] }
└── 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 18301 (Actor 159412,159411)
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 18302 (Actor 159413,159414)
StreamProject { exprs: [assets_dm.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.type) as $expr1, assets_dm.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.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.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.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.id, assets_dm.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.id, _rw_projected_row_id ]
    └── StreamProject { exprs: [assets_dm.id, assets_dm.name_en, assets_dm.type, assets_dm.ticker, assets_dm.isin, assets_dm.cusip, assets_dm.sedol, assets_dm.ric, assets_dm.figi] } { output: [ assets_dm.id, assets_dm.name_en, assets_dm.type, assets_dm.ticker, assets_dm.isin, assets_dm.cusip, assets_dm.sedol, assets_dm.ric, assets_dm.figi ], stream key: [ assets_dm.id ] }
        └── StreamFilter { predicate: IsNull(assets_dm.disabled_at) } { output: [ assets_dm.id, assets_dm.name_en, assets_dm.type, assets_dm.ticker, assets_dm.isin, assets_dm.cusip, assets_dm.sedol, assets_dm.ric, assets_dm.figi, assets_dm.disabled_at ], stream key: [ assets_dm.id ] }
            └── StreamTableScan { table: assets_dm, columns: [id, name_en, type, ticker, isin, cusip, sedol, ric, figi, disabled_at] } { output: [ assets_dm.id, assets_dm.name_en, assets_dm.type, assets_dm.ticker, assets_dm.isin, assets_dm.cusip, assets_dm.sedol, assets_dm.ric, assets_dm.figi, assets_dm.disabled_at ], stream key: [ assets_dm.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 18303 (Actor 159416,159415)
StreamProject { exprs: [assets_dm.id, 'asset':Varchar, null:Varchar, assets_dm.name_ar, Lower(assets_dm.type) as $expr2, assets_dm.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.id, 'asset':Varchar, null:Varchar, assets_dm.name_ar, $expr2, assets_dm.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.id ] }
└── StreamFilter { predicate: IsNull(assets_dm.disabled_at) } { output: [ assets_dm.id, assets_dm.name_ar, assets_dm.type, assets_dm.disabled_at ], stream key: [ assets_dm.id ] }
    └── StreamTableScan { table: assets_dm, columns: [id, name_ar, type, disabled_at] } { output: [ assets_dm.id, assets_dm.name_ar, assets_dm.type, assets_dm.disabled_at ], stream key: [ assets_dm.id ] }
        ├── Upstream { output: [ id, name_ar, type, disabled_at ], stream key: [] }
        └── BatchPlanNode { output: [ id, name_ar, type, disabled_at ], stream key: [] }

Fragment 18304 (Actor 159346,159345)
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 18305 (Actor 159418,159417)
StreamProject { exprs: [clients_dm.id, 'client':Varchar, null:Varchar, null:Varchar, ConcatOp('status:':Varchar, Lower(labels_dm.name_en)) as $expr4, clients_dm.status_label_id, clients_dm.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.status_label_id, clients_dm.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.status_label_id, clients_dm.id ] }
└── MergeExecutor { output: [ clients_dm.id, labels_dm.name_en, clients_dm.status_label_id, labels_dm.label_id ], stream key: [ clients_dm.status_label_id, clients_dm.id ] }

Fragment 18306 (Actor 159420,159419)
StreamSyncLogStore { output: [ clients_dm.id, labels_dm.name_en, clients_dm.status_label_id, labels_dm.label_id ], stream key: [ clients_dm.status_label_id, clients_dm.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.status_label_id, clients_dm.id ] }
    ├── MergeExecutor { output: [ clients_dm.id, clients_dm.status_label_id ], stream key: [ clients_dm.status_label_id, clients_dm.id ] }
    └── MergeExecutor { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }

Fragment 18307 (Actor 159421,159422)
StreamLocalityProvider { locality_columns: [clients_dm.status_label_id] } { output: [ clients_dm.id, clients_dm.status_label_id ], stream key: [ clients_dm.status_label_id, clients_dm.id ] }
└── MergeExecutor { output: [ clients_dm.id, clients_dm.status_label_id ], stream key: [ clients_dm.id ] }

Fragment 18308 (Actor 159423,159424)
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 18309 (Actor 159426,159425)
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 18310 (Actor 159427,159428)
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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 18311 (Actor 159430,159429)
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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_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 18312 (Actor 159431,159432)
StreamLocalityProvider { locality_columns: [entity_to_teams_dm.entity_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 ], stream key: [ entity_to_teams_dm.entity_id, entity_to_teams_dm.team_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 ] }

Fragment 18313 (Actor 159433,159434)
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 18314 (Actor 159435,159436)
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 18315 (Actor 157089,157088)
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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_id ] }

Fragment 18316 (Actor 157091,157090)
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.product_type_id, accounts_dm.account_id ] }
├── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.product_type_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.product_type_id, 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 18317 (Actor 159437,159438)
StreamLocalityProvider { locality_columns: [accounts_dm.product_type_id] } { output: [ accounts_dm.account_id, accounts_dm.product_type_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.product_type_id, accounts_dm.account_id ] }
└── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.product_type_id, accounts_dm.disabled_at ], stream key: [ accounts_dm.account_id ] }

Fragment 18318 (Actor 159439,159440)
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 18319 (Actor 157080,157081)
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 18320 (Actor 157078,157079)
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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_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.product_type_id, accounts_dm.account_id ] }

Fragment 18321 (Actor 159447,159448)
StreamProject { exprs: [accounts_dm.account_id, 'account':Varchar, null:Varchar, null:Varchar, ConcatOp('status:':Varchar, Lower(labels_dm.name_en)) as $expr8, accounts_dm.status_label_id, accounts_dm.account_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.status_label_id, accounts_dm.account_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.status_label_id, accounts_dm.account_id ] }
└── MergeExecutor { output: [ accounts_dm.account_id, labels_dm.name_en, accounts_dm.status_label_id, labels_dm.label_id ], stream key: [ accounts_dm.status_label_id, accounts_dm.account_id ] }

Fragment 18322 (Actor 159445,159446)
StreamSyncLogStore { output: [ accounts_dm.account_id, labels_dm.name_en, accounts_dm.status_label_id, labels_dm.label_id ], stream key: [ accounts_dm.status_label_id, accounts_dm.account_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.status_label_id, accounts_dm.account_id ] }
    ├── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.status_label_id ], stream key: [ accounts_dm.status_label_id, accounts_dm.account_id ] }
    └── MergeExecutor { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }

Fragment 18323 (Actor 159449,159450)
StreamLocalityProvider { locality_columns: [accounts_dm.status_label_id] } { output: [ accounts_dm.account_id, accounts_dm.status_label_id ], stream key: [ accounts_dm.status_label_id, accounts_dm.account_id ] }
└── MergeExecutor { output: [ accounts_dm.account_id, accounts_dm.status_label_id ], stream key: [ accounts_dm.account_id ] }

Fragment 18324 (Actor 159451,159452)
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 18325 (Actor 159454,159453)
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 18326 (Actor 159460,159459)
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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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 18327 (Actor 159461,159462)
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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.client_id, accounts_to_clients_dm.account_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.entity_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 18328 (Actor 159463,159464)
StreamLocalityProvider { locality_columns: [accounts_to_clients_dm.client_id] } { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.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.client_id, accounts_to_clients_dm.account_id, accounts_to_clients_dm.effective_start_date ] }
└── MergeExecutor { output: [ accounts_to_clients_dm.account_id, accounts_to_clients_dm.client_id, accounts_to_clients_dm.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 18329 (Actor 159359,159360)
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 18330 (Actor 159465,159466)
StreamLocalityProvider { locality_columns: [entity_to_teams_dm.entity_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 ], stream key: [ entity_to_teams_dm.entity_id, entity_to_teams_dm.team_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 ] }

Fragment 18331 (Actor 159467,159468)
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 18332 (Actor 156964,156965)
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.service_type_id, portfolios_dm.portfolio_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.service_type_id, portfolios_dm.portfolio_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.service_type_id, portfolios_dm.portfolio_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.service_type_id, portfolios_dm.portfolio_id ] }
    ├── MergeExecutor { output: [ portfolios_dm.portfolio_id, portfolios_dm.service_type_id ], stream key: [ portfolios_dm.service_type_id, 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 18333 (Actor 159469,159470)
StreamLocalityProvider { locality_columns: [portfolios_dm.service_type_id] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.service_type_id ], stream key: [ portfolios_dm.service_type_id, portfolios_dm.portfolio_id ] }
└── MergeExecutor { output: [ portfolios_dm.portfolio_id, portfolios_dm.service_type_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 18334 (Actor 159473,159474)
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 18335 (Actor 156962,156963)
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 18336 (Actor 159478,159477)
StreamProject { exprs: [portfolios_dm.portfolio_id, 'portfolio':Varchar, null:Varchar, null:Varchar, ConcatOp('status:':Varchar, Lower(labels_dm.name_en)) as $expr11, portfolios_dm.status_label_id, portfolios_dm.portfolio_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.status_label_id, portfolios_dm.portfolio_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.status_label_id, portfolios_dm.portfolio_id ] }
└── MergeExecutor { output: [ portfolios_dm.portfolio_id, labels_dm.name_en, portfolios_dm.status_label_id, labels_dm.label_id ], stream key: [ portfolios_dm.status_label_id, portfolios_dm.portfolio_id ] }

Fragment 18337 (Actor 159475,159476)
StreamSyncLogStore { output: [ portfolios_dm.portfolio_id, labels_dm.name_en, portfolios_dm.status_label_id, labels_dm.label_id ], stream key: [ portfolios_dm.status_label_id, portfolios_dm.portfolio_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.status_label_id, portfolios_dm.portfolio_id ] }
    ├── MergeExecutor { output: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ], stream key: [ portfolios_dm.status_label_id, portfolios_dm.portfolio_id ] }
    └── MergeExecutor { output: [ labels_dm.label_id, labels_dm.name_en ], stream key: [ labels_dm.label_id ] }

Fragment 18338 (Actor 159480,159479)
StreamLocalityProvider { locality_columns: [portfolios_dm.status_label_id] } { output: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ], stream key: [ portfolios_dm.status_label_id, portfolios_dm.portfolio_id ] }
└── MergeExecutor { output: [ portfolios_dm.portfolio_id, portfolios_dm.status_label_id ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 18339 (Actor 159481,159482)
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 18340 (Actor 159483,159484)
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 18341 (Actor 159487,159488)
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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 18342 (Actor 159486,159485)
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.entity_id, entity_to_teams_dm.team_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.entity_id, entity_to_teams_dm.team_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.entity_id, 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 ], stream key: [ portfolios_dm.portfolio_id ] }

Fragment 18343 (Actor 159490,159489)
StreamLocalityProvider { locality_columns: [entity_to_teams_dm.entity_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 ], stream key: [ entity_to_teams_dm.entity_id, entity_to_teams_dm.team_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 ] }

Fragment 18344 (Actor 159492,159491)
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 18345 (Actor 159493,159494)
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 18346 (Actor 159498,159497)
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 18347 (Actor 159496,159495)
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.entity_id, entity_to_teams_dm.team_id, entity_to_teams_dm.entity_type, entity_to_teams_dm.effective_start_date ] }

Fragment 18348 (Actor 159500,159499)
StreamLocalityProvider { locality_columns: [clients_portfolios_dm.client_id] } { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }
└── MergeExecutor { output: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date ] }

Fragment 18349 (Actor 159502,159501)
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 18350 (Actor 159504,159503)
StreamLocalityProvider { locality_columns: [entity_to_teams_dm.entity_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 ], stream key: [ entity_to_teams_dm.entity_id, entity_to_teams_dm.team_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 ] }

Fragment 18351 (Actor 159505,159506)
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 18352 (Actor 159510,159509)
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.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, 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.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, 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.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, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]
└── 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.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, olap_reference_identifier_terms_mv.owner_entity_id, olap_reference_identifier_terms_mv.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.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, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]

Fragment 18353 (Actor 159507,159508)
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.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, olap_reference_identifier_terms_mv.owner_entity_id, olap_reference_identifier_terms_mv.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.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, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]
└── 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.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, olap_reference_identifier_terms_mv.owner_entity_id, olap_reference_identifier_terms_mv.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.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, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]
    ├── 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.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 ]
    └── 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.owner_entity_id, olap_reference_identifier_terms_mv.owner_entity_type, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ] }

Fragment 18354 (Actor 159511,159512)
StreamLocalityProvider { locality_columns: [active_identifier_edges_mv_next.owner_entity_id, active_identifier_edges_mv_next.owner_entity_type] }
├── 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.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 ]
└── 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 ] }

Fragment 18355 (Actor 159513,159514)
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 18356 (Actor 159516,159515)
StreamLocalityProvider { locality_columns: [olap_reference_identifier_terms_mv.owner_entity_id, olap_reference_identifier_terms_mv.owner_entity_type] } { 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.owner_entity_id, olap_reference_identifier_terms_mv.owner_entity_type, olap_reference_identifier_terms_mv.reference_identifiers.id, olap_reference_identifier_terms_mv._rw_projected_row_id ] }
└── 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 18357 (Actor 159518,159517)
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 18358 (Actor 159519,159520)
StreamProject { exprs: [party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv.val, party_reference_identifier_terms_mv.ar_val, null:Varchar, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, null:Varchar, null:Date, null:Date, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._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.val, party_reference_identifier_terms_mv.ar_val, null:Varchar, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, null:Varchar, null:Date, null:Date, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._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.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ]
└── MergeExecutor
    ├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv.val, party_reference_identifier_terms_mv.ar_val, 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, party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ]
    └── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ]

Fragment 18359 (Actor 159522,159521)
StreamSyncLogStore
├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv.val, party_reference_identifier_terms_mv.ar_val, 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, party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ]
├── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ]
└── StreamHashJoin { type: Inner, predicate: party_identifier_edges_mv_next.owner_entity_id = party_reference_identifier_terms_mv.owner_entity_id AND party_identifier_edges_mv_next.owner_entity_type = party_reference_identifier_terms_mv.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.val, party_reference_identifier_terms_mv.ar_val, 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, party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ]
    ├── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ]
    ├── 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.owner_entity_type, 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.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.val, party_reference_identifier_terms_mv.ar_val, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ] }

Fragment 18360 (Actor 159524,159523)
StreamLocalityProvider { locality_columns: [party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type] } { 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.owner_entity_type, party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src ] }
└── 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.target_entity_id, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src ] }

Fragment 18361 (Actor 159526,159525)
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.target_entity_id, party_identifier_edges_mv_next.owner_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 18362 (Actor 159527,159528)
StreamLocalityProvider { locality_columns: [party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type] } { output: [ party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.val, party_reference_identifier_terms_mv.ar_val, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ] }
└── MergeExecutor { output: [ party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.val, party_reference_identifier_terms_mv.ar_val, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ] }

Fragment 18363 (Actor 159529,159530)
StreamTableScan { table: party_reference_identifier_terms_mv, columns: [owner_entity_id, owner_entity_type, val, ar_val, reference_identifiers_next.id, _rw_projected_row_id] } { output: [ party_reference_identifier_terms_mv.owner_entity_id, party_reference_identifier_terms_mv.owner_entity_type, party_reference_identifier_terms_mv.val, party_reference_identifier_terms_mv.ar_val, party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv.reference_identifiers_next.id, party_reference_identifier_terms_mv._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-ai-hub-int.ai-hub-rwm.svc.cluster.local