{"record":{"id":"d15c242bdff1a9dc","repo":"risingwavelabs/risingwave","slug":"should-always-have-a-stream-key-in-the-stream-plan","errorCode":null,"errorMessage":"should always have a stream key in the stream plan but not, sub plan: {}","messagePattern":"should always have a stream key in the stream plan but not, sub plan: (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/frontend/src/optimizer/plan_node/logical_union.rs","lineNumber":155,"sourceCode":"            Ok(BatchHashAgg::new(\n                generic::Agg::new(vec![], (0..self.base.schema().len()).collect(), batch_union)\n                    .with_enable_two_phase(false),\n            )\n            .into())\n        } else {\n            Ok(BatchUnion::new(new_logical).into())\n        }\n    }\n}\n\nimpl ToStream for LogicalUnion {\n    fn to_stream(\n        &self,\n        ctx: &mut ToStreamContext,\n    ) -> Result<crate::optimizer::plan_node::StreamPlanRef> {\n        // TODO: use round robin distribution instead of using hash distribution of all inputs.\n        let dist = RequiredDist::hash_shard(self.base.stream_key().unwrap_or_else(|| {\n            panic!(\n                \"should always have a stream key in the stream plan but not, sub plan: {}\",\n                PlanRef::from(self.clone()).explain_to_string()\n            )\n        }));\n        let new_inputs: Result<Vec<_>> = self\n            .inputs()\n            .iter()\n            .map(|input| input.to_stream_with_dist_required(&dist, ctx))\n            .collect();\n        let core = self.core.clone_with_inputs(new_inputs?);\n        assert!(\n            self.all(),\n            \"After UnionToDistinctRule, union should become union all\"\n        );\n        Ok(StreamUnion::new(core).into())\n    }\n\n    fn logical_rewrite_for_stream(","sourceCodeStart":137,"sourceCodeEnd":173,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/frontend/src/optimizer/plan_node/logical_union.rs#L137-L173","documentation":"In `LogicalUnion::to_stream`, the plan computes the required distribution from the union's stream key, and panics (`panic!` with the sub-plan explain text) if there is no stream key. Every stream subplan is expected to carry a stream key (unique key) so hash distribution can be derived; a missing key means an upstream `to_stream`/rewrite step produced a keyless plan, which is an invariant breach.","triggerScenarios":"Streaming conversion of a UNION (LogicalUnion) whose input(s) lack a derived stream key after `logical_rewrite_for_stream` — e.g. a union over relations that could not derive a unique key (keyless aggregation outputs, custom UDF sources, or non-deterministic expressions).","commonSituations":"Users hit this with `CREATE MATERIALIZED VIEW`/streaming queries that UNION keyless subqueries (e.g. aggregations without GROUP BY producing no unique key, or queries over sources without a primary key) where the frontend failed to append a key column.","solutions":["Rewrite the query so each UNION branch has a unique key: add GROUP BY keys to aggregations or include a primary-key/rowid column in each branch.","Union queries with distinct outputs (UNION vs UNION ALL) or add an explicit constant/row_number column to establish uniqueness.","Ensure all inputs have completed `logical_rewrite_for_stream` before union conversion; if stock code, file a RisingWave bug with the panicking plan text (it is included in the message).","Upgrade RisingWave — key-derivation fixes for keyless relations land regularly."],"exampleFix":"// before\nSELECT count(*) FROM t1 UNION ALL SELECT count(*) FROM t2; -- branches keyless\n// after\nSELECT count(*) AS c, 1 AS branch_key FROM t1\nUNION ALL\nSELECT count(*) AS c, 2 AS branch_key FROM t2;\n","handlingStrategy":"validation","validationCode":"// Ensure each UNION branch carries a unique key before streaming.\n-- check branches are not keyless aggregations like SELECT count(*) FROM t\n-- add GROUP BY or an explicit key column per branch","typeGuard":null,"tryCatchPattern":"match err if \"stream key\" in str(err):\n    # add key columns / GROUP BY to union branches and retry\n    query = add_synthetic_key(query)\n    retry(query)","preventionTips":["Avoid UNION over keyless subqueries (e.g. aggregations without GROUP BY).","Add a primary-key or constant/row_number key column to each UNION branch.","Prefer UNION ALL with explicit branch keys when uniqueness is unclear.","Run streaming queries against keyless relations in staging before production."],"tags":["rust","optimizer","streaming","stream-key","union"],"backgroundTag":"internal-invariant-violation","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"}