{"record":{"id":"57920caf5166d592","repo":"risingwavelabs/risingwave","slug":"dropping-sink-into-table-is-not-allowed-for-unmigr","errorCode":null,"errorMessage":"Dropping sink into table is not allowed for unmigrated table {}. Please migrate it first.","messagePattern":"Dropping sink into table is not allowed for unmigrated table (.+?)\\. Please migrate it first\\.","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/controller/catalog/drop_op.rs","lineNumber":124,"sourceCode":"            },\n        };\n\n        removed_objects.push(obj);\n        let mut removed_object_ids: HashSet<_> =\n            removed_objects.iter().map(|obj| obj.oid).collect();\n\n        for obj in &removed_objects {\n            if obj.obj_type == ObjectType::Sink {\n                let sink = Sink::find_by_id(obj.oid.as_sink_id())\n                    .one(&txn)\n                    .await?\n                    .ok_or_else(|| MetaError::catalog_id_not_found(\"sink\", obj.oid))?;\n\n                if let Some(target_table) = sink.target_table\n                    && !removed_object_ids.contains(&target_table.as_object_id())\n                    && !has_table_been_migrated(&txn, target_table).await?\n                {\n                    return Err(anyhow::anyhow!(\n                        \"Dropping sink into table is not allowed for unmigrated table {}. Please migrate it first.\",\n                        target_table\n                    ).into());\n                }\n            }\n        }\n\n        // Load all objects that belong to the dropped objects before deletion. Cascaded rows are\n        // still needed for notifications and resource cleanup.\n        let root_objects = Object::find()\n            .filter(object::Column::Oid.is_in(removed_object_ids.iter().copied()))\n            .all(&txn)\n            .await?;\n        let belonging_objects =\n            get_belong_objects_by_ids(&txn, removed_objects.iter().map(|obj| obj.oid)).await?;\n        removed_object_ids.extend(belonging_objects.iter().map(|obj| obj.oid));\n        let mut objects_to_remove = root_objects.clone();\n        objects_to_remove.extend(belonging_objects.iter().cloned());","sourceCodeStart":106,"sourceCodeEnd":142,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/catalog/drop_op.rs#L106-L142","documentation":"During drop_object, when dropping a sink whose target is a table, RisingWave checks that the target table is either being removed in the same transaction or has already been migrated (has_table_been_migrated). If the sink targets an unmigrated table, the drop is rejected with this error to avoid breaking the legacy table-sink coupling that the migration process is meant to resolve.","triggerScenarios":"DROP SINK on a sink created with 'INTO TABLE' where the target table still exists, is not part of the current removal batch (removed_object_ids), and has_table_been_migrated returns false.","commonSituations":"Upgrading clusters that require sink-into-table migration; attempting to clean up legacy sinks before running the required migration step; scripted teardown dropping sinks while their tables remain.","solutions":["Run the table migration for the target table first (as the message says), then drop the sink.","Drop the target table and the sink together in the same operation/batch.","Recreate the sink against a migrated (or external) target instead of the legacy table."],"exampleFix":"// before\nDROP SINK my_sink; -- fails: table unmigrated\n// after\n-- migrate the table first, then:\nDROP SINK my_sink;","handlingStrategy":"validation","validationCode":"-- before DROP SINK, confirm the target table is migrated or will be dropped too\n-- SQL: inspect sink targets and table state\nSHOW SINKS; -- identify sinks into tables\n-- ensure the table is migrated per your upgrade runbook before dropping the sink","typeGuard":null,"tryCatchPattern":"match drop_result {\n    Err(e) if e.to_string().contains(\"unmigrated table\") => {\n        run_table_migration(target_table)?;\n        drop_result = drop_sink(name).await; // retry once after migration\n    }\n    other => other?,\n}","preventionTips":["Run required sink-into-table migrations as part of upgrade runbooks before DDL cleanup.","Drop dependent sinks and their target tables in the same batch.","Inventory legacy table-sinks after version upgrades and migrate them proactively."],"tags":["meta","sink","migration","ddl"],"backgroundTag":"unsupported-operation","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}