RWM Console cluster: risingwave-ai-hub-int.ai-hub-rwm.svc.cluster.local

← cluster insights objects cash_values_journal_density_mv explain
Overview Objects Graph History
materialized view · insights.cash_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
49 operators
Materialize · insights.cash_values_journal_density_mv
0% idle 2 actors
Project · (asset_prices_eod_ft.date >= settled_cash_series_mv.dim_set…
— — 2 actors
Filter · (asset_prices_eod_ft.date >= settled_cash_series_mv.dim_set…
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset…
— — 2 actors
HashJoin · Inner · settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset… Stateful hash join (4 state tables) — consider a temporal join for dimension lookups
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project
— — 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Filter · asset_prices_eod_ft
0% idle 2 actors
StreamScan · asset_prices_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 · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f…
— — 2 actors
Filter · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f…
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
SyncLogStore · Inner · settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset…
— — 2 actors
HashJoin · Inner · settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset… 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
LocalityProvider
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
LocalityProvider
0% idle 2 actors
Merge
— — 2 actors
Exchange
0% idle 0 actors
Project · settled_cash_series_mv
— — 2 actors
OverWindow · settled_cash_series_mv Window state — add a WHERE rank <= N to bound it
0% idle 2 actors
StreamScan · settled_cash_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.cash_values_journal_density_mv Materialize insights.cash_values_jo… idle · 2 actors Project · (asset_prices_eod_ft.date >= settled_cash_series_mv.dim_set… Project (asset_prices_eod_ft.da… — · 2 actors Filter · (asset_prices_eod_ft.date >= settled_cash_series_mv.dim_set… Filter (asset_prices_eod_ft.da… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset… SyncLogStore Inner · settled_cash_se… — · 2 actors HashJoin · Inner · settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset… HashJoin Inner · settled_cash_se… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider 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 Filter asset_prices_eod_ft idle · 2 actors StreamScan · asset_prices_eod_ft StreamScan asset_prices_eod_ft idle · 2 actors BatchPlan BatchPlan — · 2 actors Merge Merge — · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f… Project ($expr1 <= $expr2) AND … — · 2 actors Filter · ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= f… Filter ($expr1 <= $expr2) AND … idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors SyncLogStore · Inner · settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset… SyncLogStore Inner · settled_cash_se… — · 2 actors HashJoin · Inner · settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset… HashJoin Inner · settled_cash_se… idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project Project — · 2 actors HashAgg HashAgg idle · 2 actors LocalityProvider LocalityProvider 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 LocalityProvider LocalityProvider idle · 2 actors Merge Merge — · 2 actors Exchange Exchange idle · 0 actors Project · settled_cash_series_mv Project settled_cash_series_mv — · 2 actors OverWindow · settled_cash_series_mv OverWindow settled_cash_series_mv idle · 2 actors StreamScan · settled_cash_series_mv StreamScan settled_cash_series_mv 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 19452 (Actor 164413,164412)
StreamMaterialize { columns: [account_id, asset_id, dim_value_date, type, currency_code, market_value, average_cost_per_unit, purchased_quantity, $expr2(hidden), settled_cash_series_mv.dim_settlement_date(hidden)], stream_key: [asset_id, $expr2, account_id, currency_code, settled_cash_series_mv.dim_settlement_date, dim_value_date], pk_columns: [asset_id, $expr2, account_id, currency_code, settled_cash_series_mv.dim_settlement_date, dim_value_date], pk_conflict: NoCheck }
├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, asset_prices_eod_ft.date, 'ASSET':Varchar, settled_cash_series_mv.currency_code, $expr4, asset_prices_eod_ft.close, settled_cash_series_mv.settled_cash_balance, $expr2, settled_cash_series_mv.dim_settlement_date ]
├── stream key: [ settled_cash_series_mv.asset_id, $expr2, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, asset_prices_eod_ft.date ]
└── StreamProject { exprs: [settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, asset_prices_eod_ft.date, 'ASSET':Varchar, settled_cash_series_mv.currency_code, (settled_cash_series_mv.settled_cash_balance * asset_prices_eod_ft.close) as $expr4, asset_prices_eod_ft.close, settled_cash_series_mv.settled_cash_balance, $expr2, settled_cash_series_mv.dim_settlement_date] }
    ├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, asset_prices_eod_ft.date, 'ASSET':Varchar, settled_cash_series_mv.currency_code, $expr4, asset_prices_eod_ft.close, settled_cash_series_mv.settled_cash_balance, $expr2, settled_cash_series_mv.dim_settlement_date ]
    ├── stream key: [ settled_cash_series_mv.asset_id, $expr2, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, asset_prices_eod_ft.date ]
    └── StreamFilter { predicate: (asset_prices_eod_ft.date >= settled_cash_series_mv.dim_settlement_date) AND (IsNull(first_value) OR (asset_prices_eod_ft.date < first_value)) }
        ├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id, asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, $expr3 ]
        ├── stream key: [ settled_cash_series_mv.asset_id, $expr2, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, asset_prices_eod_ft.date ]
        └── MergeExecutor
            ├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id, asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, $expr3 ]
            └── stream key: [ settled_cash_series_mv.asset_id, $expr2, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, asset_prices_eod_ft.date ]

Fragment 19453 (Actor 164424,164425)
StreamSyncLogStore
├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id, asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, $expr3 ]
├── stream key: [ settled_cash_series_mv.asset_id, $expr2, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, asset_prices_eod_ft.date ]
└── StreamHashJoin { type: Inner, predicate: settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset_id AND $expr2 = $expr3 }
    ├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id, asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, $expr3 ]
    ├── stream key: [ settled_cash_series_mv.asset_id, $expr2, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, asset_prices_eod_ft.date ]
    ├── MergeExecutor { output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id ], stream key: [ settled_cash_series_mv.asset_id, $expr2, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date ] }
    └── MergeExecutor { output: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, $expr3 ], stream key: [ asset_prices_eod_ft.asset_id, $expr3, asset_prices_eod_ft.date ] }

Fragment 19454 (Actor 164411,164410)
StreamLocalityProvider { locality_columns: [settled_cash_series_mv.asset_id, $expr2] }
├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id ]
├── stream key: [ settled_cash_series_mv.asset_id, $expr2, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date ]
└── MergeExecutor { output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id ], stream key: [ settled_cash_series_mv.asset_id, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, $expr2 ] }

Fragment 19455 (Actor 164414,164415)
StreamProject { exprs: [settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id] }
├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr2, asset_prices_eod_ft.asset_id ]
├── stream key: [ settled_cash_series_mv.asset_id, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, $expr2 ]
└── StreamFilter { predicate: ($expr1 <= $expr2) AND (IsNull(first_value) OR ($expr2 <= first_value)) }
    ├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr1, asset_prices_eod_ft.asset_id, $expr2 ]
    ├── stream key: [ settled_cash_series_mv.asset_id, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, $expr2 ]
    └── MergeExecutor { output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr1, asset_prices_eod_ft.asset_id, $expr2 ], stream key: [ settled_cash_series_mv.asset_id, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, $expr2 ] }

Fragment 19456 (Actor 164416,164417)
StreamSyncLogStore { output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr1, asset_prices_eod_ft.asset_id, $expr2 ], stream key: [ settled_cash_series_mv.asset_id, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, $expr2 ] }
└── StreamHashJoin { type: Inner, predicate: settled_cash_series_mv.asset_id = asset_prices_eod_ft.asset_id }
    ├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr1, asset_prices_eod_ft.asset_id, $expr2 ]
    ├── stream key: [ settled_cash_series_mv.asset_id, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, $expr2 ]
    ├── MergeExecutor { output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr1 ], stream key: [ settled_cash_series_mv.asset_id, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date ] }
    └── MergeExecutor { output: [ asset_prices_eod_ft.asset_id, $expr2 ], stream key: [ asset_prices_eod_ft.asset_id, $expr2 ] }

Fragment 19457 (Actor 164408,164409)
StreamLocalityProvider { locality_columns: [settled_cash_series_mv.asset_id] }
├── output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr1 ]
├── stream key: [ settled_cash_series_mv.asset_id, settled_cash_series_mv.account_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date ]
└── MergeExecutor { output: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date, settled_cash_series_mv.settled_cash_balance, first_value, $expr1 ], stream key: [ settled_cash_series_mv.account_id, settled_cash_series_mv.asset_id, settled_cash_series_mv.currency_code, settled_cash_series_mv.dim_settlement_date ] }

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

Fragment 19459 (Actor 164419,164418)
StreamProject { exprs: [asset_prices_eod_ft.asset_id, $expr2] } { output: [ asset_prices_eod_ft.asset_id, $expr2 ], stream key: [ asset_prices_eod_ft.asset_id, $expr2 ] }
└── StreamHashAgg { group_key: [asset_prices_eod_ft.asset_id, $expr2], aggs: [count] } { output: [ asset_prices_eod_ft.asset_id, $expr2, count ], stream key: [ asset_prices_eod_ft.asset_id, $expr2 ] }
    └── StreamLocalityProvider { locality_columns: [asset_prices_eod_ft.asset_id, $expr2] } { output: [ asset_prices_eod_ft.asset_id, $expr2, asset_prices_eod_ft.date ], stream key: [ asset_prices_eod_ft.asset_id, $expr2, asset_prices_eod_ft.date ] }
        └── MergeExecutor { output: [ asset_prices_eod_ft.asset_id, $expr2, asset_prices_eod_ft.date ], stream key: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date ] }

Fragment 19460 (Actor 164423,164422)
StreamProject { exprs: [asset_prices_eod_ft.asset_id, AtTimeZone(DateTrunc('MONTH':Varchar, AtTimeZone(asset_prices_eod_ft.date::Timestamp, 'UTC':Varchar), 'UTC':Varchar), 'UTC':Varchar)::Date as $expr2, asset_prices_eod_ft.date], output_watermarks: [[asset_prices_eod_ft.date]] } { output: [ asset_prices_eod_ft.asset_id, $expr2, asset_prices_eod_ft.date ], stream key: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date ] }
└── MergeExecutor { output: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close ], stream key: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date ] }

Fragment 19461 (Actor 164429,164428)
StreamFilter { predicate: Not(IsNull(asset_prices_eod_ft.close)) } { output: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close ], stream key: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date ] }
└── StreamTableScan { table: asset_prices_eod_ft, columns: [asset_id, date, close] } { output: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close ], stream key: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date ] }
    ├── Upstream { output: [ asset_id, date, close ], stream key: [] }
    └── BatchPlanNode { output: [ asset_id, date, close ], stream key: [] }

Fragment 19462 (Actor 164427,164426)
StreamLocalityProvider { locality_columns: [asset_prices_eod_ft.asset_id, $expr3] } { output: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, $expr3 ], stream key: [ asset_prices_eod_ft.asset_id, $expr3, asset_prices_eod_ft.date ] }
└── MergeExecutor { output: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, $expr3 ], stream key: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date ] }

Fragment 19463 (Actor 164406,164407)
StreamProject { exprs: [asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, AtTimeZone(DateTrunc('MONTH':Varchar, AtTimeZone(asset_prices_eod_ft.date::Timestamp, 'UTC':Varchar), 'UTC':Varchar), 'UTC':Varchar)::Date as $expr3], output_watermarks: [[asset_prices_eod_ft.date]] }
├── output: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close, $expr3 ]
├── stream key: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date ]
└── MergeExecutor { output: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date, asset_prices_eod_ft.close ], stream key: [ asset_prices_eod_ft.asset_id, asset_prices_eod_ft.date ] }