{"record":{"id":"2f6d98ea4554a3b9","repo":"risingwavelabs/risingwave","slug":"creating-sink-into-table-is-not-allowed-for-unmigr","errorCode":null,"errorMessage":"Creating sink into table is not allowed for unmigrated table {}. Please migrate it first.","messagePattern":"Creating sink into table is not allowed for unmigrated table (.+?)\\. Please migrate it first\\.","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/rpc/ddl_controller.rs","lineNumber":1093,"sourceCode":"            let _enumerator = props\n                .create_split_enumerator(SourceEnumeratorContext::dummy().into())\n                .await?;\n\n            tracing::debug!(?table_id, \"validate cdc table success\");\n            Ok(true)\n        } else {\n            Ok(false)\n        }\n    }\n\n    pub async fn validate_table_for_sink(&self, table_id: TableId) -> MetaResult<()> {\n        let migrated = self\n            .metadata_manager\n            .catalog_controller\n            .has_table_been_migrated(table_id)\n            .await?;\n        if !migrated {\n            Err(anyhow::anyhow!(\"Creating sink into table is not allowed for unmigrated table {}. Please migrate it first.\", table_id).into())\n        } else {\n            Ok(())\n        }\n    }\n\n    /// For [`CreateType::Foreground`], the function will only return after backfilling finishes\n    /// ([`crate::manager::MetadataManager::wait_streaming_job_finished`]).\n    #[await_tree::instrument(boxed, \"create_streaming_job({streaming_job})\")]\n    pub async fn create_streaming_job(\n        &self,\n        mut streaming_job: StreamingJob,\n        fragment_graph: StreamFragmentGraphProto,\n        dependencies: HashSet<ObjectId>,\n        resource_type: streaming_job_resource_type::ResourceType,\n        if_not_exists: bool,\n        refresh_interval_sec: Option<u64>,\n        replace_sink: Option<SinkId>,\n        since_timestamp_epoch: Option<u64>,","sourceCodeStart":1075,"sourceCodeEnd":1111,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/rpc/ddl_controller.rs#L1075-L1111","documentation":"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.","triggerScenarios":"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.","commonSituations":"Clusters upgraded from versions without sink-into-table support; mixing old and new table formats; automation creating sinks onto legacy tables.","solutions":["Run the provided table migration command/SQL for the target table before creating the sink.","Recreate the table under the current version if migration tooling is unavailable.","Check table_id is correct — a stale ID pointing at a legacy table will keep failing."],"exampleFix":"// before\nCREATE SINK s INTO legacy_table AS SELECT ...;\n// after\nALTER TABLE legacy_table MIGRATE; -- migrate first\nCREATE SINK s INTO legacy_table AS SELECT ...;","handlingStrategy":"validation","validationCode":"// before CREATE SINK INTO\nlet migrated = catalog.has_table_been_migrated(table_id).await?;\nif !migrated {\n    run_table_migration(table_id).await?;\n}","typeGuard":null,"tryCatchPattern":"match create_sink(...).await {\n    Err(e) if e.to_string().contains(\"unmigrated table\") => {\n        migrate_table(table_id).await?; create_sink(...).await?;\n    }\n    other => other?,\n}","preventionTips":["Migrate legacy tables after upgrading before using sink-into-table","Check migration status of target tables in deployment scripts","Track which tables predate the sink-into-table feature"],"tags":["sink","migration","table"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}