{"record":{"id":"339750cccdd80e9e","repo":"risingwavelabs/risingwave","slug":"file-sink-only-supports-parquet-and-json-encod","errorCode":null,"errorMessage":"File sink only supports `PARQUET` and `JSON` encode at present.","messagePattern":"File sink only supports `PARQUET` and `JSON` encode at present\\.","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/file_sink/opendal_sink.rs","lineNumber":164,"sourceCode":"    async fn validate(&self) -> Result<()> {\n        if matches!(self.engine_type, EngineType::Snowflake) {\n            risingwave_common::license::Feature::SnowflakeSink\n                .check_available()\n                .map_err(|e| anyhow::anyhow!(e))?;\n        }\n        if !self.is_append_only {\n            return Err(SinkError::Config(anyhow!(\n                \"File sink only supports append-only mode at present. \\\n                    Please change the query to append-only, and specify it \\\n                    explicitly after the `FORMAT ... ENCODE ...` statement. \\\n                    For example, `FORMAT xxx ENCODE xxx(force_append_only='true')`\"\n            )));\n        }\n\n        if self.format_desc.encode != SinkEncode::Parquet\n            && self.format_desc.encode != SinkEncode::Json\n        {\n            return Err(SinkError::Config(anyhow!(\n                \"File sink only supports `PARQUET` and `JSON` encode at present.\"\n            )));\n        }\n\n        match self.op.list(&self.path).await {\n            Ok(_) => Ok(()),\n            Err(e) => Err(anyhow!(e).into()),\n        }\n    }\n\n    async fn new_log_sinker(\n        &self,\n        writer_param: crate::sink::SinkWriterParam,\n    ) -> Result<Self::LogSinker> {\n        let writer = OpenDalSinkWriter::new(\n            self.op.clone(),\n            &self.path,\n            self.schema.clone(),","sourceCodeStart":146,"sourceCodeEnd":182,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/file_sink/opendal_sink.rs#L146-L182","documentation":"The file sink's `validate` restricts the encode format to PARQUET or JSON. If the sink's `format_desc.encode` is any other SinkEncode variant (e.g. Avro, Protobuf, Text, Bytes), creation is rejected because the file sink has no writer implementation for those formats.","triggerScenarios":"Creating a file sink whose FORMAT/ENCODE resolves to an encode other than PARQUET or JSON, e.g. `ENCODE AVRO`, `ENCODE PROTOBUF`, `ENCODE BYTES`, or an unset/unexpected encode in format_desc.","commonSituations":"Copying sink definitions from Kafka connectors that support Avro/Protobuf; assuming all encodes work for file sinks; typo or wrong FORMAT clause picked a different encode.","solutions":["Use `ENCODE JSON` (optionally with `force_append_only = 'true'`) or `ENCODE PARQUET` in the FORMAT...ENCODE clause","If binary/schema-registry formats are required, use a Kafka sink instead of a file sink","Check the resolved format_desc in your statement — ensure FORMAT matches a supported file-sink combination"],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector = 'gcs', ...)\nFORMAT DEBEZIUM ENCODE AVRO;\n// after\nCREATE SINK s FROM mv WITH (connector = 'gcs', ...)\nFORMAT PLAIN ENCODE JSON(force_append_only = 'true');","handlingStrategy":"validation","validationCode":"let supported_encodes = [\"json\", \"parquet\"];\nlet encode = \"json\"; // value from your ENCODE clause\nif !supported_encodes.contains(&encode.to_lowercase().as_str()) {\n    return Err(format!(\"File sink supports only PARQUET and JSON, got {}\", encode));\n}","typeGuard":"fn is_file_sink_encode(e: &str) -> bool {\n    matches!(e.to_uppercase().as_str(), \"JSON\" | \"PARQUET\")\n}","tryCatchPattern":"match err {\n    e if e.to_string().contains(\"PARQUET` and `JSON\") => {\n        eprintln!(\"Switch the ENCODE clause to JSON or PARQUET\");\n    }\n    other => return Err(other),\n}","preventionTips":["Always use ENCODE JSON or ENCODE PARQUET with file sinks","Do not copy FORMAT/ENCODE clauses from Kafka sink definitions that use Avro/Protobuf","Pin file-sink templates in team SQL libraries to supported format combinations"],"tags":["sink","file-sink","format","validation"],"backgroundTag":"unsupported-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"}