accruals_agg_mv · explain

← cluster insights objects accruals_agg_mv explain

materialized view · insights.accruals_agg_mv profiled over 5s

Job is idle — throughput ~0; structure shown.

Stateful hash join (4 state tables) — consider a temporal join for dimension lookupsAggregation state — unbounded unless keyed or temporally filtered

Heat tree Operator graph

64 operators

Materialize · insights.accruals_agg_mv

0% idle 2 actors

Project

— — 2 actors

HashAgg Aggregation state — unbounded unless keyed or temporally filtered

0% idle 2 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · (Not(IsNull(foreign_exchange_rates_eod_ft.rate)) OR (accrua…

— — 2 actors

Filter · (Not(IsNull(foreign_exchange_rates_eod_ft.rate)) OR (accrua…

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · LeftOuter · accruals_ft.currency = foreign_exchange_rates_eod_ft.source…

— — 2 actors

HashJoin · LeftOuter · accruals_ft.currency = foreign_exchange_rates_eod_ft.source… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Filter · foreign_exchange_rates_eod_ft

0% idle 2 actors

StreamScan · foreign_exchange_rates_eod_ft

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Filter · (Not(IsNull(foreign_exchange_rates_eod_ft.rate)) OR (accrua…

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · LeftOuter · accruals_ft.currency = foreign_exchange_rates_eod_ft.source…

— — 2 actors

HashJoin · LeftOuter · accruals_ft.currency = foreign_exchange_rates_eod_ft.source… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · foreign_exchange_rates_eod_ft

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

SyncLogStore · LeftOuter · accruals_ft.fact_date = holding_values_raw_mv.dim_value_dat…

— — 2 actors

HashJoin · LeftOuter · accruals_ft.fact_date = holding_values_raw_mv.dim_value_dat… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

StreamScan · holding_values_raw_mv

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

LocalityProvider

0% idle 2 actors

Merge

— — 2 actors

Exchange

0% idle 0 actors

Project · accruals_ft

— — 2 actors

Filter · accruals_ft

0% idle 2 actors

StreamScan · accruals_ft

0% idle 2 actors

BatchPlan

— — 2 actors

Merge

— — 2 actors

Heat = the operator's output-buffer backpressure over the sampling window. Click a node to fold its subtree.

Streaming operator plan from EXPLAIN ANALYZE. Node heat = backpressure. Drag to pan, scroll to zoom.

Fragments (DESCRIBE FRAGMENTS) — click to expand

Fragment 62546 (Actor 743015,743016)
StreamMaterialize { columns: [account_id, asset_id, dim_value_date, currency_code, type, accrued_amount, accrued_value_system_currency], stream_key: [account_id, asset_id, dim_value_date, currency_code, type], pk_columns: [account_id, asset_id, dim_value_date, currency_code, type], pk_conflict: NoCheck }
├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4, sum($expr5), sum($expr6) ]
├── stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4 ]
└── StreamProject { exprs: [accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4, sum($expr5), sum($expr6)] }
    ├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4, sum($expr5), sum($expr6) ]
    ├── stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4 ]
    └── StreamHashAgg { group_key: [accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4], aggs: [sum($expr5), sum($expr6), count] }
        ├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4, sum($expr5), sum($expr6), count ]
        ├── stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4 ]
        └── StreamLocalityProvider { locality_columns: [accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4] }
            ├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4, $expr5, $expr6, accruals_ft.currency, $expr2, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ]
            ├── stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4, accruals_ft.currency, $expr2, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ]
            └── MergeExecutor
                ├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4, $expr5, $expr6, accruals_ft.currency, $expr2, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ]
                └── stream key: [ accruals_ft.currency, accruals_ft.fact_date, $expr2, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ]

Fragment 62547 (Actor 743065,743066)
StreamProject { exprs: [accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, Coalesce(holding_values_raw_mv.currency_code, accruals_ft.currency) as $expr3, Case((accruals_ft.type = 'EXPENSE':Varchar), 'LIABILITY':Varchar, 'ASSET':Varchar) as $expr4, (accruals_ft.amount * Coalesce(foreign_exchange_rates_eod_ft.rate, Case((accruals_ft.currency = Coalesce(holding_values_raw_mv.currency_code, accruals_ft.currency)), 1:Int32, null:Int32)::Decimal)) as $expr5, (accruals_ft.amount * Coalesce(foreign_exchange_rates_eod_ft.rate, Case((accruals_ft.currency = 'SAR':Varchar), 1:Int32, null:Int32)::Decimal)) as $expr6, accruals_ft.currency, $expr2, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code] }
├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr3, $expr4, $expr5, $expr6, accruals_ft.currency, $expr2, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ]
├── stream key: [ accruals_ft.currency, accruals_ft.fact_date, $expr2, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ]
└── StreamFilter { predicate: (Not(IsNull(foreign_exchange_rates_eod_ft.rate)) OR (accruals_ft.currency = 'SAR':Varchar)) }
    ├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.target_currency_code ]
    ├── stream key: [ accruals_ft.currency, accruals_ft.fact_date, $expr2, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ]
    └── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.target_currency_code ], stream key: [ accruals_ft.currency, accruals_ft.fact_date, $expr2, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ] }

Fragment 62548 (Actor 743068,743067)
StreamSyncLogStore { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.target_currency_code ], stream key: [ accruals_ft.currency, accruals_ft.fact_date, $expr2, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ] }
└── StreamHashJoin { type: LeftOuter, predicate: accruals_ft.currency = foreign_exchange_rates_eod_ft.source_currency_code AND accruals_ft.fact_date = foreign_exchange_rates_eod_ft.date }
    ├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.target_currency_code ]
    ├── stream key: [ accruals_ft.currency, accruals_ft.fact_date, $expr2, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.target_currency_code ]
    ├── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ], stream key: [ accruals_ft.currency, accruals_ft.fact_date, $expr2, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
    └── MergeExecutor { output: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.target_currency_code ], stream key: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.target_currency_code ] }

Fragment 62549 (Actor 743075,743076)
StreamLocalityProvider { locality_columns: [accruals_ft.currency, accruals_ft.fact_date] } { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ], stream key: [ accruals_ft.currency, accruals_ft.fact_date, $expr2, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
└── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ], stream key: [ accruals_ft.currency, $expr2, accruals_ft.fact_date, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }

Fragment 62550 (Actor 743160,743159)
StreamFilter { predicate: (Not(IsNull(foreign_exchange_rates_eod_ft.rate)) OR (accruals_ft.currency = Coalesce(holding_values_raw_mv.currency_code, accruals_ft.currency))) } { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ], stream key: [ accruals_ft.currency, $expr2, accruals_ft.fact_date, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
└── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ], stream key: [ accruals_ft.currency, $expr2, accruals_ft.fact_date, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }

Fragment 62551 (Actor 743158,743157)
StreamSyncLogStore { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ], stream key: [ accruals_ft.currency, $expr2, accruals_ft.fact_date, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
└── StreamHashJoin { type: LeftOuter, predicate: accruals_ft.currency = foreign_exchange_rates_eod_ft.source_currency_code AND $expr2 = foreign_exchange_rates_eod_ft.target_currency_code AND accruals_ft.fact_date = foreign_exchange_rates_eod_ft.date }
    ├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, foreign_exchange_rates_eod_ft.rate, $expr2, $expr1, accruals_ft.accrual_id, foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ]
    ├── stream key: [ accruals_ft.currency, $expr2, accruals_ft.fact_date, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ]
    ├── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, $expr2, $expr1, accruals_ft.accrual_id ], stream key: [ accruals_ft.currency, $expr2, accruals_ft.fact_date, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
    └── MergeExecutor { output: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.rate ], stream key: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ] }

Fragment 62552 (Actor 743186,743185)
StreamLocalityProvider { locality_columns: [accruals_ft.currency, $expr2, accruals_ft.fact_date] } { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, $expr2, $expr1, accruals_ft.accrual_id ], stream key: [ accruals_ft.currency, $expr2, accruals_ft.fact_date, accruals_ft.account_id, accruals_ft.asset_id, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
└── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, $expr2, $expr1, accruals_ft.accrual_id ], stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }

Fragment 62553 (Actor 743215,743216)
StreamProject { exprs: [accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, Coalesce(holding_values_raw_mv.currency_code, accruals_ft.currency) as $expr2, $expr1, accruals_ft.accrual_id], output_watermarks: [[accruals_ft.fact_date]] } { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, $expr2, $expr1, accruals_ft.accrual_id ], stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
└── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, $expr1, accruals_ft.accrual_id, holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type ], stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }

Fragment 62554 (Actor 743213,743214)
StreamSyncLogStore { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, $expr1, accruals_ft.accrual_id, holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type ], stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
└── StreamHashJoin [window] { type: LeftOuter, predicate: accruals_ft.fact_date = holding_values_raw_mv.dim_value_date AND accruals_ft.account_id = holding_values_raw_mv.account_id AND accruals_ft.asset_id = holding_values_raw_mv.asset_id AND $expr1 = holding_values_raw_mv.type, conditions_to_clean_state_in_join_key: [(accruals_ft.fact_date = holding_values_raw_mv.dim_value_date)], output_watermarks: [[accruals_ft.fact_date], [holding_values_raw_mv.dim_value_date]] }
    ├── output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, holding_values_raw_mv.currency_code, $expr1, accruals_ft.accrual_id, holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type ]
    ├── stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr1, accruals_ft.type, accruals_ft.accrual_id ]
    ├── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, $expr1, accruals_ft.accrual_id ], stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
    └── MergeExecutor { output: [ holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type, holding_values_raw_mv.currency_code ], stream key: [ holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type ] }

Fragment 62555 (Actor 743218,743217)
StreamLocalityProvider { locality_columns: [accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr1] } { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, $expr1, accruals_ft.accrual_id ], stream key: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, $expr1, accruals_ft.type, accruals_ft.accrual_id ] }
└── MergeExecutor { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, $expr1, accruals_ft.accrual_id ], stream key: [ accruals_ft.type, accruals_ft.fact_date, accruals_ft.asset_id, accruals_ft.account_id, accruals_ft.accrual_id ] }

Fragment 62556 (Actor 743629,743630)
StreamProject { exprs: [accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, Case((accruals_ft.type = 'EXPENSE':Varchar), 'LIABILITY':Varchar, 'ASSET':Varchar) as $expr1, accruals_ft.accrual_id], output_watermarks: [[accruals_ft.fact_date]] } { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, $expr1, accruals_ft.accrual_id ], stream key: [ accruals_ft.type, accruals_ft.fact_date, accruals_ft.asset_id, accruals_ft.account_id, accruals_ft.accrual_id ] }
└── StreamFilter { predicate: Not(accruals_ft.is_included) } { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, accruals_ft.accrual_id, accruals_ft.is_included ], stream key: [ accruals_ft.type, accruals_ft.fact_date, accruals_ft.asset_id, accruals_ft.account_id, accruals_ft.accrual_id ] }
    └── StreamTableScan { table: accruals_ft, columns: [account_id, asset_id, fact_date, currency, amount, type, accrual_id, is_included] } { output: [ accruals_ft.account_id, accruals_ft.asset_id, accruals_ft.fact_date, accruals_ft.currency, accruals_ft.amount, accruals_ft.type, accruals_ft.accrual_id, accruals_ft.is_included ], stream key: [ accruals_ft.type, accruals_ft.fact_date, accruals_ft.asset_id, accruals_ft.account_id, accruals_ft.accrual_id ] }
        ├── Upstream { output: [ account_id, asset_id, fact_date, currency, amount, type, accrual_id, is_included ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, asset_id, fact_date, currency, amount, type, accrual_id, is_included ], stream key: [] }

Fragment 62557 (Actor 743634,743633)
StreamLocalityProvider { locality_columns: [holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type] } { output: [ holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type, holding_values_raw_mv.currency_code ], stream key: [ holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type ] }
└── MergeExecutor { output: [ holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type, holding_values_raw_mv.currency_code ], stream key: [ holding_values_raw_mv.account_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.asset_id, holding_values_raw_mv.type ] }

Fragment 62558 (Actor 743631,743632)
StreamTableScan { table: holding_values_raw_mv, columns: [account_id, asset_id, dim_value_date, type, currency_code] } { output: [ holding_values_raw_mv.account_id, holding_values_raw_mv.asset_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.type, holding_values_raw_mv.currency_code ], stream key: [ holding_values_raw_mv.account_id, holding_values_raw_mv.dim_value_date, holding_values_raw_mv.asset_id, holding_values_raw_mv.type ] }
├── Upstream { output: [ account_id, asset_id, dim_value_date, type, currency_code ], stream key: [] }
└── BatchPlanNode { output: [ account_id, asset_id, dim_value_date, type, currency_code ], stream key: [] }

Fragment 62559 (Actor 743635,743636)
StreamTableScan { table: foreign_exchange_rates_eod_ft, columns: [source_currency_code, target_currency_code, date, rate] } { output: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.rate ], stream key: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ] }
├── Upstream { output: [ source_currency_code, target_currency_code, date, rate ], stream key: [] }
└── BatchPlanNode { output: [ source_currency_code, target_currency_code, date, rate ], stream key: [] }

Fragment 62560 (Actor 743621,743622)
StreamLocalityProvider { locality_columns: [foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date] } { output: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.target_currency_code ], stream key: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.target_currency_code ] }
└── MergeExecutor { output: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.target_currency_code ], stream key: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ] }

Fragment 62561 (Actor 743638,743637)
StreamFilter { predicate: (foreign_exchange_rates_eod_ft.target_currency_code = 'SAR':Varchar) } { output: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.target_currency_code ], stream key: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ] }
└── StreamTableScan { table: foreign_exchange_rates_eod_ft, columns: [source_currency_code, date, rate, target_currency_code] } { output: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.date, foreign_exchange_rates_eod_ft.rate, foreign_exchange_rates_eod_ft.target_currency_code ], stream key: [ foreign_exchange_rates_eod_ft.source_currency_code, foreign_exchange_rates_eod_ft.target_currency_code, foreign_exchange_rates_eod_ft.date ] }
    ├── Upstream { output: [ source_currency_code, date, rate, target_currency_code ], stream key: [] }
    └── BatchPlanNode { output: [ source_currency_code, date, rate, target_currency_code ], stream key: [] }

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