{"record":{"id":"e3ee1a59d7504152","repo":"risingwavelabs/risingwave","slug":"snowflake-upsert-sinks-require-with-s3-true-so","errorCode":null,"errorMessage":"Snowflake upsert sinks require `with_s3 = true` so all CDC rows are loaded by the serialized COPY INTO task","messagePattern":"Snowflake upsert sinks require `with_s3 = true` so all CDC rows are loaded by the serialized COPY INTO task","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/snowflake_redshift/snowflake.rs","lineNumber":243,"sourceCode":"        }\n\n        Ok((jdbc_url, connection_properties))\n    }\n\n    pub fn from_btreemap(properties: &BTreeMap<String, String>) -> Result<Self> {\n        let mut config =\n            serde_json::from_value::<SnowflakeV2Config>(serde_json::to_value(properties).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        if config.r#type == SINK_TYPE_UPSERT && !config.with_s3 {\n            return Err(SinkError::Config(anyhow!(\n                \"Snowflake upsert sinks require `with_s3 = true` so all CDC rows are loaded by the serialized COPY INTO task\"\n            )));\n        }\n        let has_upsert_task_config = config.snowflake_cdc_table_name.is_some()\n            || properties.contains_key(\"write.target.interval.seconds\")\n            || config.snowflake_warehouse.is_some()\n            || config.task_serverless\n            || config.task_target_completion_interval.is_some();\n        if config.r#type != SINK_TYPE_UPSERT && has_upsert_task_config {\n            return Err(SinkError::Config(anyhow!(\n                \"`intermediate.table.name`, `write.target.interval.seconds`, `warehouse`, \\\n                 `task.serverless`, and `task.target_completion_interval` require `{}` = {}\",\n                SINK_TYPE_OPTION,\n                SINK_TYPE_UPSERT\n            )));\n        }\n        if config.task_target_completion_interval.is_some() && !config.task_serverless {\n            return Err(SinkError::Config(anyhow!(","sourceCodeStart":225,"sourceCodeEnd":261,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/snowflake.rs#L225-L261","documentation":"SnowflakeSinkConfig::from_btreemap validates the sink options. An upsert (CDC) Snowflake sink must have `with_s3 = true` so every row is staged in S3 and loaded deterministically by the single serialized COPY INTO task; without S3 staging, upserts cannot be replayed in order. The connector rejects any upsert Snowflake sink missing this flag at creation time.","triggerScenarios":"Creating a Snowflake sink via CREATE SINK with `type = 'upsert'` (or SINK_TYPE_UPSERT in code) while omitting the `with_s3 = true` option.","commonSituations":"Copying an append-only Snowflake sink DDL and changing only the type to upsert; forgetting that Snowflake differs from other sinks where S3 staging is optional; older example configs that predate the with_s3 requirement.","solutions":["Add `with_s3 = true` to the WITH options of the Snowflake sink","Provide the S3 staging configuration the sink needs (s3.region, s3.bucket, s3.path) alongside with_s3","If upsert semantics are not actually needed, keep the sink append-only instead of enabling with_s3"],"exampleFix":"// before\nCREATE SINK sf_sink FROM mv WITH (\n  connector = 'snowflake', type = 'upsert', ...\n);\n// after\nCREATE SINK sf_sink FROM mv WITH (\n  connector = 'snowflake', type = 'upsert', with_s3 = true, ...\n);","handlingStrategy":"validation","validationCode":"if sink_type == 'upsert' && !options.contains_key(\"with_s3\") {\n    return Err(\"Snowflake upsert sink requires with_s3 = true\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always pair type='upsert' with with_s3=true in Snowflake sink templates","Review CREATE SINK DDL against the Snowflake sink docs before submitting","Keep a canonical working example for upsert Snowflake sinks"],"tags":["snowflake","sink","config-validation","risingwave"],"backgroundTag":"invalid-config-value","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}