risingwavelabs/risingwave · error · SinkError::Config

Snowflake upsert sinks require `with_s3 = true` so all CDC…

Error message

Snowflake upsert sinks require `with_s3 = true` so all CDC rows are loaded by the serialized COPY INTO task

What it means

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.

Solutions

1. Add `with_s3 = true` to the WITH options of the Snowflake sink
2. Provide the S3 staging configuration the sink needs (s3.region, s3.bucket, s3.path) alongside with_s3
3. If upsert semantics are not actually needed, keep the sink append-only instead of enabling with_s3

Example fix

// before
CREATE SINK sf_sink FROM mv WITH (
  connector = 'snowflake', type = 'upsert', ...
);
// after
CREATE SINK sf_sink FROM mv WITH (
  connector = 'snowflake', type = 'upsert', with_s3 = true, ...
);

if sink_type == 'upsert' && !options.contains_key("with_s3") {
    return Err("Snowflake upsert sink requires with_s3 = true");
}

When it happens

Trigger: Creating a Snowflake sink via CREATE SINK with `type = 'upsert'` (or SINK_TYPE_UPSERT in code) while omitting the `with_s3 = true` option.

Common situations: 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.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

• `intermediate.table.name`, `write.target.interval.seconds`…
• `task.serverless` must not be combined with `warehouse`
• `task.target_completion_interval` requires…
• auth.method=key_pair_file must not set `password`
• auth.method=key_pair_file must not set `private_key_pem`

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

Appendix: source

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

        }

        Ok((jdbc_url, connection_properties))
    }

    pub fn from_btreemap(properties: &BTreeMap<String, String>) -> Result<Self> {
        let mut config =
            serde_json::from_value::<SnowflakeV2Config>(serde_json::to_value(properties).unwrap())
                .map_err(|e| SinkError::Config(anyhow!(e)))?;
        if config.r#type != SINK_TYPE_APPEND_ONLY && config.r#type != SINK_TYPE_UPSERT {
            return Err(SinkError::Config(anyhow!(
                "`{}` must be {}, or {}",
                SINK_TYPE_OPTION,
                SINK_TYPE_APPEND_ONLY,
                SINK_TYPE_UPSERT
            )));
        }
        if config.r#type == SINK_TYPE_UPSERT && !config.with_s3 {
            return Err(SinkError::Config(anyhow!(
                "Snowflake upsert sinks require `with_s3 = true` so all CDC rows are loaded by the serialized COPY INTO task"
            )));
        }
        let has_upsert_task_config = config.snowflake_cdc_table_name.is_some()
            || properties.contains_key("write.target.interval.seconds")
            || config.snowflake_warehouse.is_some()
            || config.task_serverless
            || config.task_target_completion_interval.is_some();
        if config.r#type != SINK_TYPE_UPSERT && has_upsert_task_config {
            return Err(SinkError::Config(anyhow!(
                "`intermediate.table.name`, `write.target.interval.seconds`, `warehouse`, \
                 `task.serverless`, and `task.target_completion_interval` require `{}` = {}",
                SINK_TYPE_OPTION,
                SINK_TYPE_UPSERT
            )));
        }
        if config.task_target_completion_interval.is_some() && !config.task_serverless {
            return Err(SinkError::Config(anyhow!(

View on GitHub (pinned to 6469eb736d)