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

Code before:
    // `BIGINT(20) UNSIGNED`, `INT UNSIGNED ZEROFILL`, etc.
    let ty = ty_name.trim().to_lowercase();
    let base = ty
        .split_whitespace()
        .next()
        .unwrap_or_default()
        .split('(')
        .next()
        .unwrap_or_default();
    let is_unsigned = ty.contains("unsigned");

    // Handle MySQL `... UNSIGNED` integer family in Debezium schema change.
    //
    // Notes about mapping:
    // - `TINYINT UNSIGNED`  : 0..255            -> fits in RW Int16
    // - `SMALLINT UNSIGNED` : 0..65535          -> needs RW Int32
    // - `MEDIUMINT UNSIGNED`: 0..16777215       -> fits in RW Int32
    // - `INT UNSIGNED`      : 0..4294967295     -> needs RW Int64
    // - `BIGINT UNSIGNED`   : 0..18446744073709551615 -> needs RW Decimal
    if is_unsigned {
        match base {
            // Keep as-is: RW mapping is already wide enough.
            "tinyint" => return Some(ColumnType::TinyInt(Default::default())),
            "mediumint" => return Some(ColumnType::MediumInt(Default::default())),
            // Promote to a wider signed type.
            "smallint" => return Some(ColumnType::Int(Default::default())),
            "int" => return Some(ColumnType::BigInt(Default::default())),
            // Promote to Decimal to avoid overflow beyond i64.
            "bigint" => return Some(ColumnType::Decimal(Default::default())),
            _ => {}
        }
    }

    match base {
        "bit" => Some(ColumnType::Bit(Default::default())),
        "tinyint" => Some(ColumnType::TinyInt(Default::default())),
        "smallint" => Some(ColumnType::SmallInt(Default::default())),
        "mediumint" => Some(ColumnType::MediumInt(Default::default())),
        "int" => Some(ColumnType::Int(Default::default())),
        "bigint" => Some(ColumnType::BigInt(Default::default())),
        "decimal" => Some(ColumnType::Decimal(Default::default())),
        "float" => Some(ColumnType::Float(Default::default())),
        "double" => Some(ColumnType::Double(Default::default())),
        "time" => Some(ColumnType::Time(Default::default())),
        "datetime" => Some(ColumnType::DateTime(Default::default())),
        "timestamp" => Some(ColumnType::Timestamp(Default::default())),
        "char" => Some(ColumnType::Char(Default::default())),
        "nchar" => Some(ColumnType::NChar(Default::default())),
        "varchar" => Some(ColumnType::Varchar(Default::default())),
        "nvarchar" => Some(ColumnType::NVarchar(Default::default())),
        "binary" => Some(ColumnType::Binary(Default::default())),

Review comment (refactor):
This does not look like the correct abstraction. Given the function signature, it is returning a `sea_schema::mysql::def::ColumnType` rather than a `risingwave_common::types::DataType`. This function shall return MySQL types faithfully and populate `NumericAttr::unsigned` (instead of `Default::default()`) rather than assuming it has only one caller and the caller want RisingWave data types eventually.

Code after:
    // `BIGINT(20) UNSIGNED`, `INT UNSIGNED ZEROFILL`, etc.
    let ty = ty_name.trim().to_lowercase();
    let base = ty
        .split_whitespace()
        .next()
        .unwrap_or_default()
        .split('(')
        .next()
        .unwrap_or_default();
    let is_unsigned = ty.contains("unsigned");
    let is_zero_fill = ty.contains("zerofill");

    let make_numeric_attr = || {
        let mut attr = NumericAttr::default();
        if is_unsigned {
            attr.unsigned = Some(true);
        }
        if is_zero_fill {
            attr.zero_fill = Some(true);
        }
        attr
    };

    match base {
        "bit" => Some(ColumnType::Bit(make_numeric_attr())),
        "tinyint" => Some(ColumnType::TinyInt(make_numeric_attr())),
        "smallint" => Some(ColumnType::SmallInt(make_numeric_attr())),
        "mediumint" => Some(ColumnType::MediumInt(make_numeric_attr())),
        "int" => Some(ColumnType::Int(make_numeric_attr())),
        "bigint" => Some(ColumnType::BigInt(make_numeric_attr())),
        "decimal" => Some(ColumnType::Decimal(make_numeric_attr())),
        "float" => Some(ColumnType::Float(make_numeric_attr())),
        "double" => Some(ColumnType::Double(make_numeric_attr())),
        "time" => Some(ColumnType::Time(Default::default())),
        "datetime" => Some(ColumnType::DateTime(Default::default())),
        "timestamp" => Some(ColumnType::Timestamp(Default::default())),
        "char" => Some(ColumnType::Char(Default::default())),
        "nchar" => Some(ColumnType::NChar(Default::default())),
        "varchar" => Some(ColumnType::Varchar(Default::default())),
        "nvarchar" => Some(ColumnType::NVarchar(Default::default())),
        "binary" => Some(ColumnType::Binary(Default::default())),
        "varbinary" => Some(ColumnType::Varbinary(Default::default())),
        "text" => Some(ColumnType::Text(Default::default())),
        "tinytext" => Some(ColumnType::TinyText(Default::default())),
        "mediumtext" => Some(ColumnType::MediumText(Default::default())),
        "longtext" => Some(ColumnType::LongText(Default::default())),
        "blob" => Some(ColumnType::Blob(Default::default())),
        "tinyblob" => Some(ColumnType::TinyBlob),
        "mediumblob" => Some(ColumnType::MediumBlob),
        "longblob" => Some(ColumnType::LongBlob),
        "enum" => Some(ColumnType::Enum(Default::default())),