risingwavelabs/risingwave · error

Unprocessed shared node.

Error message

Unprocessed shared node.

What it means

`pre` in MergeEqNodes prunes share nodes that have only one parent (or that share just a scan / no scan or source). Before counting, it looks up the share's share_id in the precomputed `counts` map with `expect("Unprocessed shared node.")`. The panic means the DAG traversal reached a share node whose refcount was never recorded, i.e. the counting pre-pass did not visit this share before `pre` consumed it.

Source

Thrown at src/frontend/src/optimizer/plan_node/merge_eq_nodes.rs:148

            .expect("dag cache is only used for shares")
            .share_id();
        self.cache.get(&share_id).cloned().unwrap_or_else(|| {
            let res = f(self);
            self.cache.entry(share_id).or_insert(res).clone()
        })
    }
}

impl Endo<PlanRef> for Pruner<'_> {
    fn pre(&mut self, t: PlanRef) -> PlanRef {
        let prunable = |s: &&LogicalShare| {
            // Prune if share node has only one parent
            // or it just shares a scan
            // or it doesn't share any scan or source.
            *self
                .counts
                .get(&s.share_id())
                .expect("Unprocessed shared node.")
                == 1
                || s.input().as_logical_scan().is_some()
                || !(plan_visitor::has_logical_scan(s.input())
                    || plan_visitor::has_logical_source(s.input()))
        };
        t.as_logical_share()
            .filter(prunable)
            .map_or(t.clone(), |s| self.pre(s.input()))
    }

    fn apply(&mut self, t: PlanRef) -> PlanRef {
        self.dag_apply(t)
    }
}

View on GitHub (pinned to 6469eb736d)

Solutions

1. Make sure the share-counting pass runs on the exact same plan instance immediately before dag_apply/pre.
2. Verify any custom rules that create LogicalShare nodes run before (or re-run) the counting pass.
3. Add the new share_id into `counts` whenever a share node is constructed, or rebuild counts after plan mutation.
4. Log the missing share_id at the call site to identify which rule produced the uncounted share.

Example fix

// before
let count = *self.counts.get(&s.share_id()).expect("Unprocessed shared node.");
// after
let count = match self.counts.get(&s.share_id()) {
    Some(c) => *c,
    None => {
        // share was not seen during counting; rebuild counts or treat as single-parent
        1
    }
};

// Rust: verify counts map coverage before rewriting
debug_assert!(self.counts.contains_key(&share.share_id()), "share id {} missing from counts", share.share_id());

fn counted(counts: &HashMap<ShareId, usize>, id: ShareId) -> Option<usize> { counts.get(&id).copied() }

// No exceptions in Rust; use Option fallback instead of expect.
let count = counted(&self.counts, s.share_id()).unwrap_or(1);

When it happens

Trigger: Running `pre` (from dag_apply) on a plan whose share refcount map was built for a different/older plan; a share node added after the counting pass; inconsistent plan identity between the counting traversal and the rewrite traversal.

Common situations: Custom optimizer rules that create new LogicalShare nodes after the count pre-pass; mutating the plan between counting and rewriting; rebasing or upgrading RisingWave and touching optimizer passes so the share counting step is skipped or reordered.

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

• BatchPosixFsReader should not hit this branch. refer to `bat
• internal error: entered unreachable code
• internal error: entered unreachable code
• Row sequential scan should not have input executor!
• Source should not have input executor!

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