risingwavelabs/risingwave · error · SinkError

The key encode is BYTES, but the primary key column

Error message

The key encode is BYTES, but the primary key column {} has type {}

What it means

When key encode is BYTES, the encoder maps the raw primary key bytes directly, so the key column must be of type BYTEA. The builder throws when the resolved PK column has any other type, since it refuses to reinterpret non-byte types as raw bytes.

Solutions

1. Change the key column to BYTEA (cast in the sink query)
2. Use key_encode='text' for varchar/int/bool keys
3. Use a different key encode format that supports the column's type

Example fix

// before
CREATE SINK s FROM t WITH (key_encode = 'bytes'); -- t.id is INT
// after
CREATE SINK s AS SELECT id::bytea AS id, * FROM t WITH (key_encode = 'bytes');

if key_encode == "bytes" && pk_field.data_type != DataType::Bytea {
    return Err("key encode bytes requires a bytea pk column");
}

fn is_bytes_key(dt: &DataType) -> bool { *dt == DataType::Bytea }

When it happens

Trigger: KEY ENCODE BYTES configured and ensure_only_one_pk succeeds, but schema_ref.data_type() is not DataType::Bytea (e.g. INT or VARCHAR primary key).

Common situations: Setting key_encode='bytes' out of habit on an INT-keyed table; migrating a Kafka sink config from another system where 'bytes' meant arbitrary serialization.

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

• ` ` must be set to `varchar` and `primary_key
• The key encode is TEXT, but the primary key column
• Turbopuffer document id column must be an integer or varchar
• collection.name.field
• Column ` ` in Postgres table ` ` has type ` `, but sink…

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

Appendix: source

Thrown at src/connector/src/sink/formatter/mod.rs:226

            "The primary key column index {} is out of bounds in schema {:?}",
            pk_indices[0],
            params.schema
        ))
    })?;

    Ok((pk_indices[0], schema_ref))
}

impl EncoderBuild for BytesEncoder {
    async fn build(params: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {
        match pk_indices {
            // This is being used as a key encoder
            Some(_) => {
                let (pk_index, schema_ref) = ensure_only_one_pk("BYTES", &params, &pk_indices)?;
                if let DataType::Bytea = schema_ref.data_type() {
                    Ok(BytesEncoder::new(params.schema, pk_index))
                } else {
                    Err(SinkError::Config(anyhow!(
                        "The key encode is BYTES, but the primary key column {} has type {}",
                        schema_ref.name,
                        schema_ref.data_type
                    )))
                }
            }
            // This is being used as a value encoder
            None => {
                // Ensure the schema has exactly one column and it's of type BYTEA
                if params.schema.len() != 1 {
                    return Err(SinkError::Config(anyhow!(
                        "ENCODE BYTES requires exactly one column, got {} columns",
                        params.schema.len()
                    )));
                }

                let field = &params.schema.fields[0];
                if let DataType::Bytea = field.data_type {

View on GitHub (pinned to 6469eb736d)