risingwavelabs/risingwave · error

replacement sink catalog requires a sink job

Error message

replacement sink catalog requires a sink job

What it means

A CREATE SINK statement was issued with a replace target (replacing an existing sink catalog), but the new job being created is not a sink job. Replacement of a sink catalog is only valid when the replacement itself is a sink.

Source

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

            (None, DefaultParallelism::Full) => StreamingParallelism::Adaptive,
            (None, DefaultParallelism::Default(n)) => StreamingParallelism::Fixed(n.get()),
            (Some(n), _) => StreamingParallelism::Fixed(n.parallelism as _),
        };
        let backfill_parallelism = backfill_parallelism
            .as_ref()
            .map(|p| StreamingParallelism::Fixed(p.parallelism as _))
            .or_else(|| {
                backfill_adaptive_parallelism_strategy
                    .as_ref()
                    .map(|_| StreamingParallelism::Adaptive)
            });

        ensure_user_id(streaming_job.owner() as _, &txn).await?;
        ensure_object_id(ObjectType::Database, streaming_job.database_id(), &txn).await?;
        ensure_object_id(ObjectType::Schema, streaming_job.schema_id(), &txn).await?;
        if let Some(old_sink_id) = replace_sink {
            let StreamingJob::Sink(sink, _) = streaming_job else {
                bail!("replacement sink catalog requires a sink job")
            };
            let (old_sink, old_object) = Sink::find_by_id(*old_sink_id)
                .find_also_related(Object)
                .one(&txn)
                .await?
                .and_then(|(sink, object)| object.map(|object| (sink, object)))
                .ok_or_else(|| MetaError::catalog_id_not_found("sink", *old_sink_id))?;
            let old_streaming_job = StreamingJobModel::find_by_id(old_sink_id.as_job_id())
                .one(&txn)
                .await?
                .ok_or_else(|| MetaError::catalog_id_not_found("sink", *old_sink_id))?;
            if old_object.obj_type != ObjectType::Sink
                || old_object.database_id != Some(sink.database_id)
                || old_object.schema_id != Some(sink.schema_id)
                || old_sink.name != sink.name
            {
                bail!(
                    "old sink {} does not match replacement sink {}",

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure the replacement job is a sink (CREATE SINK ... REPLACE syntax path).
2. Check the frontend planner that resolves replacement targets is passing the correct job variant.
3. Remove the replace option if a non-sink object was intended and create it as a new job.

// frontend-side check before issuing replace
if !matches!(streaming_job, StreamingJob::Sink(..)) {
    // do not set replace_sink
}

fn is_sink_job(job: &StreamingJob) -> bool { matches!(job, StreamingJob::Sink(..)) }

match catalog.create_job_catalog(job).await {
    Err(e) if e.to_string().contains("requires a sink job") => /* clear replace option or send a sink job */,
    other => other?,
}

When it happens

Trigger: Invoking create_job_catalog with replace_sink set to Some(old_sink_id) while the streaming_job is a Mv/Table/MaterializedView variant instead of StreamingJob::Sink.

Common situations: Frontend planner bug or misuse of the replace-sink internal API where an `ALTER`/replace flow substitutes a non-sink relation.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• object {} is not a sink
• SinkError::Config(anyhow!(e))
• no state table id in sink: {}
• failed to create iceberg namespace: {namespace}
• Snowflake catalog only supports iceberg sources

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