{"record":{"id":"aeab69d821da0d6b","repo":"risingwavelabs/risingwave","slug":"column-in-postgres-table-has-type","errorCode":null,"errorMessage":"Column `{}` in Postgres table `{}` has type `{}`, but sink schema defines it as type `{}`","messagePattern":"Column `(.+?)` in Postgres table `(.+?)` has type `(.+?)`, but sink schema defines it as type `(.+?)`","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/postgres.rs","lineNumber":293,"sourceCode":"                }\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\n                            )));\n                        }\n                        Some(pg_column) => {\n                            if !validate_pg_type_to_rw_type(pg_column, &sink_column.data_type()) {\n                                return Err(SinkError::Config(anyhow!(\n                                    \"Column `{}` in Postgres table `{}` has type `{}`, but sink schema defines it as type `{}`\",\n                                    sink_column.name,\n                                    self.config.table,\n                                    pg_column,\n                                    sink_column.data_type()\n                                )));\n                            }\n                        }\n                    }\n                }\n            }\n\n            // check that pk matches\n            {\n                let pg_pk_names = pg_table.pk_names();\n                let sink_pk_names = self\n                    .pk_indices\n                    .iter()","sourceCodeStart":275,"sourceCodeEnd":311,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/postgres.rs#L275-L311","documentation":"After matching names, validate_pg_type_to_rw_type checks that each sink column's RisingWave type is compatible with the Postgres column's type. If the PG column's type cannot be mapped/validated against the RW type, creation fails with this message showing both types.","triggerScenarios":"Sink schema declares e.g. an INTEGER for a PG column of type text/bytea/timestamp, or vice versa — any pairing validate_pg_type_to_rw_type rejects.","commonSituations":"PG table column altered to a different type after sink design; casting mistakes in the CREATE SINK query (e.g. sinking a varchar into an int column); using serial/bigserial columns with mismatched RW integer widths.","solutions":["CAST the column in the sink query to a type compatible with the Postgres column: SELECT col::int FROM ....","ALTER the Postgres column type (ALTER TABLE ... ALTER COLUMN ... TYPE ...) to match the sink schema.","Compare the PG type printed in the error with the RW type and pick a pair validate_pg_type_to_rw_type accepts (e.g. int4<->int, int8<->bigint, text<->varchar)."],"exampleFix":"-- before\nCREATE SINK s AS SELECT id::varchar AS id FROM mv WITH (connector='postgres', table='t'); -- t.id is int4\n-- after\nCREATE SINK s AS SELECT id::int AS id FROM mv WITH (connector='postgres', table='t');","handlingStrategy":"validation","validationCode":"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = 't';\n-- cast sink columns to matching types, e.g. col::int","typeGuard":null,"tryCatchPattern":null,"preventionTips":["CAST explicitly in the sink query rather than relying on implicit mapping.","Re-validate sinks after ALTER TABLE ... TYPE changes."],"tags":["rust","postgres","sink","type-mismatch"],"backgroundTag":"type-mismatch","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"}