risingwavelabs/risingwave · error · JoinEntryError

double inserting a join state entry

Error message

double inserting a join state entry

What it means

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.

Source

Thrown at src/stream/src/executor/join/hash_join.rs:752

/// join key will be presented in the cache.
#[derive(Default)]
pub struct JoinEntryState<E: JoinEncoding> {
    /// The full copy of the state.
    cached: JoinRowSet<PkType, E::EncodedRow>,
    kv_heap_size: KvSize,
}

impl<E: JoinEncoding> EstimateSize for JoinEntryState<E> {
    fn estimated_heap_size(&self) -> usize {
        // TODO: Add btreemap internal size.
        // https://github.com/risingwavelabs/risingwave/issues/9713
        self.kv_heap_size.size()
    }
}

#[derive(Error, Debug)]
pub enum JoinEntryError {
    #[error("double inserting a join state entry")]
    Occupied,
    #[error("removing a join state entry but it is not in the cache")]
    Remove,
}

impl<E: JoinEncoding> JoinEntryState<E> {
    /// Insert into the cache.
    pub fn insert(
        &mut self,
        key: PkType,
        value: E::EncodedRow,
    ) -> Result<&mut E::EncodedRow, JoinEntryError> {
        let mut removed = false;
        if !enable_strict_consistency() {
            // strict consistency is off, let's remove existing (if any) first
            if let Some(old_value) = self.cached.remove(&key) {
                self.kv_heap_size.sub(&key, &old_value);
                removed = true;

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check whether strict consistency mode is intentionally enabled; if this is a known duplicate-replay scenario, run with it disabled (the insert then overwrites).
2. Audit the chunk application path for a missing remove() before insert — every duplicate row arrival must be preceded by an UpdateDelete/removal.
3. Capture the failing PK and upstream chunk in logs and reproduce; this usually indicates a hash-join executor bug and should be reported upstream.
4. As a mitigation, clear/rebuild the join state table (full recovery) so cache and table re-sync.

Example fix

// before (strict mode asserts uniqueness)
let ret = self.cached.try_insert(key.clone(), value);
// after (overwrite semantics when strict consistency is off)
if !enable_strict_consistency() {
    if let Some(old) = self.cached.remove(&key) { self.kv_heap_size.sub(&key, &old); }
}
let ret = self.cached.try_insert(key.clone(), value);

// before inserting, check the key is not already cached
if join_entry_state.get(&key, &data_types).is_some() {
    // either skip the insert or remove first (allowed only when strict consistency is off)
    debug_assert!(!enable_strict_consistency(), "double inserting a join state entry");
}

fn entry_is_vacant<E: JoinEncoding>(state: &JoinEntryState<E>, key: &PkType, types: &[DataType]) -> bool {
    state.get(key, types).is_none()
}

match join_entry_state.insert(key, encoded_row) {
    Err(JoinEntryError::Occupied) => {
        // duplicate PK: overwrite or record a consistency issue instead of failing the actor
        tracing::warn!(?key, "duplicate join state insert; overwriting");
    }
    other => other?,
}

When it happens

Trigger: 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).

Common situations: 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.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• removing a join state entry but it is not in the cache
• next offset {:?} should be later than current offset {:?}
• new item epoch {} does not match current chunk offset epoch
• new item epoch {} does not exceed barrier offset epoch {}
• Division by zero

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/4d210ec571f3f0c9. Report an issue: GitHub.