{"record":{"id":"1b4ccb42228b9e95","repo":"risingwavelabs/risingwave","slug":"creating-such-a-sink-will-result-in-circular-depen","errorCode":null,"errorMessage":"Creating such a sink will result in circular dependency.","messagePattern":"Creating such a sink will result in circular dependency\\.","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/controller/streaming_job.rs","lineNumber":623,"sourceCode":"                    backfill_adaptive_parallelism_strategy,\n                    refresh_interval_sec,\n                )\n                .await?;\n                table.id = streaming_job_model.job_id.as_mv_table_id();\n                let table_model: table::ActiveModel = table.clone().into();\n                Table::insert(table_model).exec(&txn).await?;\n                streaming_job_model\n            }\n            StreamingJob::Sink(sink, belong_to_table_id) => {\n                if let Some(target_table_id) = sink.target_table\n                    && check_sink_into_table_cycle(\n                        target_table_id.into(),\n                        dependencies.iter().cloned().collect(),\n                        &txn,\n                    )\n                    .await?\n                {\n                    bail!(\"Creating such a sink will result in circular dependency.\");\n                }\n\n                let streaming_job_model = Self::create_streaming_job_obj(\n                    &txn,\n                    ObjectType::Sink,\n                    sink.owner as _,\n                    Some(\n                        belong_to_table_id\n                            .map(|table_id| table_id.as_object_id())\n                            .unwrap_or(sink.schema_id.as_object_id()),\n                    ),\n                    create_type,\n                    ctx.clone(),\n                    adaptive_parallelism_strategy,\n                    streaming_parallelism,\n                    max_parallelism,\n                    resource_type.clone(),\n                    backfill_parallelism.clone(),","sourceCodeStart":605,"sourceCodeEnd":641,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/streaming_job.rs#L605-L641","documentation":"Creating a sink whose target table would create a cycle of dependencies among relations (the sink depends on a table that transitively depends on the sink) is forbidden. The catalog dependency check found a cycle before inserting the sink catalog.","triggerScenarios":"create_job_catalog for a sink whose target_table participates in an existing dependency cycle with the sink's upstream relations, detected by the dependency check over the txn.","commonSituations":"Users wiring sinks into tables that feed other MVs which the sink's source query reads (A -> B -> A pipelines), often from incremental schema evolution.","solutions":["Redesign the pipeline to break the cycle (split into separate tables or an external sink).","Drop one of the intermediate relations creating the cycle.","Create the sink into a fresh table not read by its own upstream chain.","Validate the dependency graph manually (rw_dependencies / upstream queries) before creating the sink."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// before creating the sink, ensure the target table is not read\n// (transitively) by the sink's source query\nlet cycle = check_relation_cycle(target_table_id, dependencies, &txn).await?;\nif cycle { /* abort creation */ }","typeGuard":null,"tryCatchPattern":"match create_sink_catalog(..).await {\n    Err(e) if e.to_string().contains(\"circular dependency\") => /* redesign pipeline; drop cycle link */,\n    other => other?,\n}","preventionTips":["Model pipelines as a DAG before creating sink-into-table","Check rw_dependencies before wiring sinks into tables read by the sink's sources"],"tags":["meta","sink","dependency-cycle"],"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"}