risingwavelabs/risingwave · error · SinkError::Config

stage is required

Error message

stage is required

What it means

When the sink is configured with S3 staging (`with_s3` is true), data is unloaded through a Snowflake external stage. `build_snowflake_task_ctx_jdbc_client` reads the `stage` option and fails with this config error if the stage name is missing from the options map.

Solutions

1. Add `stage = '<stage_name>'` (the Snowflake external stage, e.g., `my_stage`) to the WITH options.
2. Create the stage in Snowflake if it does not exist: `CREATE STAGE my_stage URL='s3://...' ...`.
3. If S3 staging is not intended, disable the `with_s3` option so the stage is not required.

Example fix

// before
WITH (
  connector = 'snowflake',
  with_s3 = true
)
// after
WITH (
  connector = 'snowflake',
  with_s3 = true,
  stage = 'my_stage'
)

if (opts.with_s3 && !opts.stage) throw new Error('stage is required when with_s3 is enabled');

try { createSink(opts) } catch (e) { if (String(e).includes('stage is required')) { /* configure stage or disable with_s3 */ } }

When it happens

Trigger: Creating a Snowflake sink with `snowflake.stage` omitted while S3-based unloading is enabled (e.g., `with_s3 = true` / corresponding flag set).

Common situations: Enabling the S3 stage path without creating/configuring the stage in Snowflake, or forgetting the `stage` option key when switching from direct JDBC writes to staged writes.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• S3 configuration is required for Snowflake S3 sink
• snowflake.stage is required for S3 writer
• ambiguous auth: multiple auth options provided; remove one…
• auth.method=key_pair_object must not set `password`
• database is required

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/3c5133ed1492a364. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/sink/snowflake_redshift/snowflake.rs:395

            .snowflake_schema
            .clone()
            .ok_or(SinkError::Config(anyhow!("schema is required")))?;
        let mut snowflake_task_ctx = SnowflakeTaskContext {
            target_table_name: target_table_name.clone(),
            database,
            schema_name,
            schema: schema.clone(),
            ..Default::default()
        };

        let (jdbc_url, connection_properties) = self.build_jdbc_connection_properties()?;
        let client = JdbcJniClient::new_with_props(jdbc_url, connection_properties)?;

        if self.with_s3 {
            let stage = self
                .stage
                .clone()
                .ok_or(SinkError::Config(anyhow!("stage is required")))?;
            snowflake_task_ctx.stage = Some(stage);
            if is_append_only {
                snowflake_task_ctx.pipe_name = Some(format!("{}_pipe", target_table_name));
            }
        }
        if !is_append_only {
            let cdc_table_name = self
                .snowflake_cdc_table_name
                .clone()
                .ok_or(SinkError::Config(anyhow!(
                    "intermediate.table.name is required"
                )))?;
            snowflake_task_ctx.cdc_table_name = Some(cdc_table_name.clone());
            snowflake_task_ctx.writer_target_interval_seconds = self.writer_target_interval_seconds;
            snowflake_task_ctx.task_serverless = self.task_serverless;
            snowflake_task_ctx.task_target_completion_interval =
                self.task_target_completion_interval.clone();
            if !self.task_serverless {

View on GitHub (pinned to 6469eb736d)