What causes "missing downstream pk in iceberg sink desc" in risingwavelabs/risingwave?

Define the sink's upstream with an explicit PRIMARY KEY so the frontend populates `downstream_pk`.

missing downstream pk in iceberg sink desc

Error message

missing downstream pk in iceberg sink desc

What it means

An Iceberg sink with primary key requires a non-empty `downstream_pk` in its sink description, because pk-index maintenance (compaction resolver, position-delete handling) keys off the primary key column indices. `new_boxed_executor` bails when `pk_indices` collected from `sink_desc.downstream_pk` is empty, treating the plan fragment as corrupt.

Source

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

            .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");
        }

        let (sink_param, _) = build_sink_param(sink_desc, properties_with_secret, ICEBERG_SINK)?;

        let table = create_and_validate_table_impl(&config, &sink_param)
            .await
            .map_err(|e| StreamExecutorError::sink_error(e, sink_id))?;

        let pk_index_state_table = StateTableBuilder::new(
            node.get_pk_index_table()?,
            store,
            params.vnode_bitmap.clone().map(Arc::new),
        )
        .enable_preload_all_rows_by_config(&params.config)
        .with_op_consistency_level(StateTableOpConsistencyLevel::Inconsistent)
        .build()
        .await;

View on GitHub (pinned to 6469eb736d)

Solutions

Define the sink's upstream with an explicit PRIMARY KEY so the frontend populates `downstream_pk`.
Re-create the sink with a current frontend version that fills `downstream_pk` for pk sinks.
Verify the sink desc proto in the actor plan actually contains downstream_pk entries.
If the sink genuinely has no key, use the plain (non-pk) Iceberg sink path instead of the pk-index writer.

Example fix

-- before
CREATE SINK s FROM mv WITH (connector='iceberg', ...); -- upstream has no PK
-- after
CREATE MATERIALIZED VIEW mv AS SELECT id, ... FROM t; -- id PRIMARY KEY
CREATE SINK s FROM mv WITH (connector='iceberg', primary_key='id', ...);

Defensive patterns

Strategy: validation

Validate before calling

-- SQL-level guard
-- ensure the upstream has a PRIMARY KEY before creating a pk Iceberg sink
SELECT count(*) FROM rw_catalog.rw_columns WHERE relation = 'mv' AND is_primary_key; -- must be > 0

Type guard

fn has_downstream_pk(desc: &SinkDesc) -> bool { !desc.downstream_pk.is_empty() }

Try / catch

if pk_indices.is_empty() { bail!("missing downstream pk in iceberg sink desc"); }

Prevention

Always declare PRIMARY KEY on the source table/MV feeding a pk Iceberg sink.
Include `primary_key` in the sink's WITH options consistently.
Add frontend assertion that the pk writer path is only used with non-empty downstream_pk.

When it happens

Trigger: Creating an Iceberg pk-index sink whose proto `sink_desc.downstream_pk` is empty or unset — e.g. the sink was defined without a primary key but routed to the pk-index writer node, or the frontend failed to fill downstream_pk.

Common situations: Creating `CREATE SINK ... ` on a table/MV without a PRIMARY KEY but hitting the pk writer path; version skew where older frontends did not populate `downstream_pk`; manual proto editing or corrupted plan fragments.

Understand the failure class

Background: "must not be empty", "cannot be empty" — required-field validation errors across open-source libraries — this error's family across 41 libraries.

Related errors

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