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

Code before:
                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
            | DataType::Int32
            | DataType::Int64
            | DataType::Int256
            | DataType::Serial => {}
            _ => {
                // why we don't allow float as text for key encode: https://github.com/risingwavelabs/risingwave/pull/16377#discussion_r1591864960
                return Err(SinkError::Config(anyhow!(
                    "The key encode is TEXT, but the primary key column {} has type {}. The key encode TEXT requires the primary key column to be of type varchar, bool, small int, int, big int, serial or rw_int256.",
                    schema_ref.name,
                    schema_ref.data_type

Review comment (suggestion):
```suggestion
                        "ENCODE BYTES requires the column to be of type BYTEA, but got type {}",
```

Code after:
                } 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
            | DataType::Int256
            | DataType::Serial => {}
            _ => {
                // why we don't allow float as text for key encode: https://github.com/risingwavelabs/risingwave/pull/16377#discussion_r1591864960
                return Err(SinkError::Config(anyhow!(
                    "The key encode is TEXT, but the primary key column {} has type {}. The key encode TEXT requires the primary key column to be of type varchar, bool, small int, int, big int, serial or rw_int256.",
                    schema_ref.name,
                    schema_ref.data_type
                )));
            }