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.

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)

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 {} to source bac
• Can't create iceberg sink write result from empty data!
• iceberg sink metadata should be an object
• schema_id should be a u64
• Iceberg snapshot {snapshot_id} not found

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