risingwavelabs/risingwave · error · MetaError

iceberg pk-index writer input must be a merge after actor re

Error message

iceberg pk-index writer input must be a merge after actor rewrite

What it means

During actor rewriting of the stream graph, an Iceberg primary-key index writer node's inputs must be rewritten into Merge nodes so `allow_empty_upstream` can be set. If a rewritten input is not a `NodeBody::Merge`, the invariant that pk-index writers always consume merge inputs after rewriting is broken, and the meta service bails out. This is an internal consistency check in the frontend/meta actor-rewrite pass.

Source

Thrown at src/meta/src/stream/stream_graph/actor.rs:206

                ];
                Ok(StreamNode {
                    input,
                    ..stream_node.clone()
                })
            }

            NodeBody::IcebergWithPkIndexWriter(_) => {
                let mut new_stream_node = stream_node.clone();
                // Compaction alternates between the normal and resolver inputs, so either merge
                // must be able to remain alive while temporarily disconnected.
                for (input, new_input) in stream_node
                    .input
                    .iter()
                    .zip_eq_fast(&mut new_stream_node.input)
                {
                    *new_input = self.rewrite_inner(input, depth + 1)?;
                    let Some(NodeBody::Merge(merge)) = new_input.node_body.as_mut() else {
                        bail!("iceberg pk-index writer input must be a merge after actor rewrite");
                    };
                    merge.allow_empty_upstream = true;
                }
                Ok(new_stream_node)
            }

            // For other nodes, visit the children recursively.
            _ => {
                let mut new_stream_node = stream_node.clone();
                for (input, new_input) in stream_node
                    .input
                    .iter()
                    .zip_eq_fast(&mut new_stream_node.input)
                {
                    *new_input = self.rewrite_inner(input, depth + 1)?;
                }
                Ok(new_stream_node)
            }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the fragment/plan topology feeding the Iceberg pk-index writer and ensure its input is a Merge node before actor rewrite.
2. Check recent changes to `rewrite_inner` or NodeBody handling in src/meta/src/stream/stream_graph/actor.rs and fix the rewrite so the input becomes a Merge.
3. Verify the frontend does not build pk-index writers for Iceberg sinks with unusual upstreams; adjust plan generation instead of relaxing this check.
4. If reproducible, capture the plan (explain/graph dump) and file an issue — this is an internal invariant violation, not a user-facing config error.

Example fix

// before
let Some(NodeBody::Merge(merge)) = new_input.node_body.as_mut() else {
    bail!("iceberg pk-index writer input must be a merge after actor rewrite");
};
// after
match new_input.node_body.as_mut() {
    Some(NodeBody::Merge(merge)) => merge.allow_empty_upstream = true,
    other => return Err(anyhow::anyhow!(
        "iceberg pk-index writer input must be a merge after actor rewrite, got {:?}",
        other.map(|b| b.kind())
    )),
}

// Rust, before calling rewrite on the graph
fn input_is_merge_after_rewrite(node: &StreamNode) -> bool {
    node.input.iter().all(|i| matches!(i.node_body, Some(NodeBody::Merge(_))))
}

fn as_merge(node: &mut StreamNode) -> Option<&mut Merge> {
    match node.node_body.as_mut() {
        Some(NodeBody::Merge(m)) => Some(m),
        _ => None,
    }
}

match actor::rewrite(graph) {
    Ok(rewritten) => deploy(rewritten),
    Err(e) if e.to_string().contains("must be a merge") => {
        tracing::error!("plan topology bug around iceberg pk-index writer: {e:#}");
        // surface as internal error, do not retry blindly
    }
    Err(e) => return Err(e),
}

When it happens

Trigger: Calling the stream graph actor rewrite (`rewrite`/`rewrite_inner` in src/meta/src/stream/stream_graph/actor.rs) on a plan containing an Iceberg pk-index writer whose upstream node does not rewrite to a Merge node body.

Common situations: Developing a new stream node type or changing node rewriting logic so a pk-index writer is wired to a non-merge input; using an internal/dev build where fragment planning produced an unexpected topology around Iceberg sinks with primary keys; version mismatch between frontend plan generation and meta rewrite expectations.

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

• only iceberg table can be altered as sink
• unexpected referring object type: {}
• iceberg pk-index sink coordinator for sink {} is not registe
• there should be no ExchangeNode on the top of the plan node:
• failed to align vnode count for table {}({}): required {}, b

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