risingwavelabs/risingwave · error · SinkError::Config

Column count mismatch: Postgres table has {} columns, but si

Error message

Column count mismatch: Postgres table has {} columns, but sink schema has {} columns, sink should have less or equal columns to the Postgres table

What it means

During validation the sink connects to Postgres, fetches the target table's column descriptors, and compares the count with the RisingWave sink schema. The sink schema may not have MORE columns than the Postgres table (partial column mapping is allowed, extra sink columns are not). A mismatch aborts CREATE SINK.

Source

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

            let pg_table = PostgresExternalTable::connect(
                &pg_conn,
                &self.config.schema,
                &self.config.table,
                self.is_append_only,
                None,
            )
            .await
            .context(format!(
                "failed to connect to database: {}, schema: {}, table: {}",
                self.config.database, self.config.schema, self.config.table
            ))?;

            // Check that names and types match, order of columns doesn't matter.
            {
                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

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add the missing columns to the Postgres table (ALTER TABLE ... ADD COLUMN) so it has at least as many columns as the sink schema.
2. Remove the extra columns from the sink's SELECT list so it has <= the PG table's column count.
3. Verify you are sinking into the intended table (check `table` in the WITH options).

Example fix

-- before
CREATE SINK s FROM mv WITH (connector = 'postgres', table = 't'); -- mv has 5 cols, t has 3
-- after
ALTER TABLE t ADD COLUMN c4 int, ADD COLUMN c5 int;

SELECT count(*) FROM information_schema.columns WHERE table_name = 't';
-- must be >= number of columns in the sink schema

When it happens

Trigger: CREATE SINK whose SELECT/list of columns has more entries than the destination Postgres table has columns.

Common situations: The Postgres table was altered (columns dropped) after the sink was planned; sink query selects columns that do not exist in the PG table; wrong table name targeting a narrower table.

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 `{}` not found in Postgres table `{}`
• 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/4270ada57f4ca81f. Report an issue: GitHub.