{"record":{"id":"5db0c5ced5ca9c68","repo":"risingwavelabs/risingwave","slug":"snowflake-stage-is-required-for-s3-writer","errorCode":null,"errorMessage":"snowflake.stage is required for S3 writer","messagePattern":"snowflake\\.stage is required for S3 writer","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/snowflake_redshift/snowflake.rs","lineNumber":949,"sourceCode":"                .await?;\n        }\n        Ok(())\n    }\n\n    pub async fn execute_create_pipe(&self) -> Result<()> {\n        if let Some(pipe_name) = &self.snowflake_task_context.pipe_name {\n            let table_name =\n                if let Some(table_name) = self.snowflake_task_context.cdc_table_name.as_ref() {\n                    table_name\n                } else {\n                    &self.snowflake_task_context.target_table_name\n                };\n            let create_pipe_sql = build_create_pipe_sql(\n                table_name,\n                &self.snowflake_task_context.database,\n                &self.snowflake_task_context.schema_name,\n                self.snowflake_task_context.stage.as_ref().ok_or_else(|| {\n                    SinkError::Config(anyhow!(\"snowflake.stage is required for S3 writer\"))\n                })?,\n                pipe_name,\n                &self.snowflake_task_context.target_table_name,\n            );\n            self.jdbc_client\n                .execute_sql_sync(vec![create_pipe_sql])\n                .await?;\n        }\n        Ok(())\n    }\n\n    pub async fn execute_drop_legacy_pipe(&self) -> Result<()> {\n        // Older upsert sinks created this pipe and refreshed it from a local timer. Remove it\n        // before starting the task so no asynchronous Snowpipe load can race with MERGE/DELETE.\n        if self.snowflake_task_context.task_name.is_some()\n            && self.snowflake_task_context.stage.is_some()\n        {\n            let pipe_name = format!(\"{}_pipe\", self.snowflake_task_context.target_table_name);","sourceCodeStart":931,"sourceCodeEnd":967,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/snowflake.rs#L931-L967","documentation":"The S3-based Snowflake writer builds a COPY pipe that ingests from an S3 stage; `snowflake_task_context.stage` must be set. When it is None, pipe creation fails with this Config error because the stage name is needed for `build_create_pipe_sql`.","triggerScenarios":"Using a Snowflake sink with S3 mode enabled where the task context was built without a `snowflake.stage` option, then writing data triggers pipe creation at snowflake.rs:949.","commonSituations":"S3 mode enabled but the `snowflake.stage` option omitted; copying JDBC-mode sink definitions into S3 mode; stage deleted/renamed in Snowflake.","solutions":["Add `snowflake.stage = '<stage_name>'` to the sink WITH options when using S3 mode","Verify the stage exists in Snowflake (`CREATE STAGE ...`) and the role can read it","If no stage is desired, disable S3 mode and use the JDBC writer"],"exampleFix":"-- before\nWITH (connector='snowflake', snowflake.s3=true, ...)\n-- after\nWITH (connector='snowflake', snowflake.s3=true, snowflake.stage='MY_STAGE', ...)","handlingStrategy":"validation","validationCode":"if opts.get(\"snowflake.s3\").map_or(false, |v| v == \"true\") {\n    assert!(opts.contains_key(\"snowflake.stage\"), \"snowflake.stage is required for S3 writer\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Pair S3 mode with snowflake.stage in every S3-mode sink definition","Create and grant access to the Snowflake stage before sink creation"],"tags":["snowflake","sink","s3","stage","config"],"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-23T08:17:48.524Z"}