risingwavelabs/risingwave · error

non duplicate

Error message

non duplicate

What it means

This is an `.expect("non duplicate")` panic inside `split_fragment_map_by_database`, fired by `HashMap::try_insert` when the same fragment_id appears twice for the same database. The library asserts that the input fragment map has no duplicate fragment ids; duplicates indicate a corrupted or doubly-built map.

Solutions

1. Deduplicate the fragment map before calling: build it with `collect()` into a HashMap, which silently overwrites, or explicitly reject duplicates upstream.
2. Check the upstream producer of fragment ids for double-emission (e.g. merged iterators or repeated table scans).
3. Inspect meta-node catalog storage for duplicate fragment rows and repair them.
4. If duplicates are expected/benign, replace `try_insert(..).expect(..)` with overwrite semantics.

Example fix

// before
ret.entry(database_id).or_default().try_insert(fragment_id, value).expect("non duplicate");
// after
ret.entry(database_id).or_default().insert(fragment_id, value); // overwrite duplicates instead of panicking

// reject duplicate fragment ids up front
let mut seen = std::collections::HashSet::new();
for id in fragment_map.keys() {
    if !seen.insert(*id) { return Err(anyhow!("duplicate fragment id {}", id)); }
}

// this path panics (expect), so it cannot be caught; validate inputs beforehand
assert_unique_fragment_ids(&fragment_map)?;

When it happens

Trigger: Passing a fragment map (or upstream iterator) that contains the same fragment_id more than once — e.g. joining/merging fragment lists from multiple sources that overlap, or a catalog read returning duplicate rows.

Common situations: Buggy aggregation of per-table fragment maps; catalog scans with duplicate records after meta-node state restore; tests constructing fixture maps by hand that repeat fragment ids.

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

• failed to parse relation definition
• Failed to retrieve fragment description: fragment
• internal error: entered unreachable code
• invalid parallelism
• job fragments should exist for streaming job

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

Appendix: source

Thrown at src/meta/src/manager/metadata.rs:352

            .list_fragment_database_ids(Some(
                fragment_map
                    .keys()
                    .map(|fragment_id| *fragment_id as _)
                    .collect(),
            ))
            .await?
            .into_iter()
            .map(|(fragment_id, database_id)| (fragment_id as FragmentId, database_id))
            .collect();
        let mut ret: HashMap<_, HashMap<_, _>> = HashMap::new();
        for (fragment_id, value) in fragment_map {
            let database_id = *fragment_to_database_map
                .get(&fragment_id)
                .ok_or_else(|| anyhow!("cannot get database_id of fragment {fragment_id}"))?;
            ret.entry(database_id)
                .or_default()
                .try_insert(fragment_id, value)
                .expect("non duplicate");
        }
        Ok(ret)
    }

    pub async fn list_creating_jobs(&self) -> MetaResult<HashSet<JobId>> {
        Ok(self
            .catalog_controller
            .list_creating_jobs(false, None)
            .await?
            .into_iter()
            .map(|(job_id, _, _, _, _)| job_id)
            .collect())
    }

    pub async fn list_sources(&self) -> MetaResult<Vec<PbSource>> {
        self.catalog_controller.list_sources().await
    }

View on GitHub (pinned to 6469eb736d)