{"record":{"id":"4e00082df9f0e037","repo":"risingwavelabs/risingwave","slug":"sink-with-auto-schema-change-should-have-only-1-fr","errorCode":null,"errorMessage":"sink with auto schema change should have only 1 fragment, but got {:?}","messagePattern":"sink with auto schema change should have only 1 fragment, but got (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/fragment.rs","lineNumber":401,"sourceCode":"fn clone_fragment(fragment: &Fragment, id_generator_manager: &IdGeneratorManager) -> Fragment {\n    let fragment_id = GlobalFragmentIdGen::new(id_generator_manager, 1)\n        .to_global_id(0)\n        .as_global_id();\n    Fragment {\n        fragment_id,\n        fragment_type_mask: fragment.fragment_type_mask,\n        distribution_type: fragment.distribution_type,\n        state_table_ids: fragment.state_table_ids.clone(),\n        maybe_vnode_count: fragment.maybe_vnode_count,\n        nodes: fragment.nodes.clone(),\n    }\n}\n\npub fn check_sink_fragments_support_refresh_schema(\n    fragments: &BTreeMap<FragmentId, Fragment>,\n) -> MetaResult<()> {\n    if fragments.len() != 1 {\n        return Err(anyhow!(\n            \"sink with auto schema change should have only 1 fragment, but got {:?}\",\n            fragments.len()\n        )\n        .into());\n    }\n    let (_, fragment) = fragments.first_key_value().expect(\"non-empty\");\n    let sink_node = &fragment.nodes;\n    let PbNodeBody::Sink(_) = sink_node.node_body.as_ref().unwrap() else {\n        return Err(anyhow!(\"expect PbNodeBody::Sink but got: {:?}\", sink_node.node_body).into());\n    };\n    let [stream_input_node] = sink_node.input.as_slice() else {\n        panic!(\"Sink has more than 1 input: {:?}\", sink_node.input);\n    };\n    let stream_scan_node = match stream_input_node.node_body.as_ref().unwrap() {\n        PbNodeBody::StreamScan(_) => stream_input_node,\n        PbNodeBody::Project(_) => {\n            let [stream_scan_node] = stream_input_node.input.as_slice() else {\n                return Err(anyhow!(","sourceCodeStart":383,"sourceCodeEnd":419,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/fragment.rs#L383-L419","documentation":"`check_sink_fragments_support_refresh_schema` verifies that a sink with automatic schema change (schema refresh) consists of exactly one fragment, since schema-evolution handling is only implemented for single-fragment sinks. If the generated sink job has multiple fragments, creation fails with the fragment count reported.","triggerScenarios":"Creating a sink with auto schema change enabled whose streaming job plans into more than one fragment (`check_sink_fragments_support_refresh_schema`, fragment.rs:401, called from `generate_streaming_job`).","commonSituations":"Sinks with complex downstream transformations (e.g. with extra operators forcing multiple fragments) combined with `auto_schema_change = true`; Iceberg sinks with connectors that split the plan into several fragments; older/newer frontend behavior differences around sink fragmentation.","solutions":["Disable auto schema change on the sink (` = false`) if multi-fragment topology is required.","Simplify the sink definition so it plans into a single fragment (remove extra transformations between source and sink).","Check the sink connector's plan generation for why multiple fragments are produced and use the supported single-fragment layout."],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector = 'iceberg', auto_schema_change = true); -- plans to 2 fragments\n// after\nCREATE SINK s FROM mv WITH (connector = 'iceberg'); -- or ensure single-fragment plan with auto_schema_change = true","handlingStrategy":"validation","validationCode":"-- SQL: check sink plan fragment count before enabling auto schema change\n-- EXPLAIN CREATE SINK ... ; ensure a single sink fragment when auto_schema_change = true","typeGuard":null,"tryCatchPattern":"// Rust: result of generate_streaming_job\nmatch generate_streaming_job(...) {\n    Ok(job) => job,\n    Err(e) if e.to_string().contains(\"should have only 1 fragment\") => {\n        Err(anyhow!(\"disable auto schema change or simplify sink to one fragment: {e:#}\"))\n    }\n    Err(e) => Err(e),\n}","preventionTips":["Enable auto schema change only on sinks known to plan into one fragment.","Check EXPLAIN output for fragment count before enabling schema refresh.","Keep sink definitions simple (avoid extra transformations) when using auto_schema_change."],"tags":["sink","schema-change","meta","validation"],"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"}