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

Code before:
            TemplateRedisPubSubStreamKeyEncoderInner::PubSubStreamColumnIndex(pubsub_col) => {
                let pubsub_key = row
                    .datum_at(*pubsub_col)
                    .ok_or_else(|| SinkError::Redis("pubsub_key is null".to_owned()))?
                    .to_text()
                    .clone();
                Ok(TemplateEncoderOutput::RedisPubSubStreamKey(pubsub_key))
            }
        }
    }
}

pub struct TemplateRedisStreamValueEncoder {
    key_encoder: TemplateStringEncoder,
    value_encoder: TemplateStringEncoder,
}

impl TemplateRedisStreamValueEncoder {
    pub fn new(
        schema: Schema,
        col_indices: Option<Vec<usize>>,
        key_template: String,
        value_template: String,
    ) -> Self {
        let key_encoder =
            TemplateStringEncoder::new(schema.clone(), col_indices.clone(), key_template.clone());
        let value_encoder = TemplateStringEncoder::new(schema, col_indices, value_template);
        Self {
            key_encoder,
            value_encoder,
        }
    }

    pub fn encode_cols(
        &self,
        row: impl Row,
        col_indices: impl Iterator<Item = usize>,
    ) -> Result<TemplateEncoderOutput> {
        let col_indices: Vec<_> = col_indices.collect();
        let key = self
            .key_encoder
            .encode_cols(&row, col_indices.clone().into_iter())?;
        let value = self
            .value_encoder
            .encode_cols(row, col_indices.into_iter())?;
        Ok(TemplateEncoderOutput::RedisStreamValue((key, value)))
    }
}

pub enum TemplateEncoderOutput {
    // String formatted according to the template

Review comment (suggestion):
Unnecessary clone of `key_template` parameter. Since `key_template` is moved into this function, you can use it directly without cloning.
```suggestion
            TemplateStringEncoder::new(schema.clone(), col_indices.clone(), key_template);
```

Code after:
                let pubsub_key = row
                    .datum_at(*pubsub_col)
                    .ok_or_else(|| SinkError::Redis("pubsub_key is null".to_owned()))?
                    .to_text();
                Ok(TemplateEncoderOutput::RedisPubSubStreamKey(pubsub_key))
            }
        }
    }
}

pub struct TemplateRedisStreamValueEncoder {
    key_encoder: TemplateStringEncoder,
    value_encoder: TemplateStringEncoder,
}

impl TemplateRedisStreamValueEncoder {
    pub fn new(
        schema: Schema,
        col_indices: Option<Vec<usize>>,
        key_template: String,
        value_template: String,
    ) -> Self {
        let key_encoder =
            TemplateStringEncoder::new(schema.clone(), col_indices.clone(), key_template);
        let value_encoder = TemplateStringEncoder::new(schema, col_indices, value_template);
        Self {
            key_encoder,
            value_encoder,
        }
    }

    pub fn encode_cols(
        &self,
        row: impl Row,
        col_indices: impl Iterator<Item = usize>,
    ) -> Result<TemplateEncoderOutput> {
        let col_indices: Vec<_> = col_indices.collect();
        let key = self
            .key_encoder
            .encode_cols(&row, col_indices.clone().into_iter())?;
        let value = self
            .value_encoder
            .encode_cols(row, col_indices.into_iter())?;
        Ok(TemplateEncoderOutput::RedisStreamValue((key, value)))
    }
}

pub enum TemplateEncoderOutput {
    // String formatted according to the template
    String(String),
    // The value of redis's geospatial, including longitude and latitude