{"record":{"id":"47c46f577f71ac3e","repo":"risingwavelabs/risingwave","slug":"missing-format-encode","errorCode":null,"errorMessage":"missing FORMAT ... ENCODE ...","messagePattern":"missing FORMAT \\.\\.\\. ENCODE \\.\\.\\.","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/file_sink/opendal_sink.rs","lineNumber":209,"sourceCode":"\nimpl<S: OpendalSinkBackend> TryFrom<SinkParam> for FileSink<S> {\n    type Error = SinkError;\n\n    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {\n        let schema = param.schema();\n        let config = S::from_btreemap(param.properties)?;\n        let unknown_fields = crate::sink::UnknownFields::unknown_fields(&config);\n        let path = S::get_path(config.clone());\n        let op = S::new_operator(config.clone())?;\n        let batching_strategy = S::get_batching_strategy(config);\n        let engine_type = S::get_engine_type();\n        let format_desc = match param.format_desc {\n            Some(desc) => desc,\n            None => {\n                if let EngineType::Snowflake = engine_type {\n                    SinkFormatDesc::plain_json_for_snowflake_only()\n                } else {\n                    return Err(SinkError::Config(anyhow!(\"missing FORMAT ... ENCODE ...\")));\n                }\n            }\n        };\n        Ok(Self {\n            op,\n            path,\n            schema,\n            is_append_only: param.sink_type.is_append_only(),\n            batching_strategy,\n            format_desc,\n            engine_type,\n            unknown_fields,\n            _marker: PhantomData,\n        })\n    }\n}\n\npub struct OpenDalSinkWriter {","sourceCodeStart":191,"sourceCodeEnd":227,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/file_sink/opendal_sink.rs#L191-L227","documentation":"When building an OpenDAL-based file sink, the `FORMAT ... ENCODE ...` declaration of the sink is required to know how rows are serialized. If `param.format_desc` is None and the engine is not Snowflake (which has a plain-JSON fallback), the connector rejects creation with this config error.","triggerScenarios":"Creating a file sink (e.g. S3/opendal based) without a `FORMAT ... ENCODE ...` clause on a non-Snowflake engine — the property simply was never parsed into format_desc.","commonSituations":"Old SQL habits from other systems where FORMAT is optional; copy-pasted CREATE SINK statements stripped of the FORMAT line; using a generic sink wrapper that does not forward format options.","solutions":["Add an explicit FORMAT/ENCODE clause to the CREATE SINK statement, e.g. `FORMAT PLAIN ENCODE JSON` or `FORMAT DEBEZIUM_JSON ENCODE JSON`","If targeting Snowflake engine specifically, ensure the engine type is detected correctly (format is defaulted there)","Check that the SQL was not truncated or that a wrapper is dropping the FORMAT option"],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector='file_s3', path='s3://bucket/out/');\n// after\nCREATE SINK s FROM mv WITH (connector='file_s3', path='s3://bucket/out/') FORMAT PLAIN ENCODE JSON;","handlingStrategy":"validation","validationCode":"// Ensure the sink SQL contains a FORMAT ... ENCODE clause\nconst required = /FORMAT\\s+(PLAIN|DEBEZIUM|MAXWELL|CANAL|DEBEZIUM_JSON|UPSERT)\\s+ENCODE\\s+(JSON|AVRO|PROTOBUF)/i;\nif (!required.test(createSinkSql)) throw new Error('CREATE SINK needs a FORMAT ... ENCODE clause');","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always end file-sink WITH clauses with an explicit FORMAT/ENCODE clause","Keep a canonical CREATE SINK template per sink type","Lint generated SQL for mandatory clauses"],"tags":["sql","sink","config","missing-option"],"backgroundTag":"missing-required-config-field","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"}