risingwavelabs/risingwave · error · MetaError

sink fragment not found for sink id

Error message

sink fragment not found for sink id {}

What it means

When building upstream sink info for a sink (during sink into-table / sink upstream resolution), the code maps each incoming sink's sink_id to its fragment id via sink_fragment_ids; a missing mapping returns 'sink fragment not found for sink id {}'. The sink's fragment was never registered or was removed while its downstream sink still references it.

Solutions

1. Confirm the upstream sink exists and is fully created (SHOW SINKS) before creating the dependent sink
2. Retry the sink creation; a concurrent DROP SINK race usually resolves on retry
3. Check the table_fragments metadata for the sink fragment id mapping in the meta DB
4. If reproducible without concurrency, file a RisingWave issue with the sink id and job id

Example fix

// before
let sink_fragment_id = sink_fragment_ids.get(&sink.sink_id).cloned().ok_or(anyhow!("sink fragment not found for sink id {}", sink.sink_id))?;
// after
let sink_fragment_id = sink_fragment_ids.get(&sink.sink_id)
    .cloned()
    .ok_or_else(|| MetaError::catalog(rw_error::catalog::error::SinkNotFound::new(sink.sink_id).into()))?;

// before creating a chained sink, verify the upstream sink fragment exists
if !sink_fragment_ids.contains_key(&upstream_sink_id) { return Err(sink_not_ready(upstream_sink_id)); }

fn sink_fragment_registered(map: &HashMap<SinkId, FragmentId>, id: &SinkId) -> bool { map.contains_key(id) }

match create_sink(params).await { Err(e) if e.to_string().contains("sink fragment not found") => retry_with_backoff(params), other => other }

When it happens

Trigger: Creating a sink that reads from a table with sinks, when sink_fragment_ids (built from the table's fragment metadata) lacks an entry for one of the incoming sink_ids — e.g. the sink fragment was not yet persisted, or the sink was dropped concurrently.

Common situations: Chained sinks (sink into table, then sink from that table) created in quick succession; concurrent DROP SINK while creating the downstream sink; metadata from a failed/partial sink creation.

Understand the failure class

Background: Record Not Found Errors: "not found", RecordNotFound, and "was not found" — what they mean and how to fix them — this error's family across 28 libraries.

Related errors

• no fragment connection from source fragment
• Can't create iceberg sink write result from empty data!
• (dynamic: UTF-8 conversion error from metadata bytes)
• expected NoShuffle but got
• failed to find fragment

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

Appendix: source

Thrown at src/meta/src/controller/fragment.rs:2016

    ) -> MetaResult<Vec<UpstreamSinkInfo>>
    where
        C: ConnectionTrait,
    {
        let incoming_sinks = Sink::find()
            .filter(sink::Column::TargetTable.eq(target_table.id))
            .all(txn)
            .await?;

        let sink_ids = incoming_sinks.iter().map(|s| s.sink_id).collect_vec();
        let sink_fragment_ids = get_sink_fragment_by_ids(txn, sink_ids).await?;

        let mut upstream_sink_infos = Vec::with_capacity(incoming_sinks.len());
        for sink in &incoming_sinks {
            let sink_fragment_id =
                sink_fragment_ids
                    .get(&sink.sink_id)
                    .cloned()
                    .ok_or(anyhow::anyhow!(
                        "sink fragment not found for sink id {}",
                        sink.sink_id
                    ))?;
            let upstream_info = build_upstream_sink_info(
                sink.sink_id,
                sink.original_target_columns
                    .as_ref()
                    .map(|cols| cols.to_protobuf())
                    .unwrap_or_default(),
                sink_fragment_id,
                target_table,
                target_fragment_id,
            )?;
            upstream_sink_infos.push(upstream_info);
        }

        Ok(upstream_sink_infos)
    }

View on GitHub (pinned to 6469eb736d)