{"record":{"id":"4270ada57f4ca81f","repo":"risingwavelabs/risingwave","slug":"column-count-mismatch-postgres-table-has-colum","errorCode":null,"errorMessage":"Column count mismatch: Postgres table has {} columns, but sink schema has {} columns, sink should have less or equal columns to the Postgres table","messagePattern":"Column count mismatch: Postgres table has (.+?) columns, but sink schema has (.+?) columns, sink should have less or equal columns to the Postgres table","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/postgres.rs","lineNumber":270,"sourceCode":"            let pg_table = PostgresExternalTable::connect(\n                &pg_conn,\n                &self.config.schema,\n                &self.config.table,\n                self.is_append_only,\n                None,\n            )\n            .await\n            .context(format!(\n                \"failed to connect to database: {}, schema: {}, table: {}\",\n                self.config.database, self.config.schema, self.config.table\n            ))?;\n\n            // Check that names and types match, order of columns doesn't matter.\n            {\n                let pg_columns = pg_table.column_descs();\n                let sink_columns = self.schema.fields();\n                if pg_columns.len() < sink_columns.len() {\n                    return Err(SinkError::Config(anyhow!(\n                        \"Column count mismatch: Postgres table has {} columns, but sink schema has {} columns, sink should have less or equal columns to the Postgres table\",\n                        pg_columns.len(),\n                        sink_columns.len()\n                    )));\n                }\n\n                let pg_columns_lookup = pg_columns\n                    .iter()\n                    .map(|c| (c.name.clone(), c.data_type.clone()))\n                    .collect::<BTreeMap<_, _>>();\n                for sink_column in sink_columns {\n                    let pg_column = pg_columns_lookup.get(&sink_column.name);\n                    match pg_column {\n                        None => {\n                            return Err(SinkError::Config(anyhow!(\n                                \"Column `{}` not found in Postgres table `{}`\",\n                                sink_column.name,\n                                self.config.table","sourceCodeStart":252,"sourceCodeEnd":288,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/postgres.rs#L252-L288","documentation":"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.","triggerScenarios":"CREATE SINK whose SELECT/list of columns has more entries than the destination Postgres table has columns.","commonSituations":"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.","solutions":["Add the missing columns to the Postgres table (ALTER TABLE ... ADD COLUMN) so it has at least as many columns as the sink schema.","Remove the extra columns from the sink's SELECT list so it has <= the PG table's column count.","Verify you are sinking into the intended table (check `table` in the WITH options)."],"exampleFix":"-- before\nCREATE SINK s FROM mv WITH (connector = 'postgres', table = 't'); -- mv has 5 cols, t has 3\n-- after\nALTER TABLE t ADD COLUMN c4 int, ADD COLUMN c5 int;","handlingStrategy":"validation","validationCode":"SELECT count(*) FROM information_schema.columns WHERE table_name = 't';\n-- must be >= number of columns in the sink schema","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Compare PG table columns with the sink SELECT list before CREATE SINK.","Watch for ALTER TABLE on the target table invalidating existing sink configs."],"tags":["rust","postgres","sink","schema"],"backgroundTag":"schema-validation-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}