{"record":{"id":"533c974edbdc48e3","repo":"risingwavelabs/risingwave","slug":"column-not-found-in-sink-schema","errorCode":null,"errorMessage":"Column `{}` not found in sink schema","messagePattern":"Column `(.+?)` not found in sink schema","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/postgres.rs","lineNumber":503,"sourceCode":"            writer_param.sink_id, writer_param.actor_id\n        );\n        let client = create_pg_client(&pg_conn, tcp_keepalive, Some(&application_name)).await?;\n\n        ensure_no_foreign_key_with_client(&client, &config.schema, &config.table).await?;\n\n        // Rewrite schema types for serialization\n        let schema_types = {\n            let name_to_type = PostgresExternalTable::type_mapping(\n                &pg_conn,\n                &config.schema,\n                &config.table,\n                is_append_only,\n            )\n            .await?;\n            let mut schema_types = Vec::with_capacity(schema.fields.len());\n            for field in &schema.fields {\n                let actual_data_type = name_to_type.get(&field.name).cloned().ok_or_else(|| {\n                    SinkError::Config(anyhow!(\"Column `{}` not found in sink schema\", field.name))\n                })?;\n                schema_types.push(actual_data_type);\n            }\n            schema_types\n        };\n\n        let key_indices = if pk_indices.is_empty() {\n            (0..schema.len()).collect_vec()\n        } else {\n            pk_indices\n        };\n        let key_types = key_indices\n            .iter()\n            .map(|i| schema_types[*i].clone())\n            .collect_vec();\n\n        // validate() rejects out-of-range values at DDL time; clamp here so pre-existing sinks\n        // keep running after an upgrade.","sourceCodeStart":485,"sourceCodeEnd":521,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/postgres.rs#L485-L521","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the `primary_key` option names a column that exists in both the sink schema and the Postgres table.","Align the sink query's output column names with the Postgres table columns (use AS aliases).","Inspect the resolved schema and the PG table with \\d to find the name that diverges."],"exampleFix":"-- before\nCREATE SINK s FROM mv WITH (connector='postgres', table='t', primary_key='row_id'); -- t has no row_id\n-- after\nCREATE SINK s FROM mv WITH (connector='postgres', table='t', primary_key='id');","handlingStrategy":"validation","validationCode":"-- primary_key column must exist in the PG table before CREATE SINK\nSELECT 1 FROM information_schema.columns WHERE table_name='t' AND column_name='id';","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Verify primary_key names an existing column.","Keep sink query output names aligned with the PG table."],"tags":["rust","postgres","sink","schema"],"backgroundTag":"record-not-found","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"}