risingwavelabs/risingwave · error · SinkError::Config

` ` must be , or (SINK_TYPE_OPTION must be append-only or…

Error message

`{}` must be {}, or {} (SINK_TYPE_OPTION must be append-only or upsert)

What it means

`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.

Solutions

1. Set `type='upsert'` if the target table should reflect latest values keyed by primary key
2. Set `type='append-only'` if the stream is append-only and rows should be inserted
3. Use exactly these literal strings — check spelling and case

Example fix

// before
WITH (connector='postgres', ..., type='insert');
// after
WITH (connector='postgres', ..., type='upsert');

let sink_type = options.get("type").map(String::as_str).unwrap_or_default();
assert!(matches!(sink_type, "append-only" | "upsert"), "type must be append-only or upsert, got '{}'", sink_type);

fn is_valid_pg_sink_type(t: &str) -> bool { matches!(t, "append-only" | "upsert") }

When it happens

Trigger: 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`).

Common situations: Copy-pasting `type` values from other connectors (e.g. Kafka's `debezium`); typo like `upsertt` or using a deprecated value.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• `max_batch_rows` must be between 1 and
• {serde_json deserialization error for PostgresConfig from…
• ambiguous auth: multiple auth options provided; remove one…
• auth.method=key_pair_object must not set `password`
• {batch write failure, with context()}

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/3eb40b10f02b14bc. Report an issue: GitHub.

Appendix: source

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

    if has_foreign_key {
        return Err(SinkError::Config(anyhow!(
            "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.",
            schema,
            table,
        )));
    }

    Ok(())
}

impl PostgresConfig {
    pub fn from_btreemap(properties: BTreeMap<String, String>) -> Result<Self> {
        let config =
            serde_json::from_value::<PostgresConfig>(serde_json::to_value(properties).unwrap())
                .map_err(|e| SinkError::Config(anyhow!(e)))?;
        if config.r#type != SINK_TYPE_APPEND_ONLY && config.r#type != SINK_TYPE_UPSERT {
            return Err(SinkError::Config(anyhow!(
                "`{}` must be {}, or {}",
                SINK_TYPE_OPTION,
                SINK_TYPE_APPEND_ONLY,
                SINK_TYPE_UPSERT
            )));
        }
        Ok(config)
    }

    pub fn pg_connection_config(&self) -> PgConnectionConfig {
        PgConnectionConfig {
            host: self.host.clone(),
            port: self.port,
            user: self.user.clone(),
            password: self.password.clone(),
            database: self.database.clone(),
            ssl_mode: self.ssl_mode.clone(),
            ssl_root_cert: self.ssl_root_cert.clone(),

View on GitHub (pinned to 6469eb736d)