risingwavelabs/risingwave · error · SinkError::Http

unexpected payload column type, expected varchar or jsonb

Error message

unexpected payload column type, expected varchar or jsonb

What it means

At write time, `extract_payload` accepts only Utf8 (VARCHAR) and Jsonb scalars for the payload column. Any other scalar type in the payload column datum causes `write_chunk` to fail with this SinkError::Http, since the payload cannot be turned into a request body.

Source

Thrown at src/connector/src/sink/http.rs:369

            },
        }
    }

    fn strip_payload_for_log(&self, row: &impl Row) -> String {
        match row.datum_at(self.payload_index) {
            Some(ScalarRefImpl::Utf8(s)) => strip_text_payload(s),
            Some(ScalarRefImpl::Jsonb(j)) => strip_jsonb_payload(j),
            Some(_) => "<unexpected payload type>".to_owned(),
            None => "NULL".to_owned(),
        }
    }

    fn extract_payload(&self, row: &impl Row) -> Result<Option<String>> {
        Ok(match row.datum_at(self.payload_index) {
            Some(ScalarRefImpl::Utf8(s)) => Some(s.to_owned()),
            Some(ScalarRefImpl::Jsonb(j)) => Some(j.to_string()),
            Some(_) => {
                return Err(SinkError::Http(anyhow!(
                    "unexpected payload column type, expected varchar or jsonb"
                )));
            }
            None => None, // skip NULL rows
        })
    }
}

fn strip_text_payload(payload: &str) -> String {
    const EDGE_CHAR_COUNT: usize = 100;
    let char_count = payload.chars().count();
    if char_count <= EDGE_CHAR_COUNT * 2 {
        return payload.to_owned();
    }

    let prefix: String = payload.chars().take(EDGE_CHAR_COUNT).collect();
    let suffix: String = payload.chars().skip(char_count - EDGE_CHAR_COUNT).collect();
    format!("{prefix}...{suffix}")

View on GitHub (pinned to 6469eb736d)

Solutions

1. Recreate the sink to re-validate the payload column type
2. Cast the payload to VARCHAR or JSONB in the sink query
3. Restore the payload column's original VARCHAR/JSONB type

Example fix

// before
SELECT payload FROM t; -- payload became INT after ALTER
// after
SELECT payload::JSONB AS payload FROM t;

fn is_payload_compatible(d: &ScalarRefImpl) -> bool {
    matches!(d, ScalarRefImpl::Utf8(_) | ScalarRefImpl::Jsonb(_))
}

match res {
    Err(e) if e.to_string().contains("unexpected payload column type") => {
        // fix payload column type or recreate sink
    }
    other => other?,
}

When it happens

Trigger: A row's payload column datum is a scalar other than Utf8/Jsonb during `write_chunk` -> `extract_payload` — typically after the underlying column type changed post sink creation, or schema drift in a source table.

Common situations: Column type altered after sink creation (e.g. VARCHAR -> INT); upstream query changed so the sink now receives numeric/boolean/struct payloads; index/column misalignment in internal sink plumbing.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• unexpected url column type, expected varchar
• HTTP sink url column must be varchar, got {:?}
• HTTP sink payload column must be varchar or jsonb, got {:?}
• Can't find data
• Insert error: {:?}, error url: {:?}

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