risingwavelabs/risingwave · error · SinkError::Config

`{}` must be {}, or {}

Error message

`{}` must be {}, or {}

What it means

After deserialization, `from_btreemap` checks the sink `type` option: it must be either `append-only` or `upsert` for the Snowflake sink. Any other value fails this Config error listing the accepted values. The sink's write path depends on this mode, so it is validated eagerly at creation.

Source

Thrown at src/connector/src/sink/snowflake_redshift/snowflake.rs:235

                }
            }
            _ => {
                // This should never happen since from_btreemap validates auth_method
                unreachable!(
                    "Invalid auth_method - should have been caught during config validation"
                )
            }
        }

        Ok((jdbc_url, connection_properties))
    }

    pub fn from_btreemap(properties: &BTreeMap<String, String>) -> Result<Self> {
        let mut config =
            serde_json::from_value::<SnowflakeV2Config>(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
            )));
        }
        if config.r#type == SINK_TYPE_UPSERT && !config.with_s3 {
            return Err(SinkError::Config(anyhow!(
                "Snowflake upsert sinks require `with_s3 = true` so all CDC rows are loaded by the serialized COPY INTO task"
            )));
        }
        let has_upsert_task_config = config.snowflake_cdc_table_name.is_some()
            || properties.contains_key("write.target.interval.seconds")
            || config.snowflake_warehouse.is_some()
            || config.task_serverless
            || config.task_target_completion_interval.is_some();
        if config.r#type != SINK_TYPE_UPSERT && has_upsert_task_config {
            return Err(SinkError::Config(anyhow!(

View on GitHub (pinned to 6469eb736d)

Solutions

1. Set `type = 'append-only'` or `type = 'upsert'` in the sink WITH options.
2. Use underscores, not hyphens: `append_only` and similar misspellings are rejected.
3. Match the type to the source: upsert if the materialized view has primary keys/updates, otherwise append-only.

Example fix

// before
WITH (connector='snowflake', ..., type='append_only');
// after
WITH (connector='snowflake', ..., type='append-only');

let t = props.get("type").map(String::as_str);
if !matches!(t, Some("append-only") | Some("upsert")) {
    return Err(anyhow!("type must be append-only or upsert"));
}

When it happens

Trigger: Creating a Snowflake sink with `type` set to anything other than `append-only` or `upsert`, e.g. `type='debezium'`, `type='insert'`, or a misspelled `type='append_only'`.

Common situations: Copying sink definitions from other connectors (e.g. JDBC/debezium sinks) whose accepted type values differ; typos like `append_only`; omitting `type` when required.

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

• jdbc.url is required
• username is required
• (dynamic: serde deserialization error from sink properties)
• intermediate.table.name is required for non-append-only sink
• plan "{}" does not have a valid fragment id

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