{"record":{"id":"4d210ec571f3f0c9","repo":"risingwavelabs/risingwave","slug":"double-inserting-a-join-state-entry","errorCode":null,"errorMessage":"double inserting a join state entry","messagePattern":"double inserting a join state entry","errorType":"exception","errorClass":"JoinEntryError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/join/hash_join.rs","lineNumber":752,"sourceCode":"/// join key will be presented in the cache.\n#[derive(Default)]\npub struct JoinEntryState<E: JoinEncoding> {\n    /// The full copy of the state.\n    cached: JoinRowSet<PkType, E::EncodedRow>,\n    kv_heap_size: KvSize,\n}\n\nimpl<E: JoinEncoding> EstimateSize for JoinEntryState<E> {\n    fn estimated_heap_size(&self) -> usize {\n        // TODO: Add btreemap internal size.\n        // https://github.com/risingwavelabs/risingwave/issues/9713\n        self.kv_heap_size.size()\n    }\n}\n\n#[derive(Error, Debug)]\npub enum JoinEntryError {\n    #[error(\"double inserting a join state entry\")]\n    Occupied,\n    #[error(\"removing a join state entry but it is not in the cache\")]\n    Remove,\n}\n\nimpl<E: JoinEncoding> JoinEntryState<E> {\n    /// Insert into the cache.\n    pub fn insert(\n        &mut self,\n        key: PkType,\n        value: E::EncodedRow,\n    ) -> Result<&mut E::EncodedRow, JoinEntryError> {\n        let mut removed = false;\n        if !enable_strict_consistency() {\n            // strict consistency is off, let's remove existing (if any) first\n            if let Some(old_value) = self.cached.remove(&key) {\n                self.kv_heap_size.sub(&key, &old_value);\n                removed = true;","sourceCodeStart":734,"sourceCodeEnd":770,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/join/hash_join.rs#L734-L770","documentation":"JoinEntryError::Occupied is returned by JoinEntryState::insert when a join state entry for the same primary key already exists in the in-memory cache and strict consistency is enabled. The hash join expects each PK to be inserted at most once between removals; a second insert means upstream updates/duplicates are not balanced correctly. With strict consistency disabled, the code silently removes and re-inserts, logging a consistency_error instead.","triggerScenarios":"Calling JoinEntryState::insert (hash_join executor) with a key already present in the cache while enable_strict_consistency() is true — typically caused by receiving a duplicate join row without a matching prior removal (e.g. replayed upstream update, degree-tracking mismatch).","commonSituations":"Bugs in hash-join chunk application where an 'UpdateDelete' or degree-decrement path skipped removal; replaying the same upstream batch after a recovery; state cache and state table drifting out of sync under strict consistency testing.","solutions":["Check whether strict consistency mode is intentionally enabled; if this is a known duplicate-replay scenario, run with it disabled (the insert then overwrites).","Audit the chunk application path for a missing remove() before insert — every duplicate row arrival must be preceded by an UpdateDelete/removal.","Capture the failing PK and upstream chunk in logs and reproduce; this usually indicates a hash-join executor bug and should be reported upstream.","As a mitigation, clear/rebuild the join state table (full recovery) so cache and table re-sync."],"exampleFix":"// before (strict mode asserts uniqueness)\nlet ret = self.cached.try_insert(key.clone(), value);\n// after (overwrite semantics when strict consistency is off)\nif !enable_strict_consistency() {\n    if let Some(old) = self.cached.remove(&key) { self.kv_heap_size.sub(&key, &old); }\n}\nlet ret = self.cached.try_insert(key.clone(), value);","handlingStrategy":"validation","validationCode":"// before inserting, check the key is not already cached\nif join_entry_state.get(&key, &data_types).is_some() {\n    // either skip the insert or remove first (allowed only when strict consistency is off)\n    debug_assert!(!enable_strict_consistency(), \"double inserting a join state entry\");\n}","typeGuard":"fn entry_is_vacant<E: JoinEncoding>(state: &JoinEntryState<E>, key: &PkType, types: &[DataType]) -> bool {\n    state.get(key, types).is_none()\n}","tryCatchPattern":"match join_entry_state.insert(key, encoded_row) {\n    Err(JoinEntryError::Occupied) => {\n        // duplicate PK: overwrite or record a consistency issue instead of failing the actor\n        tracing::warn!(?key, \"duplicate join state insert; overwriting\");\n    }\n    other => other?,\n}","preventionTips":["Verify every UpdateInsert row has a matching UpdateDelete/removal in the chunk application path.","Run with strict consistency enabled in CI to surface double-inserts before production.","After recovery, always rebuild the cache from the state table before applying new chunks.","Log offending PKs to detect replaying or duplicating upstream sources."],"tags":["streaming","hash-join","state","rust"],"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"}