risingwavelabs/risingwave · error

should exist

Error message

should exist

What it means

`fetch_incoming_sinks` calls `expect("should exist")` after looking up each incoming sink ID in the target schema's catalog. The ID came from the schema's own `incoming_sinks` list, so the sink must exist; if it does not, the catalog metadata is inconsistent (e.g. during concurrent DROP SINK or after a partial deletion), and the handler panics instead of returning an error.

Solutions

1. Retry the operation — transient races with DROP SINK usually resolve once catalogs resync.
2. Ensure sinks are dropped/created outside of concurrent table operations in scripts.
3. Check catalog consistency (`SHOW SINKS`) and recreate dangling sink references if any.
4. Internal fix: return a `catalog` error for missing sink IDs instead of `expect`.

Example fix

// before
sinks.push(schema.get_sink_by_id(*sink_id).expect("should exist").clone());
// after
let sink = schema.get_sink_by_id(*sink_id)
    .ok_or_else(|| anyhow!("sink {:?} not found in catalog", sink_id))?;
sinks.push(sink.clone());

// retry transient catalog races
for (let i = 0; i < 3; i++) {
  try { return await fetchIncomingSinks(tableId); }
  catch (e) {
    if (String(e).includes('should exist') && i < 2) { await sleep(500); continue; }
    throw e;
  }
}

When it happens

Trigger: Listing/fetching downstream sinks of a table (`fetch_incoming_sinks`, used by e.g. table drop checks or SHOW) while a sink referencing the table was just dropped, leaving a stale entry in `incoming_sinks`; meta/catalog cache out of sync with the schema catalog.

Common situations: Race between `DROP SINK` and operations that enumerate incoming sinks (like `DROP TABLE` dependency checks); catalog cache staleness in the frontend after meta node failover.

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

• database should exist for streaming job
• failed to create iceberg namespace
• failed to decode persisted `DefaultColumnDesc`
• IcebergSinkWriter should be initialized before barrier
• no state table id in sink

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

Appendix: source

Thrown at src/frontend/src/handler/create_sink.rs:946

    Ok(SinkCreateMode::Replace { original_sink })
}

pub fn fetch_incoming_sinks(
    session: &Arc<SessionImpl>,
    table: &TableCatalog,
) -> Result<Vec<Arc<SinkCatalog>>> {
    let reader = session.env().catalog_reader().read_guard();
    let schema = reader.get_schema_by_id(table.database_id, table.schema_id)?;
    let Some(incoming_sinks) = schema.table_incoming_sinks(table.id) else {
        return Ok(vec![]);
    };
    let mut sinks = vec![];
    for sink_id in incoming_sinks {
        sinks.push(
            schema
                .get_sink_by_id(*sink_id)
                .expect("should exist")
                .clone(),
        );
    }
    Ok(sinks)
}

fn derive_sink_to_table_expr(
    sink_schema: &Schema,
    idx: usize,
    target_type: &DataType,
) -> Result<ExprImpl> {
    let input_type = &sink_schema.fields()[idx].data_type;

    if !target_type.equals_datatype(input_type) {
        bail!(
            "column type mismatch: {:?} vs {:?}, column name: {:?}",
            target_type,
            input_type,

View on GitHub (pinned to 6469eb736d)