risingwavelabs/risingwave · error

should always have a stream key in the stream plan but not,

Error message

should always have a stream key in the stream plan but not, sub plan: {}

What it means

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.

Source

Thrown at src/frontend/src/optimizer/plan_node/logical_union.rs:155

            Ok(BatchHashAgg::new(
                generic::Agg::new(vec![], (0..self.base.schema().len()).collect(), batch_union)
                    .with_enable_two_phase(false),
            )
            .into())
        } else {
            Ok(BatchUnion::new(new_logical).into())
        }
    }
}

impl ToStream for LogicalUnion {
    fn to_stream(
        &self,
        ctx: &mut ToStreamContext,
    ) -> Result<crate::optimizer::plan_node::StreamPlanRef> {
        // TODO: use round robin distribution instead of using hash distribution of all inputs.
        let dist = RequiredDist::hash_shard(self.base.stream_key().unwrap_or_else(|| {
            panic!(
                "should always have a stream key in the stream plan but not, sub plan: {}",
                PlanRef::from(self.clone()).explain_to_string()
            )
        }));
        let new_inputs: Result<Vec<_>> = self
            .inputs()
            .iter()
            .map(|input| input.to_stream_with_dist_required(&dist, ctx))
            .collect();
        let core = self.core.clone_with_inputs(new_inputs?);
        assert!(
            self.all(),
            "After UnionToDistinctRule, union should become union all"
        );
        Ok(StreamUnion::new(core).into())
    }

    fn logical_rewrite_for_stream(

View on GitHub (pinned to 6469eb736d)

Solutions

1. 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.
2. Union queries with distinct outputs (UNION vs UNION ALL) or add an explicit constant/row_number column to establish uniqueness.
3. 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).
4. Upgrade RisingWave — key-derivation fixes for keyless relations land regularly.

Example fix

// before
SELECT count(*) FROM t1 UNION ALL SELECT count(*) FROM t2; -- branches keyless
// after
SELECT count(*) AS c, 1 AS branch_key FROM t1
UNION ALL
SELECT count(*) AS c, 2 AS branch_key FROM t2;

// Ensure each UNION branch carries a unique key before streaming.
-- check branches are not keyless aggregations like SELECT count(*) FROM t
-- add GROUP BY or an explicit key column per branch

match err if "stream key" in str(err):
    # add key columns / GROUP BY to union branches and retry
    query = add_synthetic_key(query)
    retry(query)

When it happens

Trigger: 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).

Common situations: 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.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• update should always be converted to batch plan
• upsert stream is not supported as input of {}, plan: {}
• next offset {:?} should be later than current offset {:?}
• new item epoch {} does not match current chunk offset epoch
• new item epoch {} does not exceed barrier offset epoch {}

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/d15c242bdff1a9dc. Report an issue: GitHub.