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

Code before:
        return Err(SinkError::Config(anyhow!(
            "{}Encoder requires primary key columns to be specified",
            data_type_name
        )));
    };
    if pk_indices.len() != 1 {
        return Err(SinkError::Config(anyhow!(
            "KEY ENCODE {} expects only one primary key, but got {}",
            data_type_name,
            pk_indices.len(),
        )));
    }

    let schema_ref = params.schema.fields().get(pk_indices[0]).ok_or_else(|| {
        SinkError::Config(anyhow!(
            "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!(
                        "BYTES format requires exactly one column, got {} columns",
                        params.schema.len()
                    )));
                }

                let field = &params.schema.fields[0];

Review comment (refactor):
The nested match structure duplicates the pk_indices check logic already performed in `ensure_only_one_pk`. Consider extracting the key encoder logic to a helper function or restructuring to avoid the redundant pattern matching, which would improve code maintainability.

Code after:
        )));
    };
    if pk_indices.len() != 1 {
        return Err(SinkError::Config(anyhow!(
            "KEY ENCODE {} expects only one primary key, but got {}",
            data_type_name,
            pk_indices.len(),
        )));
    }

    let schema_ref = params.schema.fields().get(pk_indices[0]).ok_or_else(|| {
        SinkError::Config(anyhow!(
            "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 {
                    Ok(BytesEncoder::new(params.schema, 0))
                } else {