{"record":{"id":"1d424a441e4655cc","repo":"risingwavelabs/risingwave","slug":"unreachable-prev-curr","errorCode":null,"errorMessage":"unreachable (prev != curr)","messagePattern":"unreachable \\(prev != curr\\)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/dynamic_filter.rs","lineNumber":256,"sourceCode":"                    let range = match self.comparator {\n                        GreaterThan | LessThanOrEqual => (Excluded(c), Included(p)),\n                        GreaterThanOrEqual | LessThan => (Included(c), Excluded(p)),\n                        _ => unreachable!(),\n                    };\n                    let is_insert = matches!(self.comparator, GreaterThan | GreaterThanOrEqual);\n                    (range, true, is_insert)\n                } else {\n                    // c > p\n                    let range = match self.comparator {\n                        GreaterThan | LessThanOrEqual => (Excluded(p), Included(c)),\n                        GreaterThanOrEqual | LessThan => (Included(p), Excluded(c)),\n                        _ => unreachable!(),\n                    };\n                    let is_insert = matches!(self.comparator, LessThan | LessThanOrEqual);\n                    (range, false, is_insert)\n                }\n            }\n            (None, None) => unreachable!(), // prev != curr\n        }\n    }\n\n    fn to_row_bound(bound: Bound<ScalarImpl>) -> Bound<impl Row> {\n        bound.map(|s| once(Some(s)))\n    }\n\n    #[try_stream(ok = Message, error = StreamExecutorError)]\n    async fn execute_inner(mut self) {\n        let input_l = self.source_l.take().unwrap();\n        let input_r = self.source_r.take().unwrap();\n\n        // Derive the dynamic expression\n        let l_data_type = input_l.schema().data_types()[self.key_l].clone();\n        let r_data_type = input_r.schema().data_types()[0].clone();\n        // The types are aligned by frontend.\n        assert_eq!(l_data_type, r_data_type);\n","sourceCodeStart":238,"sourceCodeEnd":274,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/dynamic_filter.rs#L238-L274","documentation":"This is a deliberate panic in RisingWave's dynamic filter executor. In `get_range`, the executor computes a cache-lookup range based on the relationship between the previous and current watermark values; the `(None, None)` arm assumes that if both are None this branch is never taken because `prev != curr` is an invariant maintained by the caller (`execute_inner` only enters this path when the watermark actually advanced). If the invariant is violated the executor panics with `unreachable!()` instead of returning a recoverable error.","triggerScenarios":"Executing the dynamic filter executor with a None-to-None watermark transition where the cache-range calculation expects an actual prev-to-curr watermark move; i.e., `prev_watermark` and `curr_watermark` are both None while control flow reaches the `(None, None)` match arm in `get_range`.","commonSituations":"A bug in watermark propagation upstream (e.g., a source or merge executor emitting an initial watermark that is None twice), or a custom/modified executor wiring where `execute_inner` calls `get_range` without first checking that the watermark advanced. Real users should essentially never hit this; hitting it indicates an internal invariant break.","solutions":["Inspect the upstream watermark chain feeding the dynamic filter (source, merge, watermark decoders) to find why a None watermark was re-delivered; fix the producer so watermarks only advance.","Check for a recent RisingWave version regression and upgrade/downgrade the binary accordingly; report the issue with the panic backtrace if reproducible.","As a defensive code change, return an error or log and treat (None, None) as an empty range instead of panicking, so the actor can be gracefully failed instead of aborting the process.","Verify that `execute_inner` gates `get_range` on `prev != curr` (or prev being Some) before the call."],"exampleFix":"// before\n(None, None) => unreachable!(), // prev != curr\n// after\n(None, None) => {\n    tracing::warn!(\"dynamic filter: prev and curr watermark both None; using empty range\");\n    (Bound::Unbounded, Bound::Unbounded, false)\n}","handlingStrategy":"try-catch","validationCode":"// Before wiring a dynamic filter, ensure the watermark source always advances monotonically:\n// prev.is_some() implies curr.is_some() && curr > prev when get_range is invoked.\nfn watermark_advanced(prev: &Option<ScalarImpl>, curr: &Option<ScalarImpl>) -> bool {\n    match (prev, curr) {\n        (Some(p), Some(c)) => c > p,\n        (None, Some(_)) => true,\n        _ => false,\n    }\n}","typeGuard":"fn has_valid_watermark(w: &Option<ScalarImpl>) -> bool { w.is_some() }","tryCatchPattern":"// Run the stream cluster under a supervisor; a panic here fails the actor.\n// Capture the panic backtrace from .risingwave/log and report upstream:\nmatch actor_join_handle.await {\n    Ok(_) => {},\n    Err(e) => log::error!(\"dynamic filter actor panicked: {e}\"),\n}","preventionTips":["Never feed the dynamic filter from sources that can re-emit stale/None watermarks.","Pin and test executor wiring changes against watermark-monotonicity invariants.","Keep RisingWave updated; this panic usually indicates an internal bug."],"tags":["streaming","watermark","invariant-violation","rust","panic"],"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-14T11:17:12.474Z"}