What causes "old sink {} does not match replacement sink {}" in risingwavelabs/risingwave?

Make the replacement sink use the same name, database, and schema as the sink being replaced.

old sink {} does not match replacement sink {}

Error message

old sink {} does not match replacement sink {}

What it means

When replacing a sink, the old sink must actually match the replacement: same object type (Sink), same database and schema, and same name. This error means the old sink id resolved to an object that differs from the replacement sink on one of those attributes.

Source

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

            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 {}",
                    old_sink_id,
                    sink.name
                );
            }
            if old_sink.target_table.is_some() || sink.target_table.is_some() {
                bail!("replace sink into table is not supported");
            }
            if old_streaming_job.job_status != JobStatus::Created {
                bail!("sink {} is not ready to be replaced", old_sink_id);
            }
        } else {
            check_relation_name_duplicate(
                &streaming_job.name(),
                streaming_job.database_id(),
                streaming_job.schema_id(),
                &txn,
            )

View on GitHub (pinned to 6469eb736d)

Solutions

Make the replacement sink use the same name, database, and schema as the sink being replaced.
Verify the old sink id is correct (query rw_catalog / system tables for the sink).
If a different name/location is desired, drop the old sink and create a new one instead of replacing.

Example fix

-- before
CREATE SINK orders_sink2 AS SELECT ... REPLACE orders_sink;
-- after
CREATE SINK orders_sink AS SELECT ... REPLACE orders_sink;

Defensive patterns

Strategy: validation

Validate before calling

// verify match before replacing
SELECT obj_type, database_id, schema_id, name
FROM rw_catalog rw_objects
WHERE object_id = <old_sink_id>;
-- ensure name/db/schema equal the replacement sink

Try / catch

if let Err(e) = catalog.create_job_catalog(job).await {
    if e.to_string().contains("does not match replacement sink") {
        // re-resolve old sink id by name from the catalog and retry
    }
}

Prevention

Always reference the replace target by its current name/db/schema
Refresh the frontend catalog cache before replace DDL

When it happens

Trigger: CREATE SINK ... REPLACE (or create_job_catalog with replace_sink) where the old sink lives in a different database/schema, has a different name, or is not a sink.

Common situations: User renamed the sink in the replacement statement but pointed at the old id; sink moved schemas; typo in the replace target; stale frontend catalog cache.

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

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