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

Code before:
                        )));
                    }

                    if let Some(channel_column) = channel_column
                        && let Some(channel_column_type) = all_map.get(channel_column)
                        && (channel_column_type != &DataType::Varchar)
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{CHANNEL_COLUMN}` must be set to `varchar`"
                        )));
                    }

                    let value_format =
                        self.format_desc.options.get(VALUE_FORMAT).ok_or_else(|| {
                            SinkError::Config(anyhow!("Cannot find `{VALUE_FORMAT}`"))
                        })?;
                    TemplateStringEncoder::check_string_format(value_format, &all_map)?;
                }
                Some(REDIS_VALUE_TYPE_STREAM) => {
                    let stream = self.format_desc.options.get(STREAM);
                    let stream_column = self.format_desc.options.get(STREAM_COLUMN);
                    if (stream.is_none() && stream_column.is_none())
                        || (stream.is_some() && stream_column.is_some())
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{STREAM}` and `{STREAM_COLUMN}` only one can be set"
                        )));
                    }

                    if let Some(stream_column) = stream_column
                        && let Some(stream_column_type) = all_map.get(stream_column)
                        && (stream_column_type != &DataType::Varchar)
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{STREAM_COLUMN}` must be set to `varchar`"
                        )));
                    }

                    let value_format =
                        self.format_desc.options.get(VALUE_FORMAT).ok_or_else(|| {
                            SinkError::Config(anyhow!("Cannot find `{VALUE_FORMAT}`"))
                        })?;
                    let key_format = self.format_desc.options.get(KEY_FORMAT).ok_or_else(|| {
                        SinkError::Config(anyhow!(
                            "Cannot find '{KEY_FORMAT}', please set it or use JSON"
                        ))
                    })?;
                    TemplateStringEncoder::check_string_format(key_format, &pk_map)?;
                    TemplateStringEncoder::check_string_format(value_format, &all_map)?;
                }
                _ => {

Review comment (suggestion):
nits:
```
Please specific either `{STREAM}` or `{STREAM_COLUMN}`. They are mutually exclusive options.
```

Code after:
                    }

                    if let Some(channel_column) = channel_column
                        && let Some(channel_column_type) = all_map.get(channel_column)
                        && (channel_column_type != &DataType::Varchar)
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{CHANNEL_COLUMN}` must be set to `varchar`"
                        )));
                    }

                    let value_format =
                        self.format_desc.options.get(VALUE_FORMAT).ok_or_else(|| {
                            SinkError::Config(anyhow!("Cannot find `{VALUE_FORMAT}`"))
                        })?;
                    TemplateStringEncoder::check_string_format(value_format, &all_map)?;
                }
                Some(REDIS_VALUE_TYPE_STREAM) => {
                    tracing::error!("test:for bug");
                    risingwave_common::license::Feature::RedisSinkStream
                        .check_available()
                        .map_err(|e| anyhow::anyhow!(e))?;
                    let stream = self.format_desc.options.get(STREAM);
                    let stream_column = self.format_desc.options.get(STREAM_COLUMN);
                    if (stream.is_none() && stream_column.is_none())
                        || (stream.is_some() && stream_column.is_some())
                    {
                        return Err(SinkError::Config(anyhow!(
                            "Please specific either `{STREAM}` or `{STREAM_COLUMN}`. They are mutually exclusive options."
                        )));
                    }

                    if let Some(stream_column) = stream_column
                        && let Some(stream_column_type) = all_map.get(stream_column)
                        && (stream_column_type != &DataType::Varchar)
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{STREAM_COLUMN}` must be set to `varchar`"
                        )));
                    }

                    let value_format =
                        self.format_desc.options.get(VALUE_FORMAT).ok_or_else(|| {
                            SinkError::Config(anyhow!("Cannot find `{VALUE_FORMAT}`"))
                        })?;
                    let key_format = self.format_desc.options.get(KEY_FORMAT).ok_or_else(|| {
                        SinkError::Config(anyhow!(
                            "Cannot find '{KEY_FORMAT}', please set it or use JSON"
                        ))
                    })?;
                    TemplateStringEncoder::check_string_format(key_format, &pk_map)?;