risingwavelabs/risingwave · error · SinkError

{e}

Error message

{e}

What it means

`S3Config::from_btreemap` serializes the sink's WITH-options BTreeMap to JSON and deserializes it into `S3Config`. Any field failing serde deserialization (unknown/invalid value, wrong shape) is mapped to `SinkError::Config(anyhow!(e))`, producing this error with the serde message.

Source

Thrown at src/connector/src/sink/file_sink/s3.rs:142

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct S3Sink;

impl UnknownFields for S3Config {
    fn unknown_fields(&self) -> HashMap<String, String> {
        self.unknown_fields.clone()
    }
}

crate::impl_sink_unknown_fields!(S3Config);

impl OpendalSinkBackend for S3Sink {
    type Properties = S3Config;

    const SINK_NAME: &'static str = S3_SINK;

    fn from_btreemap(btree_map: BTreeMap<String, String>) -> Result<Self::Properties> {
        let config = serde_json::from_value::<S3Config>(serde_json::to_value(btree_map).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
            )));
        }
        Ok(config)
    }

    fn new_operator(properties: S3Config) -> Result<Operator> {
        FileSink::<S3Sink>::new_s3_sink(&properties.common)
    }

    fn get_path(properties: Self::Properties) -> String {
        properties.common.path.unwrap_or_default()
    }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the embedded serde message to identify the offending field and fix its value
2. Check each WITH option against the S3Config struct definition (types and expected formats)
3. Remove unsupported options and rely on defaults; only pass documented S3 sink options

Example fix

// before
WITH (connector='s3', bucket_url='s3://bucket', path='out/', match_path='true')
// after
WITH (connector='s3', bucket_url='s3://bucket', path='out/')

// Validate S3 options before submitting
const s3Opts = { connector: 's3', bucket_url: 's3://bucket', path: 'out/' };
for (const [k, v] of Object.entries(s3Opts)) {
  if (typeof v !== 'string' || v.length === 0) throw new Error(`invalid value for ${k}`);
}

When it happens

Trigger: `CREATE SINK ... WITH (connector='s3', ...)` where one of the S3 options does not deserialize into S3Config — e.g. a value with the wrong type/shape, an unparseable URL, or an option typed as bool/int given a non-numeric string.

Common situations: Quoting mistakes leaving nested JSON-shaped options malformed; passing `bucket_url`/`path` values with unexpected characters; schema changes to S3Config making older saved properties invalid.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• serde_json deserialization error of ClickHouseConfig from pr
• `{}` must be {}, or {}
• {e}
• {e}
• SinkError::Config(anyhow!(e))

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