risingwavelabs/risingwave · error · anyhow::Error

no state table id in sink

Error message

no state table id in sink: {}

What it means

When a committed-epoch subscriber is created for a sink, the meta node looks up the sink's state table IDs and requires at least one to subscribe its committed epoch in Hummock. A sink with no state table cannot be coordinated for commit epochs, so the subscriber task returns this error and the sink coordination fails to start.

Solutions

1. Inspect the sink catalog (`rw_catalog.rw_sinks` / meta metadata) to confirm the sink has an associated state (internal) table; recreate the sink if the state table is missing.
2. If the sink id belongs to a dropped sink, guard against coordinating already-removed sinks and clean up stale coordination state.
3. Ensure DDL completes atomically — re-run the sink creation if a previous `CREATE SINK` failed midway.
4. Upgrade if migrating from a version where some sink types had no state table.

Example fix

// before
let Some(table_id) = state_table_ids.first() else {
    return Err(anyhow!("no state table id in sink: {}", sink_id).into());
};
// after
let Some(table_id) = state_table_ids.first() else {
    return Err(SinkError::from(anyhow!(
        "no state table id in sink: {}; sink may be corrupted or dropped, recreate the sink",
        sink_id
    )));
};

// Before starting sink coordination, verify the sink has a state table
let state_table_ids = metadata_manager.get_sink_state_table_ids(sink_id).await?;
if state_table_ids.is_empty() {
    return Err(anyhow!("sink {} has no state table; recreate the sink", sink_id));
}

match new_committed_epoch_subscriber(sink_id).await {
    Err(e) if e.to_string().contains("no state table id") => {
        tracing::error!("sink {} is missing its state table; check catalog or recreate", sink_id);
    }
    other => { /* normal handling */ }
}

When it happens

Trigger: `start_worker` spawns `new_committed_epoch_subscriber`; `get_sink_state_table_ids(sink_id)` returns an empty list for the given sink id.

Common situations: Sink created without a state table (corrupt/incomplete catalog state, e.g. interrupted DDL or migration from an older format); querying a sink id that was dropped concurrently; append-only sinks of a legacy version that lacked state tables.

Understand the failure class

Background: EmptyResultError / "no results found": when an API or scraper succeeds but returns zero rows — this error's family across 9 libraries.

Related errors

• job in database has no state table after recovery
• object is not a sink
• replacement sink catalog requires a sink job
• SinkError::Config(anyhow!(e))
• state table is not registered to hummock

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

Appendix: source

Thrown at src/meta/src/manager/sink_coordination/manager.rs:90

#[derive(Clone)]
pub struct SinkCoordinatorManager {
    request_tx: mpsc::Sender<ManagerRequest>,
}
fn new_committed_epoch_subscriber(
    hummock_manager: HummockManagerRef,
    metadata_manager: MetadataManager,
) -> SinkCommittedEpochSubscriber {
    Arc::new(move |sink_id| {
        let hummock_manager = hummock_manager.clone();
        let metadata_manager = metadata_manager.clone();
        async move {
            let state_table_ids = metadata_manager
                .get_sink_state_table_ids(sink_id)
                .await
                .map_err(SinkError::from)?;
            let Some(table_id) = state_table_ids.first() else {
                return Err(anyhow!("no state table id in sink: {}", sink_id).into());
            };
            hummock_manager
                .subscribe_table_committed_epoch(*table_id)
                .await
                .map_err(SinkError::from)
        }
        .boxed()
    })
}

impl SinkCoordinatorManager {
    pub fn start_worker(
        db: DatabaseConnection,
        hummock_manager: HummockManagerRef,
        metadata_manager: MetadataManager,
        iceberg_compact_stat_sender: UnboundedSender<IcebergSinkCompactionUpdate>,
        await_tree_reg: await_tree::Registry,
    ) -> (Self, (JoinHandle<()>, Sender<()>)) {

View on GitHub (pinned to 6469eb736d)