risingwavelabs/risingwave · error · SinkError

ENCODE BYTES requires exactly one column, got

Error message

ENCODE BYTES requires exactly one column, got {} columns

What it means

As a value encoder, ENCODE BYTES passes the single row value through as raw bytes, so the schema must contain exactly one column. The builder throws when the sink schema has any other number of columns.

Solutions

1. Reduce the sink output to a single BYTEA column (SELECT payload FROM ...)
2. Use VALUE ENCODE JSON/AVRO/PROTOBUF for multi-column rows
3. If a key is needed, configure key encode separately — value bytes mode never allows extra columns

Example fix

// before
CREATE SINK s AS SELECT a, b FROM t WITH (value_encode = 'bytes');
// after
CREATE SINK s AS SELECT encode(a::varchar || b::varchar, 'escape')::bytea AS payload FROM t WITH (value_encode = 'bytes');

if value_encode == "bytes" && schema.len() != 1 {
    return Err(format!("value encode bytes requires exactly 1 column, got {}", schema.len()));
}

fn is_single_col(schema: &Schema) -> bool { schema.len() == 1 }

When it happens

Trigger: build() in the value-encoder branch (pk_indices = None) where params.schema.len() != 1 — e.g. sinking a multi-column table with VALUE ENCODE BYTES.

Common situations: Applying 'bytes' value encoding to a normal relational sink; forgetting that bytes value encoding is meant for opaque pre-serialized payloads.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• auto schema refresh sink must have only one fragment, but…
• Can't find data
• Cannot find
• Cannot find
• collection.name.field

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

Appendix: source

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

        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 {
                    Ok(BytesEncoder::new(params.schema, 0))
                } else {
                    Err(SinkError::Config(anyhow!(
                        "ENCODE BYTES requires the column to be of type BYTEA, but got type {}",
                        field.data_type
                    )))
                }
            }
        }
    }
}

View on GitHub (pinned to 6469eb736d)