{"record":{"id":"3eb40b10f02b14bc","repo":"risingwavelabs/risingwave","slug":"must-be-or-sink-type-option-must-be-a","errorCode":null,"errorMessage":"`{}` must be {}, or {} (SINK_TYPE_OPTION must be append-only or upsert)","messagePattern":"`(.+?)` must be (.+?), or (.+?) \\(SINK_TYPE_OPTION must be append-only or upsert\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/postgres.rs","lineNumber":156,"sourceCode":"\n    if has_foreign_key {\n        return Err(SinkError::Config(anyhow!(\n            \"Postgres sink does not support target table \\\"{}\\\".\\\"{}\\\" with foreign key constraints. Please remove foreign key constraints from the target table or choose a different sink table.\",\n            schema,\n            table,\n        )));\n    }\n\n    Ok(())\n}\n\nimpl PostgresConfig {\n    pub fn from_btreemap(properties: BTreeMap<String, String>) -> Result<Self> {\n        let config =\n            serde_json::from_value::<PostgresConfig>(serde_json::to_value(properties).unwrap())\n                .map_err(|e| SinkError::Config(anyhow!(e)))?;\n        if config.r#type != SINK_TYPE_APPEND_ONLY && config.r#type != SINK_TYPE_UPSERT {\n            return Err(SinkError::Config(anyhow!(\n                \"`{}` must be {}, or {}\",\n                SINK_TYPE_OPTION,\n                SINK_TYPE_APPEND_ONLY,\n                SINK_TYPE_UPSERT\n            )));\n        }\n        Ok(config)\n    }\n\n    pub fn pg_connection_config(&self) -> PgConnectionConfig {\n        PgConnectionConfig {\n            host: self.host.clone(),\n            port: self.port,\n            user: self.user.clone(),\n            password: self.password.clone(),\n            database: self.database.clone(),\n            ssl_mode: self.ssl_mode.clone(),\n            ssl_root_cert: self.ssl_root_cert.clone(),","sourceCodeStart":138,"sourceCodeEnd":174,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/postgres.rs#L138-L174","documentation":"`PostgresConfig::from_btreemap` requires the sink `type` option to be either `append-only` or `upsert`; any other value is rejected with this message. It means the configured sink mode is not one of the two supported by the Postgres sink.","triggerScenarios":"Creating a Postgres sink whose WITH options contain `type` with a value other than `append-only` or `upsert` (e.g. `type='debezium'`, `type='insert'`, or a typo like `append_only`).","commonSituations":"Copy-pasting `type` values from other connectors (e.g. Kafka's `debezium`); typo like `upsertt` or using a deprecated value.","solutions":["Set `type='upsert'` if the target table should reflect latest values keyed by primary key","Set `type='append-only'` if the stream is append-only and rows should be inserted","Use exactly these literal strings — check spelling and case"],"exampleFix":"// before\nWITH (connector='postgres', ..., type='insert');\n// after\nWITH (connector='postgres', ..., type='upsert');","handlingStrategy":"validation","validationCode":"let sink_type = options.get(\"type\").map(String::as_str).unwrap_or_default();\nassert!(matches!(sink_type, \"append-only\" | \"upsert\"), \"type must be append-only or upsert, got '{}'\", sink_type);","typeGuard":"fn is_valid_pg_sink_type(t: &str) -> bool { matches!(t, \"append-only\" | \"upsert\") }","tryCatchPattern":null,"preventionTips":["Use exactly type='upsert' or type='append-only'","Do not copy type values from Kafka connectors (e.g. debezium)","Double-check spelling when editing sink options"],"tags":["postgres","sink","config","invalid-enum"],"backgroundTag":"invalid-enum-value","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}