risingwavelabs/risingwave · error

column id not found in sink

Error message

column id {} not found in sink

What it means

During sink distribution key planning, the meta node maps each requested distribution column (by column id) to its index among the sink's visible columns. This error is thrown when a column id present in the table's distribution key does not exist in the sink's column list, i.e. the sink schema and the table schema have diverged. It protects against building a sink whose downstream key columns reference nonexistent columns.

Solutions

1. Inspect the sink definition and ensure every distribution key column of the source table is included in the sink's columns (check the sink's SELECT/output list).
2. Recreate the sink so its schema is rebuilt from the current table schema: DROP SINK then CREATE SINK.
3. If a column was dropped/altered, update the table distribution key to reference existing columns before creating the sink.
4. If neither table nor sink was recently changed, check for meta-store schema corruption or a version upgrade mismatch and consult the RisingWave issue tracker.

Example fix

// before: sink output omits the distribution column
CREATE SINK s FROM t WITH (...) DISTRIBUTION KEY (id);
-- table t distributed by (id, tenant_id), sink dropped tenant_id

// after: include all distribution key columns in the sink
CREATE SINK s AS SELECT id, tenant_id, v FROM t WITH (...) DISTRIBUTION KEY (id, tenant_id);

-- before CREATE SINK, ensure every table distribution column appears in the sink output
SELECT c.name
FROM rw_catalog.rw_tables t,
     unnest(t.distribution_key) AS dk(col_id)
JOIN rw_catalog.rw_columns c ON c.table_id = t.id AND c.column_id = dk.col_id
WHERE t.name = 'my_table'
  AND c.name NOT IN (/* sink output column list */);

When it happens

Trigger: CREATE SINK / ALTER operations where the sink's downstream column_desc lacks a column id that the source table's distribution key references; typically after schema changes (dropped/renamed columns) between table and sink, or a partially planned sink descriptor.

Common situations: Altering a table's distribution key or dropping a column while a sink depends on it; creating a sink with an explicit distribution key on a column filtered out of the sink; version-skew where an older sink schema is stored in meta while the table was migrated.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• auto schema refresh sink must have only one fragment, but…
• Can't find data
• Cannot find
• Cannot find
• collection.name.field

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

Appendix: source

Thrown at src/meta/src/rpc/ddl_controller.rs:2542

            .iter()
            .enumerate()
            .map(|(idx, col)| {
                let column_id = col.column_desc.as_ref().unwrap().column_id;
                (column_id, idx as u32)
            })
            .collect::<HashMap<_, _>>();
        target_table
            .distribution_key
            .iter()
            .map(|dist_idx| {
                let column_id = target_table.columns[*dist_idx as usize]
                    .column_desc
                    .as_ref()
                    .unwrap()
                    .column_id;
                let sink_idx = sink_idx_by_col_id
                    .get(&column_id)
                    .ok_or_else(|| anyhow::anyhow!("column id {} not found in sink", column_id))?;
                Ok(*sink_idx)
            })
            .collect::<anyhow::Result<Vec<_>>>()?
    };
    let dist_key_indices =
        dist_key_indices.map_err(|e| e.context("failed to get distribution key indices"))?;
    let downstream_fragment_id = target_fragment_id as _;
    let new_downstream_relation = DownstreamFragmentRelation {
        downstream_fragment_id,
        dispatcher_type: DispatcherType::Hash,
        dist_key_indices,
        output_mapping: PbDispatchOutputMapping::simple(output_indices),
    };
    let current_target_columns = target_table.get_columns();
    let project_exprs = build_select_node_list(&sink_columns, current_target_columns)?;
    Ok(UpstreamSinkInfo {
        sink_id,
        sink_fragment_id: sink_fragment_id as _,

View on GitHub (pinned to 6469eb736d)