risingwavelabs/risingwave · error · SinkError::Config

Column `{}` not found in sink schema

Error message

Column `{}` not found in sink schema

What it means

In `new()`, the sink maps each field of the (possibly derived) sink schema to the actual Postgres column type via a name_to_type lookup built from the PG table. If a schema field's name has no corresponding entry, creation fails. Despite the message text, this is reached when the schema field name is absent from the PG name->type map used here.

Source

Thrown at src/connector/src/sink/postgres.rs:503

            writer_param.sink_id, writer_param.actor_id
        );
        let client = create_pg_client(&pg_conn, tcp_keepalive, Some(&application_name)).await?;

        ensure_no_foreign_key_with_client(&client, &config.schema, &config.table).await?;

        // Rewrite schema types for serialization
        let schema_types = {
            let name_to_type = PostgresExternalTable::type_mapping(
                &pg_conn,
                &config.schema,
                &config.table,
                is_append_only,
            )
            .await?;
            let mut schema_types = Vec::with_capacity(schema.fields.len());
            for field in &schema.fields {
                let actual_data_type = name_to_type.get(&field.name).cloned().ok_or_else(|| {
                    SinkError::Config(anyhow!("Column `{}` not found in sink schema", field.name))
                })?;
                schema_types.push(actual_data_type);
            }
            schema_types
        };

        let key_indices = if pk_indices.is_empty() {
            (0..schema.len()).collect_vec()
        } else {
            pk_indices
        };
        let key_types = key_indices
            .iter()
            .map(|i| schema_types[*i].clone())
            .collect_vec();

        // validate() rejects out-of-range values at DDL time; clamp here so pre-existing sinks
        // keep running after an upgrade.

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure the `primary_key` option names a column that exists in both the sink schema and the Postgres table.
2. Align the sink query's output column names with the Postgres table columns (use AS aliases).
3. Inspect the resolved schema and the PG table with \d to find the name that diverges.

Example fix

-- before
CREATE SINK s FROM mv WITH (connector='postgres', table='t', primary_key='row_id'); -- t has no row_id
-- after
CREATE SINK s FROM mv WITH (connector='postgres', table='t', primary_key='id');

-- primary_key column must exist in the PG table before CREATE SINK
SELECT 1 FROM information_schema.columns WHERE table_name='t' AND column_name='id';

When it happens

Trigger: Creating the sink when a field in the resolved schema (including the configured primary_key column) has no matching entry in the name_to_type map built from the Postgres table.

Common situations: primary_key option references a column not present in the table/schema; schema derivation after joins/renames producing column names absent from the PG table.

Understand the failure class

Background: Record Not Found Errors: "not found", RecordNotFound, and "was not found" — what they mean and how to fix them — this error's family across 28 libraries.

Related errors

• Column count mismatch: Postgres table has {} columns, but si
• Column `{}` not found in Postgres table `{}`
• If you want to use upsert, please set the keysType of doris
• Can't find data
• Cannot find {}

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