{"record":{"id":"662c996700fe1d95","repo":"risingwavelabs/risingwave","slug":"match-recognize-sum-measure-slot-has-no-kernel","errorCode":null,"errorMessage":"MATCH_RECOGNIZE SUM measure slot has no kernel","messagePattern":"MATCH_RECOGNIZE SUM measure slot has no kernel","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"src/stream/src/executor/match_recognize/executor.rs","lineNumber":290,"sourceCode":"                .iter()\n                .enumerate()\n                .filter(|(_, l)| matches(l))\n                .filter_map(|(j, _)| rows[start + j].row.datum_at(self.col_idx))\n                .min_by(|a, b| a.default_cmp(b))\n                .map(|r| r.into_scalar_impl()),\n            MeasureSlotKind::Max => labels\n                .iter()\n                .enumerate()\n                .filter(|(_, l)| matches(l))\n                .filter_map(|(j, _)| rows[start + j].row.datum_at(self.col_idx))\n                .max_by(|a, b| a.default_cmp(b))\n                .map(|r| r.into_scalar_impl()),\n            // Rejected in `from_protobuf`; a constructed slot never carries it. NULL (the\n            // non-strict convention) rather than a panic, should that invariant ever break.\n            MeasureSlotKind::Unspecified => None,\n            MeasureSlotKind::Sum => {\n                let agg = self.agg.as_ref().ok_or_else(|| {\n                    anyhow::anyhow!(\"MATCH_RECOGNIZE SUM measure slot has no kernel\")\n                })?;\n                // Feed the kernel a single-column chunk of the col values over the matching rows.\n                let input: Vec<(Op, OwnedRow)> = labels\n                    .iter()\n                    .enumerate()\n                    .filter(|(_, l)| matches(l))\n                    .map(|(j, _)| (Op::Insert, OwnedRow::new(vec![col_at(j)])))\n                    .collect();\n                if input.is_empty() {\n                    None\n                } else {\n                    let chunk = StreamChunk::from_rows(&input, std::slice::from_ref(&agg.col_type));\n                    // A kernel error here is a DATA error — numeric overflow in SUM is the\n                    // canonical one — on rows that recovery will replay verbatim: propagating it\n                    // kills the actor and every restart replays the same rows into the same\n                    // overflow, an unrecoverable crash loop from one bad match. Mirror what\n                    // `NonStrictExpression` does for every other expression in this operator:\n                    // report through the actor's error report and yield NULL for the measure.","sourceCodeStart":272,"sourceCodeEnd":308,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/match_recognize/executor.rs#L272-L308","documentation":"During result resolution (`resolve`, called from `next_pos`), a SUM measure slot has no aggregation kernel attached (`self.agg` is None). Per the code's own invariant this cannot happen because from_protobuf rejects such slots; the executor deliberately returns a NULL-convention error instead of panicking should that invariant ever break.","triggerScenarios":"resolve() hits MeasureSlotKind::Sum while self.agg is None — only possible if the from_protobuf invariant was bypassed (e.g. struct constructed directly in tests or via state recovery of a corrupt plan).","commonSituations":"Directly constructed AggSlot/measure structs in unit tests; corrupted recovered state; future code changes breaking the deserialization invariant.","solutions":["Construct measure slots only through from_protobuf, which enforces agg presence for SUM.","Add/update tests to build slots via the deserialization path.","Check whether state recovery loaded a plan from a mismatched/older version; re-create the job."],"exampleFix":"// before\nlet slot = MeasureSlot { kind: MeasureSlotKind::Sum, agg: None };\n// after\nlet slot = MeasureSlot::from_protobuf(&pb_with_agg_call, &error_report)?;","handlingStrategy":"type-guard","validationCode":"// Before resolving, ensure the kernel exists:\nif slot.kind == MeasureSlotKind::Sum && slot.agg.is_none() {\n    return Ok(ScalarImpl::Float64(ordered_float::OrderedFloat(f64::NAN))); // or error early\n}","typeGuard":"fn agg_kernel(slot: &MeasureSlot) -> Option<&Arc<dyn StreamingAggState>> {\n    slot.agg.as_ref().map(|a| &a.func)\n}","tryCatchPattern":"// Defensive resolution at the call site:\nlet Some(agg) = &slot.agg else {\n    log::warn!(\"SUM measure without kernel; returning NULL\");\n    return Ok(None);\n};","preventionTips":["Only construct MeasureSlot via from_protobuf.","Add debug_assert!(agg.is_some()) for Sum kinds in constructors.","Cover SUM/AVG slots in deserialization unit tests."],"tags":["match-recognize","aggregation","internal-invariant"],"backgroundTag":"internal-invariant-violation","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}