risingwavelabs/risingwave · error · SinkError::Config

intermediate.table.name is required

Error message

intermediate.table.name is required

What it means

For non-append-only sinks (upsert/Debezium CDC), the Snowflake connector writes to an intermediate CDC table before merging into the target. `build_snowflake_task_ctx_jdbc_client` requires the `intermediate.table.name` option and throws this config error when it is absent.

Source

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

        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 {
                snowflake_task_ctx.warehouse = Some(
                    self.snowflake_warehouse
                        .clone()
                        .ok_or(SinkError::Config(anyhow!("warehouse is required")))?,
                );
            }
            let pk_column_names: Vec<_> = schema
                .fields
                .iter()
                .enumerate()

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add `intermediate.table.name = '<intermediate_table>'` to the WITH options.
2. Ensure the intermediate table name differs from the target table name and exists (or is creatable) in Snowflake.
3. If the data is truly append-only, define the sink without a primary key so the intermediate table is not required.

Example fix

// before
WITH (
  connector = 'snowflake',
  primary_key = 'id'
)
// after
WITH (
  connector = 'snowflake',
  primary_key = 'id',
  intermediate.table.name = 'my_table_rw_cdc'
)

if (opts.primary_key && !opts['intermediate.table.name']) {
  throw new Error('intermediate.table.name is required for non-append-only snowflake sink');
}

try { createSink(opts) } catch (e) { if (String(e).includes('intermediate.table.name is required')) { /* add intermediate table option */ } }

When it happens

Trigger: Creating a non-append-only Snowflake sink (e.g., with a primary key / upsert) without `snowflake_cdc_table_name` / `intermediate.table.name` in the WITH options.

Common situations: Following append-only sink examples for a CDC/upsert sink, or dropping the option when changing the sink's append-only property.

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

• intermediate.table.name is required for non-append-only sink
• auth.method=key_pair_object must not set `password`
• ambiguous auth: multiple auth options provided; remove one o
• table.name is required
• database is required

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