IcebergWithPkIndexWriterExecutor requires exactly two inputs

Error message

IcebergWithPkIndexWriterExecutor requires exactly two inputs

What it means

The `IcebergWithPkIndexWriterExecutor` is a two-input operator: a main input plus a dedicated resolver (compaction) input. During executor construction, `params.input.try_into()` into an exact two-element array fails when the frontend attached any other number of inputs, and the code maps that failure to this anyhow error. It protects against malformed or version-skewed plan fragments.

Source

Thrown at src/stream/src/from_proto/iceberg_with_pk_index/writer.rs:49

use crate::executor::{Executor, IcebergWriterImpl, StreamExecutorError, WriterExecutor};
use crate::from_proto::ExecutorBuilder;
use crate::task::ExecutorParams;
pub struct IcebergWithPkIndexWriterExecutorBuilder;

impl_stream_node_body!(IcebergWithPkIndexWriter(IcebergWithPkIndexWriterNode) => IcebergWithPkIndexWriterExecutorBuilder);

impl ExecutorBuilder for IcebergWithPkIndexWriterExecutorBuilder {
    type Node = IcebergWithPkIndexWriterNode;

    async fn new_boxed_executor(
        params: ExecutorParams,
        node: &Self::Node,
        store: impl StateStore,
    ) -> StreamResult<Executor> {
        let [input, resolver_input] = params
            .input
            .try_into()
            .map_err(|_| anyhow!("IcebergWithPkIndexWriterExecutor requires exactly two inputs"))?;
        let sink_desc = node.sink_desc.as_ref().unwrap();
        let sink_id: SinkId = sink_desc.get_id();
        let sink_name = sink_desc.get_name().to_owned();

        let properties_with_secret = LocalSecretManager::global().fill_secrets(
            sink_desc.get_properties().clone(),
            sink_desc.get_secret_refs().clone(),
        )?;
        let config = IcebergConfig::from_btreemap(properties_with_secret.clone())
            .map_err(|err| StreamExecutorError::from((err, sink_id)))?;

        let pk_indices = sink_desc
            .downstream_pk
            .iter()
            .map(|&idx| idx as usize)
            .collect::<Vec<_>>();
        if pk_indices.is_empty() {
            bail!("missing downstream pk in iceberg sink desc");

View on GitHub (pinned to 6469eb736d)

Solutions

Regenerate/re-plan the streaming job so the frontend emits both the writer input and the resolver input.
Ensure frontend and compute-node binaries are from the same version (upgrade together).
Inspect the serialized plan fragment (actor proto) to confirm the writer node has exactly two inputs.
If reproducible with matching versions, file a planner bug — the frontend should never produce wrong arity.

Example fix

// before
let [input] = params.input.try_into().unwrap(); // wrong arity for this executor
// after
let [input, resolver_input] = params.input.try_into().map_err(|_| anyhow!("IcebergWithPkIndexWriterExecutor requires exactly two inputs"))?;

Defensive patterns

Strategy: validation

Validate before calling

// verify plan arity before dispatch to the builder
if node.input.len() != 2 { return Err(anyhow!("expected writer + resolver inputs")); }

Type guard

fn two_inputs(inputs: &[StreamNode]) -> Option<(&StreamNode, &StreamNode)> {
    match inputs { [a, b] => Some((a, b)), _ => None }
}

Try / catch

let [input, resolver_input] = params.input.try_into().map_err(|_| anyhow!("IcebergWithPkIndexWriterExecutor requires exactly two inputs"))?;

Prevention

Keep frontend and compute binaries on the same version.
Add planner tests asserting the pk writer node always has two inputs.
When replaying plan fragments, validate arity before executing.

When it happens

Trigger: `new_boxed_executor` receives a `StreamNode` for the Iceberg pk writer whose `input` list has fewer or more than two children (e.g. one input because the resolver edge was dropped, or three after a buggy rewrite).

Common situations: Frontend/backend version skew where the plan serializer omitted the resolver input; hand-crafted or replayed plan fragments; internal planner bugs producing the wrong arity for this sink node.

Related errors

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