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

Fragment 63636 (Actor 746635,746636)
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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Int32, null:Int32, null:Int32, null:Date, $src ] }

Fragment 63637 (Actor 746637,746638)
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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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, 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, 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, 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, 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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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, 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, 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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar#3, party_items_mv_next.null:Varchar#4, 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.null:Int32#2, 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:Varchar#3, party_items_mv_next.null:Varchar#4, party_items_mv_next.null:Int32#1, party_items_mv_next.null:Int32#2, party_items_mv_next.$src ]
├── MergeExecutor { output: [ portfolios_dm.portfolio_id, 'portfolio':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $2, Replace($2, '-':Varchar, '':Varchar))), null:Varchar, portfolios_dm.portfolio_id, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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, 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, 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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, null:Int32, 11:Int32 ], stream key: [ clients_portfolios_dm.client_id, clients_portfolios_dm.portfolio_id, clients_portfolios_dm.effective_start_date, clients_contacts_dm.type, clients_contacts_dm.value, _rw_projected_row_id ] }
├── MergeExecutor { output: [ assets_dm_next.id, 'asset':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar))), null:Varchar, $expr1, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, null:Int32, 12:Int32 ], stream key: [ assets_dm_next.id, _rw_projected_row_id ] }
├── MergeExecutor { output: [ assets_dm_next.id, 'asset':Varchar, null:Varchar, assets_dm_next.name_ar, $expr2, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, null:Int32, null:Int32, null:Int32, 13:Int32 ], stream key: [ assets_dm_next.id ] }
├── MergeExecutor { output: [ clients_dm.id, 'client':Varchar, null:Varchar, null:Varchar, $expr3, clients_dm.id, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, 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:Varchar#1
│   │   ├── active_identifier_edges_mv_next.null:Varchar#2
│   │   ├── active_identifier_edges_mv_next.null:Date
│   │   ├── null:Date
│   │   ├── olap_reference_identifier_terms_mv.reference_identifiers_next.id
│   │   ├── olap_reference_identifier_terms_mv._rw_projected_row_id
│   │   ├── active_identifier_edges_mv_next.null:Int32
│   │   ├── active_identifier_edges_mv_next.null:Int32#1
│   │   ├── 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers_next.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_next.val
    │   ├── party_reference_identifier_terms_mv_next.ar_val
    │   ├── null:Varchar
    │   ├── party_identifier_edges_mv_next.owner_entity_id
    │   ├── party_identifier_edges_mv_next.owner_entity_type
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id
    │   ├── null:Date
    │   ├── null:Date
    │   ├── party_reference_identifier_terms_mv_next.reference_identifiers_next.id
    │   ├── party_reference_identifier_terms_mv_next._rw_projected_row_id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.$src
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1
    │   ├── 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.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ]

Fragment 63638 (Actor 746640,746639)
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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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 63639 (Actor 746643,746644)
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, 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, 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, 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, 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 63640 (Actor 746642,746641)
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 63641 (Actor 746665,746666)
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 63642 (Actor 746667,746668)
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 63643 (Actor 746670,746669)
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 63644 (Actor 746674,746673)
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, 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, 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, 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, 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 63645 (Actor 746671,746672)
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 63646 (Actor 746675,746676)
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 63647 (Actor 746678,746677)
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 63648 (Actor 746679,746680)
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 63649 (Actor 746681,746682)
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 63650 (Actor 746685,746686)
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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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 63651 (Actor 746684,746683)
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 63652 (Actor 746690,746689)
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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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 63653 (Actor 746695,746696)
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, 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, 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, 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, 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 63654 (Actor 746697,746698)
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 63655 (Actor 746694,746693)
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 63656 (Actor 746700,746699)
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 63657 (Actor 746701,746702)
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 63658 (Actor 746706,746705)
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 63659 (Actor 746710,746709)
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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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 63660 (Actor 746708,746707)
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 63661 (Actor 746712,746711)
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 63662 (Actor 746714,746713)
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 63663 (Actor 746716,746715)
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 63664 (Actor 746717,746718)
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:Varchar#3, party_items_mv_next.null:Varchar#4, 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.null:Int32#2, 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:Varchar#3, party_items_mv_next.null:Varchar#4, 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.null:Int32#2, 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:Varchar#3, party_items_mv_next.null:Varchar#4, party_items_mv_next.null:Int32#1, party_items_mv_next.null:Int32#2, party_items_mv_next.$src ]
└── StreamTableScan { table: party_items_mv_next, columns: [entity_id, entity_type, val, ar_val, filter_val, active_parties_mv.id, _rw_projected_row_id, null:Varchar, null:Int32, null:Int64, null:Varchar#1, null:Varchar#2, null:Date, null:Varchar#3, null:Varchar#4, null:Int32#1, null:Int32#2, $src] }
    ├── output: [ party_items_mv_next.entity_id, party_items_mv_next.entity_type, party_items_mv_next.val, party_items_mv_next.ar_val, party_items_mv_next.filter_val, party_items_mv_next.active_parties_mv.id, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Int32, party_items_mv_next.null:Int64, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, party_items_mv_next.null:Varchar#3, party_items_mv_next.null:Varchar#4, party_items_mv_next.null:Int32#1, party_items_mv_next.null:Int32#2, party_items_mv_next.$src ]
    ├── stream key: [ party_items_mv_next.active_parties_mv.id, party_items_mv_next._rw_projected_row_id, party_items_mv_next.null:Varchar, party_items_mv_next.null:Int32, party_items_mv_next.null:Int64, party_items_mv_next.null:Varchar#1, party_items_mv_next.null:Varchar#2, party_items_mv_next.null:Date, party_items_mv_next.null:Varchar#3, party_items_mv_next.null:Varchar#4, party_items_mv_next.null:Int32#1, party_items_mv_next.null:Int32#2, party_items_mv_next.$src ]
    ├── Upstream { output: [ entity_id, entity_type, val, ar_val, filter_val, active_parties_mv.id, _rw_projected_row_id, null:Varchar, null:Int32, null:Int64, null:Varchar#1, null:Varchar#2, null:Date, null:Varchar#3, null:Varchar#4, null:Int32#1, null:Int32#2, $src ], stream key: [] }
    └── BatchPlanNode { output: [ entity_id, entity_type, val, ar_val, filter_val, active_parties_mv.id, _rw_projected_row_id, null:Varchar, null:Int32, null:Int64, null:Varchar#1, null:Varchar#2, null:Date, null:Varchar#3, null:Varchar#4, null:Int32#1, null:Int32#2, $src ], stream key: [] }

Fragment 63665 (Actor 746719,746720)
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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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 63666 (Actor 746725,746726)
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, 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, 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, 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, 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 63667 (Actor 746723,746724)
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 63668 (Actor 746728,746727)
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 63669 (Actor 746730,746729)
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 63670 (Actor 746732,746731)
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 63671 (Actor 746735,746736)
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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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 63672 (Actor 746737,746738)
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 63673 (Actor 746739,746740)
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 63674 (Actor 746742,746741)
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 63675 (Actor 746744,746743)
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 63676 (Actor 746747,746748)
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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, 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 63677 (Actor 746745,746746)
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 63678 (Actor 746750,746749)
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 63679 (Actor 746751,746752)
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 63680 (Actor 746754,746753)
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 63681 (Actor 746755,746756)
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 63682 (Actor 746757,746758)
StreamProject { exprs: [assets_dm_next.id, 'asset':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar))), null:Varchar, Lower(assets_dm_next.type) as $expr1, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, null:Int32, 12:Int32] }
├── output: [ assets_dm_next.id, 'asset':Varchar, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar))), null:Varchar, $expr1, assets_dm_next.id, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Varchar, null:Date, null:Date, _rw_projected_row_id, null:Int64, null:Int32, null:Int32, null:Int32, null:Int32, 12:Int32 ]
├── stream key: [ assets_dm_next.id, _rw_projected_row_id ]
└── StreamProjectSet { select_list: [$0, $2, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar)))] }
    ├── output: [ _rw_projected_row_id, assets_dm_next.id, assets_dm_next.type, Unnest(Array($1, Replace($1, '-':Varchar, '':Varchar), $3, Replace($3, '-':Varchar, '':Varchar), $4, Replace($4, '-':Varchar, '':Varchar), $7, Replace($7, '-':Varchar, '':Varchar), $5, Replace($5, '-':Varchar, '':Varchar), $6, Replace($6, '-':Varchar, '':Varchar), $8, Replace($8, '-':Varchar, '':Varchar))) ]
    ├── stream key: [ assets_dm_next.id, _rw_projected_row_id ]
    └── StreamProject { exprs: [assets_dm_next.id, assets_dm_next.name_en, assets_dm_next.type, assets_dm_next.ticker, assets_dm_next.isin, assets_dm_next.cusip, assets_dm_next.sedol, assets_dm_next.ric, assets_dm_next.figi] } { output: [ assets_dm_next.id, assets_dm_next.name_en, assets_dm_next.type, assets_dm_next.ticker, assets_dm_next.isin, assets_dm_next.cusip, assets_dm_next.sedol, assets_dm_next.ric, assets_dm_next.figi ], stream key: [ assets_dm_next.id ] }
        └── StreamFilter { predicate: IsNull(assets_dm_next.disabled_at) } { output: [ assets_dm_next.id, assets_dm_next.name_en, assets_dm_next.type, assets_dm_next.ticker, assets_dm_next.isin, assets_dm_next.cusip, assets_dm_next.sedol, assets_dm_next.ric, assets_dm_next.figi, assets_dm_next.disabled_at ], stream key: [ assets_dm_next.id ] }
            └── StreamTableScan { table: assets_dm_next, columns: [id, name_en, type, ticker, isin, cusip, sedol, ric, figi, disabled_at] } { output: [ assets_dm_next.id, assets_dm_next.name_en, assets_dm_next.type, assets_dm_next.ticker, assets_dm_next.isin, assets_dm_next.cusip, assets_dm_next.sedol, assets_dm_next.ric, assets_dm_next.figi, assets_dm_next.disabled_at ], stream key: [ assets_dm_next.id ] }
                ├── Upstream { output: [ id, name_en, type, ticker, isin, cusip, sedol, ric, figi, disabled_at ], stream key: [] }
                └── BatchPlanNode { output: [ id, name_en, type, ticker, isin, cusip, sedol, ric, figi, disabled_at ], stream key: [] }

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

Fragment 63684 (Actor 746687,746688)
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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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 63685 (Actor 746763,746764)
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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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 63686 (Actor 746761,746762)
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 63687 (Actor 746765,746766)
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 63688 (Actor 746768,746767)
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 63689 (Actor 746769,746770)
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 63690 (Actor 746773,746774)
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, null:Varchar, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, 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 63691 (Actor 746771,746772)
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 63692 (Actor 746775,746776)
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 63693 (Actor 746778,746777)
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 63694 (Actor 746780,746779)
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 63695 (Actor 742336,742337)
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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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 63696 (Actor 742328,742329)
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 63697 (Actor 746782,746781)
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 63698 (Actor 746784,746783)
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 63699 (Actor 742330,742331)
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 63700 (Actor 742335,742334)
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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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 63701 (Actor 746786,746785)
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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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 63702 (Actor 746788,746787)
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 63703 (Actor 746789,746790)
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 63704 (Actor 746791,746792)
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 63705 (Actor 746793,746794)
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 63706 (Actor 746797,746798)
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, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, accounts_to_clients_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, 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 63707 (Actor 746799,746800)
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 63708 (Actor 746801,746802)
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 63709 (Actor 746692,746691)
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 63710 (Actor 746803,746804)
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 63711 (Actor 741895,741896)
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 63712 (Actor 742420,742421)
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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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 63713 (Actor 741906,741907)
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 63714 (Actor 741912,741911)
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 63715 (Actor 742423,742422)
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 63716 (Actor 742009,742010)
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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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:Varchar, null:Varchar, null:Date, null:Date, null:Int64, null:Int64, null:Int32, 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 63717 (Actor 742008,742007)
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 63718 (Actor 742022,742021)
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 63719 (Actor 742101,742102)
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 63720 (Actor 742116,742115)
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 63721 (Actor 742410,742411)
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, null:Varchar, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, entity_to_teams_dm.effective_start_date, null:Date, null:Int64, null:Int64, null:Int32, 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 63722 (Actor 742412,742413)
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 63723 (Actor 742467,742468)
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 63724 (Actor 742850,742851)
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 63725 (Actor 742982,742981)
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 63726 (Actor 743057,743058)
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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, 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, null:Varchar, null:Varchar, clients_portfolios_dm.effective_start_date, entity_to_teams_dm.effective_start_date, null:Int64, null:Int64, null:Int32, 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 63727 (Actor 743055,743056)
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 63728 (Actor 743060,743059)
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 63729 (Actor 743211,743212)
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 63730 (Actor 743226,743225)
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 63731 (Actor 743360,743359)
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 63732 (Actor 743901,743902)
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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Date, null:Date, olap_reference_identifier_terms_mv.reference_identifiers_next.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Date, null:Date, olap_reference_identifier_terms_mv.reference_identifiers_next.id, olap_reference_identifier_terms_mv._rw_projected_row_id, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers_next.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:Varchar#1
    │   ├── active_identifier_edges_mv_next.null:Varchar#2
    │   ├── active_identifier_edges_mv_next.null:Int32
    │   ├── active_identifier_edges_mv_next.null:Int32#1
    │   ├── 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_next.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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers_next.id, olap_reference_identifier_terms_mv._rw_projected_row_id ]

Fragment 63733 (Actor 743904,743903)
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:Varchar#1
│   ├── active_identifier_edges_mv_next.null:Varchar#2
│   ├── active_identifier_edges_mv_next.null:Int32
│   ├── active_identifier_edges_mv_next.null:Int32#1
│   ├── 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_next.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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers_next.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:Varchar#1
    │   ├── active_identifier_edges_mv_next.null:Varchar#2
    │   ├── active_identifier_edges_mv_next.null:Int32
    │   ├── active_identifier_edges_mv_next.null:Int32#1
    │   ├── 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_next.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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, active_identifier_edges_mv_next.$src, olap_reference_identifier_terms_mv.reference_identifiers_next.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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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_next.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_next.id, olap_reference_identifier_terms_mv._rw_projected_row_id ] }

Fragment 63734 (Actor 743926,743925)
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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, active_identifier_edges_mv_next.$src ]

Fragment 63735 (Actor 743935,743936)
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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, null:Varchar#2, null:Int32, null:Int32#1, $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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, active_identifier_edges_mv_next.null:Varchar#2, active_identifier_edges_mv_next.null:Int32, active_identifier_edges_mv_next.null:Int32#1, 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:Varchar#1, null:Varchar#2, null:Int32, null:Int32#1, $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:Varchar#1, null:Varchar#2, null:Int32, null:Int32#1, $src ], stream key: [] }

Fragment 63736 (Actor 743958,743957)
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_next.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_next.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_next.id, olap_reference_identifier_terms_mv._rw_projected_row_id ], stream key: [ olap_reference_identifier_terms_mv.reference_identifiers_next.id, olap_reference_identifier_terms_mv._rw_projected_row_id ] }

Fragment 63737 (Actor 743992,743991)
StreamTableScan { table: olap_reference_identifier_terms_mv, columns: [owner_entity_id, owner_entity_type, val, ar_val, reference_identifiers_next.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_next.id, olap_reference_identifier_terms_mv._rw_projected_row_id ], stream key: [ olap_reference_identifier_terms_mv.reference_identifiers_next.id, olap_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: [] }

Fragment 63738 (Actor 744197,744198)
StreamProject { exprs: [party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, null:Varchar, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, null:Date, null:Date, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1, null:Int32, null:Int32, 26:Int32] }
├── output: [ party_identifier_edges_mv_next.target_entity_id, party_identifier_edges_mv_next.target_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, null:Varchar, party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, null:Date, null:Date, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1, 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.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._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_next.val
    │   ├── party_reference_identifier_terms_mv_next.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.accounts_dm.account_id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.$src
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1
    │   ├── party_reference_identifier_terms_mv_next.owner_entity_id
    │   ├── party_reference_identifier_terms_mv_next.owner_entity_type
    │   ├── party_reference_identifier_terms_mv_next.reference_identifiers_next.id
    │   └── party_reference_identifier_terms_mv_next._rw_projected_row_id
    └── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ]

Fragment 63739 (Actor 744200,744199)
StreamSyncLogStore
├── output:
│   ┌── party_identifier_edges_mv_next.target_entity_id
│   ├── party_identifier_edges_mv_next.target_entity_type
│   ├── party_reference_identifier_terms_mv_next.val
│   ├── party_reference_identifier_terms_mv_next.ar_val
│   ├── party_identifier_edges_mv_next.owner_entity_id
│   ├── party_identifier_edges_mv_next.owner_entity_type
│   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id
│   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id
│   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id
│   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id
│   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.$src
│   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1
│   ├── party_reference_identifier_terms_mv_next.owner_entity_id
│   ├── party_reference_identifier_terms_mv_next.owner_entity_type
│   ├── party_reference_identifier_terms_mv_next.reference_identifiers_next.id
│   └── party_reference_identifier_terms_mv_next._rw_projected_row_id
├── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ]
└── StreamHashJoin { type: Inner, predicate: party_identifier_edges_mv_next.owner_entity_id = party_reference_identifier_terms_mv_next.owner_entity_id AND party_identifier_edges_mv_next.owner_entity_type = party_reference_identifier_terms_mv_next.owner_entity_type }
    ├── output:
    │   ┌── party_identifier_edges_mv_next.target_entity_id
    │   ├── party_identifier_edges_mv_next.target_entity_type
    │   ├── party_reference_identifier_terms_mv_next.val
    │   ├── party_reference_identifier_terms_mv_next.ar_val
    │   ├── party_identifier_edges_mv_next.owner_entity_id
    │   ├── party_identifier_edges_mv_next.owner_entity_type
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.$src
    │   ├── party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1
    │   ├── party_reference_identifier_terms_mv_next.owner_entity_id
    │   ├── party_reference_identifier_terms_mv_next.owner_entity_type
    │   ├── party_reference_identifier_terms_mv_next.reference_identifiers_next.id
    │   └── party_reference_identifier_terms_mv_next._rw_projected_row_id
    ├── stream key: [ party_identifier_edges_mv_next.owner_entity_id, party_identifier_edges_mv_next.owner_entity_type, party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._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.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1 ]
    │   └── stream key: [ 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.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1 ]
    └── MergeExecutor { output: [ party_reference_identifier_terms_mv_next.owner_entity_id, party_reference_identifier_terms_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv_next.owner_entity_id, party_reference_identifier_terms_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ] }

Fragment 63740 (Actor 744202,744201)
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.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1 ]
├── stream key: [ 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.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1 ]
└── 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.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1 ]
    └── stream key: [ party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1 ]

Fragment 63741 (Actor 744214,744213)
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.accounts_dm.account_id, party_holder_edges_mv_next.customer_relationships_next.id, party_holder_edges_mv_next.active_parties_mv.id, party_holder_edges_mv_next.party_involvements_dm.id, party_holder_edges_mv_next.$src, party_holder_edges_mv_next.$src#1] }
├── 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.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1 ]
├── stream key: [ party_identifier_edges_mv_next.party_holder_edges_mv_next.accounts_dm.account_id, party_identifier_edges_mv_next.party_holder_edges_mv_next.customer_relationships_next.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.active_parties_mv.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.party_involvements_dm.id, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src, party_identifier_edges_mv_next.party_holder_edges_mv_next.$src#1 ]
├── Upstream { output: [ target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, party_holder_edges_mv_next.accounts_dm.account_id, party_holder_edges_mv_next.customer_relationships_next.id, party_holder_edges_mv_next.active_parties_mv.id, party_holder_edges_mv_next.party_involvements_dm.id, party_holder_edges_mv_next.$src, party_holder_edges_mv_next.$src#1 ], stream key: [] }
└── BatchPlanNode { output: [ target_entity_id, target_entity_type, owner_entity_id, owner_entity_type, party_holder_edges_mv_next.accounts_dm.account_id, party_holder_edges_mv_next.customer_relationships_next.id, party_holder_edges_mv_next.active_parties_mv.id, party_holder_edges_mv_next.party_involvements_dm.id, party_holder_edges_mv_next.$src, party_holder_edges_mv_next.$src#1 ], stream key: [] }

Fragment 63742 (Actor 744218,744217)
StreamLocalityProvider { locality_columns: [party_reference_identifier_terms_mv_next.owner_entity_id, party_reference_identifier_terms_mv_next.owner_entity_type] } { output: [ party_reference_identifier_terms_mv_next.owner_entity_id, party_reference_identifier_terms_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv_next.owner_entity_id, party_reference_identifier_terms_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ] }
└── MergeExecutor { output: [ party_reference_identifier_terms_mv_next.owner_entity_id, party_reference_identifier_terms_mv_next.owner_entity_type, party_reference_identifier_terms_mv_next.val, party_reference_identifier_terms_mv_next.ar_val, party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ], stream key: [ party_reference_identifier_terms_mv_next.reference_identifiers_next.id, party_reference_identifier_terms_mv_next._rw_projected_row_id ] }

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