risingwavelabs/risingwave · error · SinkError

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

Error message

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

What it means

The fs sink checks that FsConfig.r#type is exactly 'append-only' or 'upsert' after deserialization and rejects anything else with a SinkError::Config naming the option and its two legal values. This guards the sink against write modes it cannot honor on the filesystem.

Source

Thrown at src/connector/src/sink/file_sink/fs.rs:84

            .layer(LoggingLayer::default())
            .layer(RetryLayer::default());
        Ok(operator)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FsSink;

impl OpendalSinkBackend for FsSink {
    type Properties = FsConfig;

    const SINK_NAME: &'static str = FS_SINK;

    fn from_btreemap(btree_map: BTreeMap<String, String>) -> Result<Self::Properties> {
        let config = serde_json::from_value::<FsConfig>(serde_json::to_value(btree_map).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)
    }

    fn new_operator(properties: FsConfig) -> Result<Operator> {
        FileSink::<FsSink>::new_fs_sink(properties)
    }

    fn get_path(properties: Self::Properties) -> String {
        properties.common.path
    }

    fn get_engine_type() -> super::opendal_sink::EngineType {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Change the option to type='append-only' for insert-only streams.
2. Change to type='upsert' when the stream may contain update/delete events (primary key required).
3. Use exact lowercase spelling with the hyphen, as printed in the error.
4. If unsure whether the stream is append-only, check for primary keys or retracting upstream operators and choose 'upsert'.

Example fix

// before
WITH (connector = 'fs', type = 'AppendOnly');
// after
WITH (connector = 'fs', type = 'append-only');

match opts.get("r#type").map(String::as_str) {
    Some("append-only") | Some("upsert") => Ok(()),
    other => Err(format!("fs sink r#type must be 'append-only' or 'upsert', got {other:?}")),
}

match FsSink::from_btreemap(opts) {
    Ok(cfg) => proceed(cfg),
    Err(SinkError::Config(e)) => eprintln!("invalid sink type: {e:#}"),
    Err(e) => return Err(e),
}

When it happens

Trigger: CREATE SINK ... WITH (connector='fs', type='...') with any value other than the literal strings 'append-only' or 'upsert' — e.g. 'append', 'overwrite', 'AppendOnly', or a value with stray whitespace.

Common situations: Typos or casing mistakes in DDL; copying type values from other systems; assuming 'overwrite' exists for fs like in some batch tools; hand-editing generated SQL and breaking the 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

• `commit_checkpoint_interval` must be greater than 0
• `{}` must be {}, or {}
• config deserialization error: {e}
• `{}` must be {}, or {}
• `{}` must be {}, or {}

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