risingwavelabs/risingwave · critical

job fragments should exist for streaming job

Error message

job fragments should exist for streaming job

What it means

During CatalogController::into_database_contexts, the loaded fragment context is split per database. For every (job_id, database_id) pair in streaming_job_databases, job_fragments.remove(&job_id) is expected to yield the fragment map. Hitting this expect() panic means a streaming job is recorded in the job-to-database mapping but has no fragment rows in the LoadedFragmentContext built by the load stage.

Source

Thrown at src/meta/src/controller/scale.rs:369

        for (job_id, database_id) in streaming_job_databases {
            let context = contexts.entry(database_id).or_insert_with(|| {
                let database_model = database_map
                    .remove(&database_id)
                    .expect("database should exist for streaming job");
                Self {
                    ensembles: Vec::new(),
                    job_fragments: HashMap::new(),
                    job_map: HashMap::new(),
                    streaming_job_databases: HashMap::new(),
                    database_map: HashMap::from([(database_id, database_model)]),
                    fragment_source_ids: HashMap::new(),
                    fragment_splits: HashMap::new(),
                }
            });

            let fragments = job_fragments
                .remove(&job_id)
                .expect("job fragments should exist for streaming job");
            for fragment_id in fragments.keys().copied() {
                fragment_databases.insert(fragment_id, database_id);
                if let Some(source_id) = fragment_source_ids.remove(&fragment_id) {
                    context.fragment_source_ids.insert(fragment_id, source_id);
                }
                if let Some(splits) = fragment_splits.remove(&fragment_id) {
                    context.fragment_splits.insert(fragment_id, splits);
                }
            }

            assert!(
                context
                    .job_map
                    .insert(
                        job_id,
                        job_map
                            .remove(&job_id)
                            .expect("streaming job should exist for loaded context"),

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the job_id via the panic backtrace and check the meta catalog tables (fragments, streaming jobs) for that job; restore missing fragment rows from a backup or drop/recreate the job.
2. Verify the load stage (load_fragment_context / build_loaded_context) actually inserts every job in streaming_job_databases into job_fragments; fix the loader if a path skips fragments.
3. Check for concurrent job cancellation running during rescale; retry the rescale when no DDL is in flight.
4. If reproducible, file a bug with the meta store snapshot — this is an internal invariant violation.

Example fix

// before
let fragments = job_fragments
    .remove(&job_id)
    .expect("job fragments should exist for streaming job");
// after
let fragments = job_fragments.remove(&job_id).ok_or_else(|| {
    anyhow!("job fragments for streaming job {job_id} not found in loaded context")
})?;

// before calling into_database_contexts
for (job_id, _db) in &context.streaming_job_databases {
    assert!(context.job_fragments.contains_key(job_id), "job {job_id} lacks fragments");
}

fn has_fragments(ctx: &LoadedFragmentContext, job_id: JobId) -> bool { ctx.job_fragments.contains_key(&job_id) }

// Rust: panics cannot be caught; guard instead.
match context.job_fragments.get(&job_id) {
    Some(_) => split_ok,
    None => return Err(anyhow!("job fragments missing for {job_id} in loaded context")),
}

When it happens

Trigger: Calling into_database_contexts on a LoadedFragmentContext where build_loaded_context loaded a job whose streaming_job_databases entry exists but whose fragments were never inserted into job_fragments — e.g. the job's fragments were deleted concurrently, or the load stage assembled job_fragments incompletely relative to streaming_job_databases.

Common situations: Metadata inconsistency after a partially-failed scale/replace operation, manual DB edits to the catalog, or a bug in load_fragment_context that skips a job's fragments while still mapping the job to a database during rescale or recovery-driven rescheduling.

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

• streaming job should exist for loaded context
• expect fn
• logical join should store predicate as Condition
• core predicate must exist
• checked above

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