Repository: risingwavelabs/risingwave
File: src/connector/src/source/cdc/external/mysql.rs (Rust)

Code before:
                order_key,
            )
        };
        let mut conn = self.pool.get_conn().await?;
        // Set session timezone to UTC
        conn.exec_drop("SET time_zone = \"+00:00\"", ()).await?;

        if let Some(start_pk_row) = start_pk_row {
            let field_map = self
                .rw_schema
                .fields
                .iter()
                .map(|f| (f.name.as_str(), f.data_type.clone()))
                .collect::<HashMap<_, _>>();

            // fill in start primary key params
            let params: Vec<_> = primary_keys
                .iter()
                .zip_eq_fast(start_pk_row.into_iter())
                .map(|(pk, datum)| {
                    if let Some(value) = datum {
                        let ty = field_map.get(pk.as_str()).unwrap();
                        let val = match ty {
                            DataType::Boolean => Value::from(value.into_bool()),
                            DataType::Int16 => {
                                let int16_val = value.into_int16();
                                if int16_val < 0 && self.is_unsigned_type(pk.as_str()) {
                                    Value::from(self.convert_negative_to_unsigned_i16(int16_val))
                                } else {
                                    Value::from(int16_val)
                                }
                            }
                            DataType::Int32 => {
                                let int32_val = value.into_int32();
                                if int32_val < 0 && self.is_unsigned_type(pk.as_str()) {
                                    Value::from(self.convert_negative_to_unsigned_i32(int32_val))
                                } else {
                                    Value::from(int32_val)
                                }
                            }
                            DataType::Int64 => {
                                let int64_val = value.into_int64();
                                if int64_val < 0 && self.is_unsigned_type(pk.as_str()) {
                                    Value::from(self.convert_negative_to_unsigned(int64_val))
                                } else {
                                    Value::from(int64_val)
                                }
                            }
                            DataType::Float32 => Value::from(value.into_float32().into_inner()),
                            DataType::Float64 => Value::from(value.into_float64().into_inner()),
                            DataType::Varchar => Value::from(String::from(value.into_utf8())),

Review comment (refactor):
What if a later refactoring removes `int16_val < 0`? Will existing test cases catch the error, or it is safe to remove?

Code after:
            )
        };
        let mut conn = self.pool.get_conn().await?;
        // Set session timezone to UTC
        conn.exec_drop("SET time_zone = \"+00:00\"", ()).await?;

        if let Some(start_pk_row) = start_pk_row {
            let field_map = self
                .rw_schema
                .fields
                .iter()
                .map(|f| (f.name.as_str(), f.data_type.clone()))
                .collect::<HashMap<_, _>>();

            // fill in start primary key params
            let params: Vec<_> = primary_keys
                .iter()
                .zip_eq_fast(start_pk_row.into_iter())
                .map(|(pk, datum)| {
                    if let Some(value) = datum {
                        let ty = field_map.get(pk.as_str()).unwrap();
                        let val = match ty {
                            DataType::Boolean => Value::from(value.into_bool()),
                            DataType::Int16 => {
                                let int16_val = value.into_int16();
                                if int16_val < 0 && self.is_unsigned_type(pk.as_str()) {
                                    Value::from(self.convert_negative_to_unsigned_i16(int16_val))
                                } else {
                                    Value::from(int16_val)
                                }
                            }
                            DataType::Int32 => {
                                let int32_val = value.into_int32();
                                if int32_val < 0 && self.is_unsigned_type(pk.as_str()) {
                                    Value::from(self.convert_negative_to_unsigned_i32(int32_val))
                                } else {
                                    Value::from(int32_val)
                                }
                            }
                            DataType::Int64 => {
                                let int64_val = value.into_int64();
                                if int64_val < 0 && self.is_unsigned_type(pk.as_str()) {
                                    Value::from(self.convert_negative_to_unsigned(int64_val))
                                } else {
                                    Value::from(int64_val)
                                }
                            }
                            DataType::Float32 => Value::from(value.into_float32().into_inner()),
                            DataType::Float64 => Value::from(value.into_float64().into_inner()),
                            DataType::Varchar => Value::from(String::from(value.into_utf8())),
                            DataType::Date => Value::from(value.into_date().0),