Job is idle — throughput ~0; structure shown.
Fragment 51704 (Actor 736537,736536)
StreamMaterialize { columns: [benchmark_id, fact_date, daily_subperiod_return], stream_key: [benchmark_id, fact_date], pk_columns: [benchmark_id, fact_date], pk_conflict: NoCheck }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, $expr1 ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
└── StreamProject { exprs: [benchmark_values_ft.benchmark_id, benchmark_values_ft.date, ((benchmark_values_ft.amount - first_value) / first_value) as $expr1] }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, $expr1 ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
└── StreamFilter { predicate: Not(IsNull(first_value)) AND (first_value <> 0:Decimal) }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id, first_value ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
└── StreamOverWindow { window_functions: [first_value(benchmark_values_ft.amount) OVER(PARTITION BY benchmark_values_ft.benchmark_id ORDER BY benchmark_values_ft.date ASC ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING)] }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id, first_value ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
└── StreamLocalityProvider { locality_columns: [benchmark_values_ft.benchmark_id] }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
└── MergeExecutor
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id ]
└── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
Fragment 51705 (Actor 736538,736539)
StreamSyncLogStore { output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id ], stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ] }
└── StreamHashJoin { type: Inner, predicate: benchmark_values_ft.benchmark_id = benchmarks_dm.id }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount, benchmarks_dm.id ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
├── MergeExecutor { output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount ], stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ] }
└── MergeExecutor { output: [ benchmarks_dm.id ], stream key: [ benchmarks_dm.id ] }
Fragment 51706 (Actor 736541,736540)
StreamLocalityProvider { locality_columns: [benchmark_values_ft.benchmark_id] }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
└── MergeExecutor { output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount ], stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ] }
Fragment 51707 (Actor 736557,736556)
StreamTableScan { table: benchmark_values_ft, columns: [benchmark_id, date, amount] }
├── output: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date, benchmark_values_ft.amount ]
├── stream key: [ benchmark_values_ft.benchmark_id, benchmark_values_ft.date ]
├── Upstream { output: [ benchmark_id, date, amount ], stream key: [] }
└── BatchPlanNode { output: [ benchmark_id, date, amount ], stream key: [] }
Fragment 51708 (Actor 736548,736549)
StreamProject { exprs: [benchmarks_dm.id] } { output: [ benchmarks_dm.id ], stream key: [ benchmarks_dm.id ] }
└── StreamFilter { predicate: IsNull(benchmarks_dm.disabled_at) } { output: [ benchmarks_dm.id, benchmarks_dm.disabled_at ], stream key: [ benchmarks_dm.id ] }
└── StreamTableScan { table: benchmarks_dm, columns: [id, disabled_at] } { output: [ benchmarks_dm.id, benchmarks_dm.disabled_at ], stream key: [ benchmarks_dm.id ] }
├── Upstream { output: [ id, disabled_at ], stream key: [] }
└── BatchPlanNode { output: [ id, disabled_at ], stream key: [] }