{"record":{"id":"daabceedce736dcf","repo":"risingwavelabs/risingwave","slug":"removing-a-join-state-entry-but-it-is-not-in-the-c","errorCode":null,"errorMessage":"removing a join state entry but it is not in the cache","messagePattern":"removing a join state entry but it is not in the cache","errorType":"exception","errorClass":"JoinEntryError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/join/hash_join.rs","lineNumber":754,"sourceCode":"pub 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;\n            }\n        }","sourceCodeStart":736,"sourceCodeEnd":772,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/join/hash_join.rs#L736-L772","documentation":"JoinEntryError::Remove is returned by JoinEntryState::remove when the primary key being deleted is not present in the join's in-memory cache while strict consistency is enabled. The join executor should only remove entries it previously inserted; deleting a missing key means delete/update bookkeeping has drifted from the cache contents. Without strict consistency the miss is only logged as a consistency_error and removal succeeds vacuously.","triggerScenarios":"Calling JoinEntryState::remove(pk) (hash_join executor) for a key absent from `cached` while enable_strict_consistency() is true — e.g. a duplicated UpdateDelete row, a delete applied twice, or a cache that was evicted/rebuilt while the state table still contains the key.","commonSituations":"Replayed delete rows after recovery; hash-join chunk application bugs where insert was skipped (e.g. duplicate update collapsed) but the corresponding delete still arrives; divergence between state-table contents and the in-memory cache under strict consistency validation.","solutions":["Check whether strict consistency is enabled for testing; disabling it downgrades this to a logged consistency_error.","Audit the join chunk application path for double-applied UpdateDelete or delete-after-delete sequences and fix the bookkeeping.","Perform a full recovery/rebuild of the join state so the cache is repopulated from the state table before processing resumes.","Report with the offending PK and input chunk — repeated occurrences indicate a hash-join executor bug."],"exampleFix":"// before (errors on missing key in strict mode)\nif self.cached.remove(&pk).is_none() { return Err(JoinEntryError::Remove); }\n// after (tolerate missing key, log consistency error)\nif let Some(value) = self.cached.remove(&pk) {\n    self.kv_heap_size.sub(&pk, &value);\n} else {\n    consistency_error!(?pk, \"removing a join state entry but it's not in the cache\");\n}","handlingStrategy":"validation","validationCode":"// before removing, ensure the key exists in the cache\nif join_entry_state.get(&pk, &data_types).is_none() {\n    debug_assert!(!enable_strict_consistency(), \"removing a join state entry but it is not in the cache\");\n}","typeGuard":"fn entry_exists<E: JoinEncoding>(state: &JoinEntryState<E>, pk: &PkType, types: &[DataType]) -> bool {\n    state.get(pk, types).is_some()\n}","tryCatchPattern":"match join_entry_state.remove(pk) {\n    Err(JoinEntryError::Remove) => {\n        // tolerate a duplicate delete under lenient mode; log for diagnosis\n        tracing::warn!(?pk, \"join state remove missed cache entry; ignoring\");\n    }\n    other => other?,\n}","preventionTips":["Ensure deletes are applied exactly once — guard against duplicated UpdateDelete rows in chunk application.","Pair every insert with its removal path; never remove a key whose insert was skipped.","Run strict-consistency CI tests to catch cache/state drift early.","After cache eviction/rebuild, re-sync from the state table before processing deletes."],"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"}