risingwavelabs/risingwave · error

upstream fragment not found in initial upstream actors

Error message

upstream fragment {} not found in initial upstream actors

What it means

Thrown in `UpstreamSinkUnion::new` when the `upstream_fragment_id` supplied by the meta service has no entry in the `initial_upstream_actors` map. The executor cannot build its merge inputs without knowing which upstream actors to read from. This indicates the fragment passed in was never provisioned in the actor's upstream fragment list, i.e. an inconsistency between the stream plan and the actor deployment info.

Solutions

1. Check that the sink node's upstream fragment IDs in the stream plan are present in the actor's `initial_upstream_actors` map (proto field) before dispatch.
2. Upgrade meta and stream nodes together; a version skew can cause the producer of `initial_upstream_actors` to omit new fragment types.
3. Re-create the affected materialization/sink so the fragment graph is regenerated and redeployed.
4. If reproducible, file a bug with the fragment graph: this is an internal plan-consistency violation.

Example fix

// before
let actors = initial_upstream_actors.get(&upstream_fragment_id).ok_or_else(...)?;
// after
// Ensure the fragment id exists before constructing:
assert!(
    initial_upstream_actors.contains_key(&upstream_fragment_id),
    "upstream fragment {upstream_fragment_id} missing from initial_upstream_actors"
);

// before dispatching the actor
assert!(
    initial_upstream_actors.contains_key(&upstream_fragment_id),
    "upstream fragment {upstream_fragment_id} missing"
);

fn fragment_exists(map: &HashMap<u32, _>, id: u32) -> bool { map.contains_key(&id) }

// Rust: this is an anyhow error surfaced as StreamResult; handle at actor-build level
match UpstreamSinkUnion::new(...) {
    Ok(e) => e,
    Err(err) if err.to_string().contains("not found in initial upstream actors") => {
        // fail actor creation and report plan inconsistency to meta
        return Err(err);
    }
}

When it happens

Trigger: Calling `UpstreamSinkUnion::new` with an `upstream_fragment_id` that is absent from the `initial_upstream_actors` HashMap handed to the constructor, typically during actor creation from a StreamFragmentGraph / actor deployment where the sink's upstream fragment was not recorded.

Common situations: Meta-node bugs where sink upstream fragments are pruned or not serialized into the actor proto; a stale or partially upgraded cluster where new executor code expects fragments older dispatch code never populated; hand-modified or corrupted fragment graph during recovery.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• iceberg pk-index writer
• iceberg pk-index writer
• invalid state, new_count = 0 and is_new_entry is true
• left barrier received while right stream end
• lhs and rhs chunk cardinality should be the same

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

Appendix: source

Thrown at src/stream/src/executor/upstream_sink_union.rs:136

pub struct UpstreamFragmentInfo {
    pub upstream_fragment_id: FragmentId,
    pub upstream_actors: Vec<PbActorInfo>,
    pub merge_schema: Schema,
    pub project_exprs: Vec<NonStrictExpression>,
}

impl UpstreamFragmentInfo {
    pub fn new(
        upstream_fragment_id: FragmentId,
        initial_upstream_actors: &HashMap<FragmentId, UpstreamActors>,
        sink_output_schema: &[PbField],
        project_exprs: &[PbExprNode],
        error_report: impl EvalErrorReport + 'static,
    ) -> StreamResult<Self> {
        let actors = initial_upstream_actors
            .get(&upstream_fragment_id)
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "upstream fragment {} not found in initial upstream actors",
                    upstream_fragment_id
                )
            })?;
        let merge_schema = sink_output_schema.iter().map(Field::from).collect();
        let project_exprs = project_exprs
            .iter()
            .map(|e| build_non_strict_from_prost(e, error_report.clone()))
            .try_collect()
            .map_err(|err| anyhow::anyhow!(err))?;
        Ok(Self {
            upstream_fragment_id,
            upstream_actors: actors.actors.clone(),
            merge_schema,
            project_exprs,
        })
    }
}

View on GitHub (pinned to 6469eb736d)