risingwavelabs/risingwave · error

Creating sink into table is not allowed for unmigrated table

Error message

Creating sink into table is not allowed for unmigrated table {}. Please migrate it first.

What it means

Creating a sink whose target is a table is only allowed after the target table has been migrated to the sink-into-table compatible layout. validate_table_for_sink checks has_table_been_migrated and rejects sinks pointing at unmigrated tables, instructing the user to migrate first.

Solutions

1. Run the provided table migration command/SQL for the target table before creating the sink.
2. Recreate the table under the current version if migration tooling is unavailable.
3. Check table_id is correct — a stale ID pointing at a legacy table will keep failing.

Example fix

// before
CREATE SINK s INTO legacy_table AS SELECT ...;
// after
ALTER TABLE legacy_table MIGRATE; -- migrate first
CREATE SINK s INTO legacy_table AS SELECT ...;

// before CREATE SINK INTO
let migrated = catalog.has_table_been_migrated(table_id).await?;
if !migrated {
    run_table_migration(table_id).await?;
}

match create_sink(...).await {
    Err(e) if e.to_string().contains("unmigrated table") => {
        migrate_table(table_id).await?; create_sink(...).await?;
    }
    other => other?,
}

When it happens

Trigger: CREATE SINK INTO <table> where the target table_id has not been migrated (has_table_been_migrated returns false), typically an older table created before the sink-into-table migration feature existed.

Common situations: Clusters upgraded from versions without sink-into-table support; mixing old and new table formats; automation creating sinks onto legacy tables.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Dropping sink into table is not allowed for unmigrated table
• ALTER SINK_RATE_LIMIT is not for sink into table
• ambiguous auth: multiple auth options provided; remove one…
• auth.method=key_pair_file must not set `password`
• auth.method=key_pair_file must not set `private_key_pem`

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

Appendix: source

Thrown at src/meta/src/rpc/ddl_controller.rs:1093

            let _enumerator = props
                .create_split_enumerator(SourceEnumeratorContext::dummy().into())
                .await?;

            tracing::debug!(?table_id, "validate cdc table success");
            Ok(true)
        } else {
            Ok(false)
        }
    }

    pub async fn validate_table_for_sink(&self, table_id: TableId) -> MetaResult<()> {
        let migrated = self
            .metadata_manager
            .catalog_controller
            .has_table_been_migrated(table_id)
            .await?;
        if !migrated {
            Err(anyhow::anyhow!("Creating sink into table is not allowed for unmigrated table {}. Please migrate it first.", table_id).into())
        } else {
            Ok(())
        }
    }

    /// For [`CreateType::Foreground`], the function will only return after backfilling finishes
    /// ([`crate::manager::MetadataManager::wait_streaming_job_finished`]).
    #[await_tree::instrument(boxed, "create_streaming_job({streaming_job})")]
    pub async fn create_streaming_job(
        &self,
        mut streaming_job: StreamingJob,
        fragment_graph: StreamFragmentGraphProto,
        dependencies: HashSet<ObjectId>,
        resource_type: streaming_job_resource_type::ResourceType,
        if_not_exists: bool,
        refresh_interval_sec: Option<u64>,
        replace_sink: Option<SinkId>,
        since_timestamp_epoch: Option<u64>,

View on GitHub (pinned to 6469eb736d)