risingwavelabs/risingwave · error · anyhow::Error

obj has no database id

Error message

obj has no database id: {:?}

What it means

When aborting a creating streaming job, the object row associated with the job has no database_id set, which is required for cleanup. This is an internal invariant violation: every streaming job object must record its database.

Solutions

1. Inspect the object row for the job in the metadata store; this is data corruption, not user-fixable via SQL.
2. Restore the meta store from backup.
3. Drop the orphaned object/job records manually if recovery is acceptable.
4. File a bug with the job_id from the message.

fn has_database_id(obj: &Object) -> Option<DatabaseId> { obj.database_id }

match catalog.try_abort_creating_streaming_job(job_id).await {
    Err(e) if e.to_string().contains("obj has no database id") => /* meta-store corruption: restore from backup; do not retry */,
    other => other?,
}

When it happens

Trigger: Calling try_abort_creating_streaming_job for a job whose Object model row has database_id = None.

Common situations: Corrupted metadata after partial write/failed migration, manual metadata-store edits, or bugs in object creation.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• catalog object has no database
• id not found
• named already exists
• batch refresh job has no snapshot backfill info
• cannot find subscription with id

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

Appendix: source

Thrown at src/meta/src/controller/streaming_job.rs:1045

        let mut inner = self.inner.write().await;
        let txn = inner.db.begin().await?;

        let obj = Object::find_by_id(job_id).one(&txn).await?;
        let Some(obj) = obj else {
            tracing::warn!(
                id = %job_id,
                "streaming job not found when aborting creating, might be cancelled already or cleaned by recovery"
            );
            return Ok(AbortCreatingJobResult {
                aborted: true,
                database_id: None,
                aborted_sink_ids: vec![],
                cancel_info: None,
            });
        };
        let database_id = obj
            .database_id
            .ok_or_else(|| anyhow!("obj has no database id: {:?}", obj))?;
        let streaming_job = streaming_job::Entity::find_by_id(job_id).one(&txn).await?;

        if let Some(streaming_job) = &streaming_job {
            if streaming_job.job_status == JobStatus::Created {
                tracing::warn!(%job_id, "streaming job is already created, ignore abort request");
                return Ok(AbortCreatingJobResult {
                    aborted: false,
                    database_id: Some(database_id),
                    aborted_sink_ids: vec![],
                    cancel_info: None,
                });
            }

            if !is_cancelled && streaming_job.job_status == JobStatus::Creating {
                if (obj.obj_type == ObjectType::Table || obj.obj_type == ObjectType::Sink)
                    && check_if_belongs_to_iceberg_table(&txn, job_id).await?
                {
                    // If the job belongs to an Iceberg table, we still need to clean it.

View on GitHub (pinned to 6469eb736d)