RWM Console cluster: risingwave.platform.svc.cluster.local

← cluster insights objects holding_values_journal_density_mv explain
Overview Objects Graph History
materialized view · insights.holding_values_journal_density_mv profiled over 5s
seconds (1–30)

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 filteredWindow state — add a WHERE rank <= N to bound it
42 operators
Materialize · insights.holding_values_journal_density_mv
0% idle 2 actors
Project · LeftOuter · settled_position_series_mv.account_id = settled_cost_basis_…
2 actors
HashJoin · LeftOuter · settled_position_series_mv.account_id = settled_cost_basis_… 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 · settled_cost_basis_series_mv
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project · Inner · (asset_prices_eod_ft_next.date >= settled_position_series_m…
2 actors
Filter · Inner · (asset_prices_eod_ft_next.date >= settled_position_series_m…
0% idle 2 actors
HashJoin · Inner · (asset_prices_eod_ft_next.date >= settled_position_series_m… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups
0% idle 2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Filter · asset_prices_eod_ft_next
0% idle 2 actors
StreamScan · asset_prices_eod_ft_next
0% idle 2 actors
BatchPlan
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project · Inner · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f…
2 actors
Filter · Inner · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f…
0% idle 2 actors
HashJoin · Inner · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups
0% idle 2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project
2 actors
HashAgg Aggregation state — unbounded unless keyed or temporally filtered
0% idle 2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project
2 actors
Merge
2 actors
Merge
2 actors
Exchange
0% idle 0 actors
Project · settled_position_series_mv
2 actors
OverWindow · settled_position_series_mv Window state — add a WHERE rank <= N to bound it
0% idle 2 actors
StreamScan · settled_position_series_mv
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.
Materialize · insights.holding_values_journal_density_mv Materialize insights.holding_values… idle · 2 actors Project · LeftOuter · settled_position_series_mv.account_id = settled_cost_basis_… Project LeftOuter · settled_pos… — · 2 actors HashJoin · LeftOuter · settled_position_series_mv.account_id = settled_cost_basis_… HashJoin LeftOuter · settled_pos… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors StreamScan · settled_cost_basis_series_mv StreamScan settled_cost_basis_seri… idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · Inner · (asset_prices_eod_ft_next.date >= settled_position_series_m… Project Inner · (asset_prices_e… — · 2 actors Filter · Inner · (asset_prices_eod_ft_next.date >= settled_position_series_m… Filter Inner · (asset_prices_e… idle · 2 actors HashJoin · Inner · (asset_prices_eod_ft_next.date >= settled_position_series_m… HashJoin Inner · (asset_prices_e… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Filter · asset_prices_eod_ft_next Filter asset_prices_eod_ft_next idle · 2 actors StreamScan · asset_prices_eod_ft_next StreamScan asset_prices_eod_ft_next idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · Inner · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f… Project Inner · ($expr1 <= $exp… — · 2 actors Filter · Inner · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f… Filter Inner · ($expr1 <= $exp… idle · 2 actors HashJoin · Inner · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f… HashJoin Inner · ($expr1 <= $exp… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · settled_position_series_mv Project settled_position_series… — · 2 actors OverWindow · settled_position_series_mv OverWindow settled_position_series… idle · 2 actors StreamScan · settled_position_series_mv StreamScan settled_position_series… idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors
Streaming operator plan from EXPLAIN ANALYZE. Node heat = backpressure. Drag to pan, scroll to zoom.
Fragments (DESCRIBE FRAGMENTS) — click to expand
Fragment 12213 (Actor 33315,33316)
StreamMaterialize { columns: [account_id, asset_id, dim_value_date, type, currency_code, market_value, average_cost_per_unit, average_cost_per_unit_system_currency, total_cost_system_currency, cost_fx_provenance, purchased_quantity, settled_position_series_mv.dim_settlement_date(hidden), $expr2(hidden), settled_cost_basis_series_mv.effective_from(hidden), settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum(hidden)], stream_key: [account_id, asset_id, currency_code, settled_position_series_mv.dim_settlement_date, $expr2, dim_value_date, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum], pk_columns: [account_id, asset_id, currency_code, settled_position_series_mv.dim_settlement_date, $expr2, dim_value_date, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum], pk_conflict: NoCheck }
├── output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, asset_prices_eod_ft_next.date, 'ASSET':Varchar, settled_position_series_mv.currency_code, $expr4, settled_cost_basis_series_mv.average_cost_per_unit, settled_cost_basis_series_mv.average_cost_per_unit_system_currency, settled_cost_basis_series_mv.total_cost_system_currency, settled_cost_basis_series_mv.cost_fx_provenance, settled_position_series_mv.settled_quantity, settled_position_series_mv.dim_settlement_date, $expr2, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ]
├── stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.date, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ]
└── StreamProject { exprs: [settled_position_series_mv.account_id, settled_position_series_mv.asset_id, asset_prices_eod_ft_next.date, 'ASSET':Varchar, settled_position_series_mv.currency_code, (settled_position_series_mv.settled_quantity * asset_prices_eod_ft_next.reference_price) as $expr4, settled_cost_basis_series_mv.average_cost_per_unit, settled_cost_basis_series_mv.average_cost_per_unit_system_currency, settled_cost_basis_series_mv.total_cost_system_currency, settled_cost_basis_series_mv.cost_fx_provenance, settled_position_series_mv.settled_quantity, settled_position_series_mv.dim_settlement_date, $expr2, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum] }
    ├── output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, asset_prices_eod_ft_next.date, 'ASSET':Varchar, settled_position_series_mv.currency_code, $expr4, settled_cost_basis_series_mv.average_cost_per_unit, settled_cost_basis_series_mv.average_cost_per_unit_system_currency, settled_cost_basis_series_mv.total_cost_system_currency, settled_cost_basis_series_mv.cost_fx_provenance, settled_position_series_mv.settled_quantity, settled_position_series_mv.dim_settlement_date, $expr2, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ]
    ├── stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.date, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ]
    └── StreamHashJoin { type: LeftOuter, predicate: settled_position_series_mv.account_id = settled_cost_basis_series_mv.account_id AND settled_position_series_mv.asset_id = settled_cost_basis_series_mv.asset_id AND settled_position_series_mv.currency_code = settled_cost_basis_series_mv.currency_code AND (settled_cost_basis_series_mv.effective_from <= asset_prices_eod_ft_next.date) AND (IsNull(settled_cost_basis_series_mv.effective_to) OR (asset_prices_eod_ft_next.date < settled_cost_basis_series_mv.effective_to)) }
        ├── output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.settled_quantity, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, settled_cost_basis_series_mv.average_cost_per_unit, settled_cost_basis_series_mv.average_cost_per_unit_system_currency, settled_cost_basis_series_mv.total_cost_system_currency, settled_cost_basis_series_mv.cost_fx_provenance, settled_position_series_mv.dim_settlement_date, $expr2, settled_cost_basis_series_mv.account_id, settled_cost_basis_series_mv.asset_id, settled_cost_basis_series_mv.currency_code, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ]
        ├── stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.date, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ]
        ├── MergeExecutor { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.settled_quantity, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.asset_id ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.date ] }
        └── MergeExecutor { output: [ settled_cost_basis_series_mv.account_id, settled_cost_basis_series_mv.asset_id, settled_cost_basis_series_mv.currency_code, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.effective_to, settled_cost_basis_series_mv.average_cost_per_unit, settled_cost_basis_series_mv.average_cost_per_unit_system_currency, settled_cost_basis_series_mv.total_cost_system_currency, settled_cost_basis_series_mv.cost_fx_provenance, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ], stream key: [ settled_cost_basis_series_mv.account_id, settled_cost_basis_series_mv.asset_id, settled_cost_basis_series_mv.currency_code, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ] }

Fragment 12214 (Actor 33317,33318)
StreamProject { exprs: [settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.settled_quantity, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.asset_id] } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.settled_quantity, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.asset_id ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.date ] }
└── StreamFilter { predicate: (asset_prices_eod_ft_next.date >= settled_position_series_mv.dim_settlement_date) AND (IsNull(first_value) OR (asset_prices_eod_ft_next.date < first_value)) } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr2, asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, $expr3 ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.date ] }
    └── StreamHashJoin { type: Inner, predicate: settled_position_series_mv.asset_id = asset_prices_eod_ft_next.asset_id AND $expr2 = $expr3 } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr2, asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, $expr3 ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2, asset_prices_eod_ft_next.date ] }
        ├── MergeExecutor { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr2, asset_prices_eod_ft_next.asset_id ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2 ] }
        └── MergeExecutor { output: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, $expr3 ], stream key: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date ] }

Fragment 12215 (Actor 33320,33319)
StreamProject { exprs: [settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr2, asset_prices_eod_ft_next.asset_id] } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr2, asset_prices_eod_ft_next.asset_id ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2 ] }
└── StreamFilter { predicate: ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= first_value)) } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr1, asset_prices_eod_ft_next.asset_id, $expr2 ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2 ] }
    └── StreamHashJoin { type: Inner, predicate: settled_position_series_mv.asset_id = asset_prices_eod_ft_next.asset_id } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr1, asset_prices_eod_ft_next.asset_id, $expr2 ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, $expr2 ] }
        ├── MergeExecutor { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr1 ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date ] }
        └── MergeExecutor { output: [ asset_prices_eod_ft_next.asset_id, $expr2 ], stream key: [ asset_prices_eod_ft_next.asset_id, $expr2 ] }

Fragment 12216 (Actor 33289,33290)
StreamProject { exprs: [settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, AtTimeZone(DateTrunc('MONTH':Varchar, AtTimeZone(settled_position_series_mv.dim_settlement_date::Timestamp, 'UTC':Varchar), 'UTC':Varchar), 'UTC':Varchar)::Date as $expr1] } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value, $expr1 ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date ] }
└── StreamOverWindow { window_functions: [first_value(settled_position_series_mv.dim_settlement_date) OVER(PARTITION BY settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code ORDER BY settled_position_series_mv.dim_settlement_date ASC ROWS BETWEEN 1 FOLLOWING AND 1 FOLLOWING)] } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity, first_value ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date ] }
    └── StreamTableScan { table: settled_position_series_mv, columns: [account_id, asset_id, currency_code, dim_settlement_date, settled_quantity] } { output: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date, settled_position_series_mv.settled_quantity ], stream key: [ settled_position_series_mv.account_id, settled_position_series_mv.asset_id, settled_position_series_mv.currency_code, settled_position_series_mv.dim_settlement_date ] }
        ├── Upstream { output: [ account_id, asset_id, currency_code, dim_settlement_date, settled_quantity ], stream key: [] }
        └── BatchPlanNode { output: [ account_id, asset_id, currency_code, dim_settlement_date, settled_quantity ], stream key: [] }

Fragment 12217 (Actor 33322,33321)
StreamProject { exprs: [asset_prices_eod_ft_next.asset_id, $expr2] } { output: [ asset_prices_eod_ft_next.asset_id, $expr2 ], stream key: [ asset_prices_eod_ft_next.asset_id, $expr2 ] }
└── StreamHashAgg { group_key: [asset_prices_eod_ft_next.asset_id, $expr2], aggs: [count] } { output: [ asset_prices_eod_ft_next.asset_id, $expr2, count ], stream key: [ asset_prices_eod_ft_next.asset_id, $expr2 ] }
    └── MergeExecutor { output: [ asset_prices_eod_ft_next.asset_id, $expr2, asset_prices_eod_ft_next.date ], stream key: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date ] }

Fragment 12218 (Actor 33291,33292)
StreamProject { exprs: [asset_prices_eod_ft_next.asset_id, AtTimeZone(DateTrunc('MONTH':Varchar, AtTimeZone(asset_prices_eod_ft_next.date::Timestamp, 'UTC':Varchar), 'UTC':Varchar), 'UTC':Varchar)::Date as $expr2, asset_prices_eod_ft_next.date], output_watermarks: [[asset_prices_eod_ft_next.date]] } { output: [ asset_prices_eod_ft_next.asset_id, $expr2, asset_prices_eod_ft_next.date ], stream key: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date ] }
└── MergeExecutor { output: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price ], stream key: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date ] }

Fragment 12219 (Actor 33294,33293)
StreamFilter { predicate: Not(IsNull(asset_prices_eod_ft_next.reference_price)) } { output: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price ], stream key: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date ] }
└── StreamTableScan { table: asset_prices_eod_ft_next, columns: [asset_id, date, reference_price] } { output: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price ], stream key: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date ] }
    ├── Upstream { output: [ asset_id, date, reference_price ], stream key: [] }
    └── BatchPlanNode { output: [ asset_id, date, reference_price ], stream key: [] }

Fragment 12220 (Actor 33296,33295)
StreamProject { exprs: [asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, AtTimeZone(DateTrunc('MONTH':Varchar, AtTimeZone(asset_prices_eod_ft_next.date::Timestamp, 'UTC':Varchar), 'UTC':Varchar), 'UTC':Varchar)::Date as $expr3], output_watermarks: [[asset_prices_eod_ft_next.date]] } { output: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price, $expr3 ], stream key: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date ] }
└── MergeExecutor { output: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date, asset_prices_eod_ft_next.reference_price ], stream key: [ asset_prices_eod_ft_next.asset_id, asset_prices_eod_ft_next.date ] }

Fragment 12221 (Actor 33298,33297)
StreamTableScan { table: settled_cost_basis_series_mv, columns: [account_id, asset_id, currency_code, effective_from, effective_to, average_cost_per_unit, average_cost_per_unit_system_currency, total_cost_system_currency, cost_fx_provenance, settled_cost_basis_carried_mv.sum] }
├── output: [ settled_cost_basis_series_mv.account_id, settled_cost_basis_series_mv.asset_id, settled_cost_basis_series_mv.currency_code, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.effective_to, settled_cost_basis_series_mv.average_cost_per_unit, settled_cost_basis_series_mv.average_cost_per_unit_system_currency, settled_cost_basis_series_mv.total_cost_system_currency, settled_cost_basis_series_mv.cost_fx_provenance, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ]
├── stream key: [ settled_cost_basis_series_mv.account_id, settled_cost_basis_series_mv.asset_id, settled_cost_basis_series_mv.currency_code, settled_cost_basis_series_mv.effective_from, settled_cost_basis_series_mv.settled_cost_basis_carried_mv.sum ]
├── Upstream { output: [ account_id, asset_id, currency_code, effective_from, effective_to, average_cost_per_unit, average_cost_per_unit_system_currency, total_cost_system_currency, cost_fx_provenance, settled_cost_basis_carried_mv.sum ], stream key: [] }
└── BatchPlanNode { output: [ account_id, asset_id, currency_code, effective_from, effective_to, average_cost_per_unit, average_cost_per_unit_system_currency, total_cost_system_currency, cost_fx_provenance, settled_cost_basis_carried_mv.sum ], stream key: [] }