fix(connector): handle backwards compat parsing for boolean datatype (

#19251)

Co-authored-by: William Wen <[email protected]>

e2e_test/source_legacy/cdc/cdc.share_stream.slt

@@ -232,10 +232,10 @@ SELECT order_id,order_date,customer_name,product_id FROM orders_test order by or
 10003 2020-07-30 12:00:30 Edward 106
 
 query IIIIITTTTTTTTT
-SELECT c_tinyint, c_smallint, c_mediumint, c_integer, c_bigint, c_decimal, c_float, c_double, c_char_255, c_varchar_10000, c_date, c_time, c_datetime, c_timestamp FROM mysql_all_types order by c_bigint;
+SELECT c_boolean, c_bit, c_tinyint, c_smallint, c_mediumint, c_integer, c_bigint, c_decimal, c_float, c_double, c_char_255, c_varchar_10000, c_date, c_time, c_datetime, c_timestamp FROM mysql_all_types order by c_bigint;
 ----
--128 -32767 -8388608 -2147483647 -9223372036854775807 -10 -10000 -10000 a b 1001-01-01 00:00:00 1998-01-01 00:00:00 1970-01-01 00:00:01+00:00
-NULL NULL -8388608 -2147483647 9223372036854775806 -10 -10000 -10000 c d 1001-01-01 NULL 2000-01-01 00:00:00 NULL
+t t -128 -32767 -8388608 -2147483647 -9223372036854775807 -10 -10000 -10000 a b 1001-01-01 00:00:00 1998-01-01 00:00:00 1970-01-01 00:00:01+00:00
+f f NULL NULL -8388608 -2147483647 9223372036854775806 -10 -10000 -10000 c d 1001-01-01 NULL 2000-01-01 00:00:00 NULL
 
 statement ok
 create secret pg_pwd with (

src/connector/src/parser/mysql.rs

@@ -75,6 +75,14 @@ pub fn mysql_datum_to_rw_datum(
 ) -> Result<Datum, anyhow::Error> {
     match rw_data_type {
         DataType::Boolean => {
+            // TinyInt(1) is used to represent boolean in MySQL
+            // This handles backwards compatibility,
+            // before https://github.com/risingwavelabs/risingwave/pull/19071
+            // we permit boolean and tinyint(1) to be equivalent to boolean in RW.
+            if let Some(Ok(val)) = mysql_row.get_opt::<Option<bool>, _>(mysql_datum_index) {
+                let _ = mysql_row.take::<bool, _>(mysql_datum_index);
+                return Ok(val.map(ScalarImpl::from));
+            }
             // Bit(1)
             match mysql_row.take_opt::<Option<Vec<u8>>, _>(mysql_datum_index) {
                 None => bail!(