{"record":{"id":"045ba1eb484a33f9","repo":"risingwavelabs/risingwave","slug":"must-be-or-045ba1","errorCode":null,"errorMessage":"`{}` must be {}, or {}","messagePattern":"`(.+?)` must be (.+?), or (.+?)","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/file_sink/fs.rs","lineNumber":84,"sourceCode":"            .layer(LoggingLayer::default())\n            .layer(RetryLayer::default());\n        Ok(operator)\n    }\n}\n\n#[derive(Debug, Clone, Copy, PartialEq, Eq)]\npub struct FsSink;\n\nimpl OpendalSinkBackend for FsSink {\n    type Properties = FsConfig;\n\n    const SINK_NAME: &'static str = FS_SINK;\n\n    fn from_btreemap(btree_map: BTreeMap<String, String>) -> Result<Self::Properties> {\n        let config = serde_json::from_value::<FsConfig>(serde_json::to_value(btree_map).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    fn new_operator(properties: FsConfig) -> Result<Operator> {\n        FileSink::<FsSink>::new_fs_sink(properties)\n    }\n\n    fn get_path(properties: Self::Properties) -> String {\n        properties.common.path\n    }\n\n    fn get_engine_type() -> super::opendal_sink::EngineType {","sourceCodeStart":66,"sourceCodeEnd":102,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/file_sink/fs.rs#L66-L102","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Change the option to type='append-only' for insert-only streams.","Change to type='upsert' when the stream may contain update/delete events (primary key required).","Use exact lowercase spelling with the hyphen, as printed in the error.","If unsure whether the stream is append-only, check for primary keys or retracting upstream operators and choose 'upsert'."],"exampleFix":"// before\nWITH (connector = 'fs', type = 'AppendOnly');\n// after\nWITH (connector = 'fs', type = 'append-only');","handlingStrategy":"validation","validationCode":"match opts.get(\"r#type\").map(String::as_str) {\n    Some(\"append-only\") | Some(\"upsert\") => Ok(()),\n    other => Err(format!(\"fs sink r#type must be 'append-only' or 'upsert', got {other:?}\")),\n}","typeGuard":null,"tryCatchPattern":"match FsSink::from_btreemap(opts) {\n    Ok(cfg) => proceed(cfg),\n    Err(SinkError::Config(e)) => eprintln!(\"invalid sink type: {e:#}\"),\n    Err(e) => return Err(e),\n}","preventionTips":["Use the exact literals 'append-only' or 'upsert'; the check is case-sensitive.","Prefer 'upsert' when the stream has a primary key or possible retractions.","Reuse a validated DDL template rather than retyping the value.","Trim whitespace around option values."],"tags":["sink","config","validation","filesystem"],"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-14T16:17:12.679Z"}