Repository: risingwavelabs/risingwave
File: src/connector/src/sink/formatter/mod.rs (Rust)

Code before:

    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!(
                        "BYTES format 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!(
                        "BYTES format requires the column to be of type BYTEA, but got type {}",
                        field.data_type
                    )))
                }
            }
        }
    }
}

impl EncoderBuild for TextEncoder {
    async fn build(params: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {
        let (pk_index, schema_ref) = ensure_only_one_pk("TEXT", &params, &pk_indices)?;
        match &schema_ref.data_type() {
            DataType::Varchar
            | DataType::Boolean
            | DataType::Int16

Review comment (suggestion):
```suggestion
                        "ENCODE BYTES requires exactly one column, got {} columns",
```

Code after:
}

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 {
                    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
                    )))
                }
            }
        }
    }
}

impl EncoderBuild for TextEncoder {
    async fn build(params: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {
        let (pk_index, schema_ref) = ensure_only_one_pk("TEXT", &params, &pk_indices)?;
        match &schema_ref.data_type() {
            DataType::Varchar
            | DataType::Boolean
            | DataType::Int16
            | DataType::Int32
            | DataType::Int64