risingwavelabs/risingwave · error · SinkError::Config

Column `{}` not found in Postgres table `{}`

Error message

Column `{}` not found in Postgres table `{}`

What it means

Each column name in the sink schema must exist in the destination Postgres table. Validation builds a BTreeMap lookup of PG column names and fails as soon as a sink column name is absent, since there would be no target column to write into.

Source

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

                let pg_columns = pg_table.column_descs();
                let sink_columns = self.schema.fields();
                if pg_columns.len() < sink_columns.len() {
                    return Err(SinkError::Config(anyhow!(
                        "Column count mismatch: Postgres table has {} columns, but sink schema has {} columns, sink should have less or equal columns to the Postgres table",
                        pg_columns.len(),
                        sink_columns.len()
                    )));
                }

                let pg_columns_lookup = pg_columns
                    .iter()
                    .map(|c| (c.name.clone(), c.data_type.clone()))
                    .collect::<BTreeMap<_, _>>();
                for sink_column in sink_columns {
                    let pg_column = pg_columns_lookup.get(&sink_column.name);
                    match pg_column {
                        None => {
                            return Err(SinkError::Config(anyhow!(
                                "Column `{}` not found in Postgres table `{}`",
                                sink_column.name,
                                self.config.table
                            )));
                        }
                        Some(pg_column) => {
                            if !validate_pg_type_to_rw_type(pg_column, &sink_column.data_type()) {
                                return Err(SinkError::Config(anyhow!(
                                    "Column `{}` in Postgres table `{}` has type `{}`, but sink schema defines it as type `{}`",
                                    sink_column.name,
                                    self.config.table,
                                    pg_column,
                                    sink_column.data_type()
                                )));
                            }
                        }
                    }
                }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Rename the column in the sink query (AS <pg_column_name>) to match an existing Postgres column.
2. Add the missing column to the Postgres table with the exact name (quote it if it contains uppercase).
3. Check identifier case: Postgres folds unquoted names to lowercase; use quoted identifiers consistently on both sides.

Example fix

-- before
CREATE SINK s AS SELECT user_id AS uid FROM mv WITH (connector='postgres', table='t');
-- after
CREATE SINK s AS SELECT user_id AS user_id FROM mv WITH (connector='postgres', table='t');

SELECT column_name FROM information_schema.columns WHERE table_name = 't';
-- every sink output column name must appear here

When it happens

Trigger: CREATE SINK selects or renames a column (e.g. via AS alias) that does not exist in the Postgres table specified by `table`.

Common situations: Case-sensitivity mismatches (unquoted PG identifiers are lowercased); renamed column in RW query; PG table schema drifted from what the sink was designed against.

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

• Column count mismatch: Postgres table has {} columns, but si
• Column `{}` not found in sink schema
• 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/0f6f7245a47078b2. Report an issue: GitHub.