risingwavelabs/risingwave · error · SinkError

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

Error message

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

What it means

WebHDFS file sink config validation in from_btreemap: the `type` property (r#type) held a value other than the two supported sink types (append-only / deb-mutating, per the `{}` must be {}, or {} template). Same strict enum guard as the S3 backend, applied after serde-parsing WebhdfsConfig.

Source

Thrown at src/connector/src/sink/file_sink/webhdfs.rs:89

impl UnknownFields for WebhdfsConfig {
    fn unknown_fields(&self) -> HashMap<String, String> {
        self.unknown_fields.clone()
    }
}

crate::impl_sink_unknown_fields!(WebhdfsConfig);

impl OpendalSinkBackend for WebhdfsSink {
    type Properties = WebhdfsConfig;

    const SINK_NAME: &'static str = WEBHDFS_SINK;

    fn from_btreemap(btree_map: BTreeMap<String, String>) -> Result<Self::Properties> {
        let config =
            serde_json::from_value::<WebhdfsConfig>(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: WebhdfsConfig) -> Result<Operator> {
        FileSink::<WebhdfsSink>::new_webhdfs_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. Use `type='append-only'` or `type='upsert'` verbatim
2. Check spelling and separator characters
3. Choose based on data characteristics: upsert only if the stream contains update/delete operations

Example fix

// before
WITH (connector='webhdfs', type='upsert-only', ...)
// after
WITH (connector='webhdfs', type='upsert', ...)

const allowed = ['append-only', 'upsert'];
if (opts.type !== undefined && !allowed.includes(opts.type)) {
  throw new Error(`webhdfs sink type must be ${allowed.join(' or ')}`);
}

const isValidWebhdfsSinkType = (t) => t === 'append-only' || t === 'upsert';

try { createSink(opts); } catch (e) { if (String(e).includes('must be append-only, or upsert')) { opts.type = 'append-only'; createSink(opts); } else { throw e; } }

When it happens

Trigger: `CREATE SINK ... WITH (connector='webhdfs', type='...')` with a value other than the two allowed literals (typos, wrong case, underscores).

Common situations: Underscore vs hyphen mistakes ('append_only'), capitalization, or reusing `type` values from unrelated systems.

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 {}
• `{}` must be {}, or {}
• `{}` must be {}, or {}
• `{}` must be {}, or {}

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