tell go-mysql to leave those times alone

Author: serprex

flow/connectors/mysql/cdc.go

@@ -235,7 +235,7 @@ func (c *MySqlConnector) startSyncer(ctx context.Context) (*replication.BinlogSy
 		Logger:     logger,
 		Dialer:     c.Dialer(),
 		UseDecimal: true,
-		ParseTime:  true,
+		ParseTime:  false,
 		TLSConfig:  tlsConfig,
 	}), nil
 }

flow/connectors/mysql/qvalue_convert.go

@@ -469,13 +469,13 @@ func QValueFromMysqlRowEvent(
 			return types.QValueTime{Val: tm}, nil
 		case types.QValueKindDate:
 			if val == "0000-00-00" {
-				return types.QValueDate{Val: time.Unix(0, 0)}, nil
+				return types.QValueDate{Val: time.Unix(0, 0).UTC()}, nil
 			}
 			val, err := time.Parse(time.DateOnly, strings.ReplaceAll(val, "-00", "-01"))
 			if err != nil {
 				return nil, err
 			}
-			return types.QValueDate{Val: val}, nil
+			return types.QValueDate{Val: val.UTC()}, nil
 		case types.QValueKindTimestamp: // 0000-00-00 ends up here
 			if mytype == mysql.MYSQL_TYPE_TIME || mytype == mysql.MYSQL_TYPE_TIME2 {
 				tm, err := processTime(val)
@@ -485,13 +485,13 @@ func QValueFromMysqlRowEvent(
 				return types.QValueTimestamp{Val: time.Unix(0, 0).UTC().Add(tm)}, nil
 			}
 			if strings.HasPrefix(val, "0000-00-00") {
-				return types.QValueTimestamp{Val: time.Unix(0, 0)}, nil
+				return types.QValueTimestamp{Val: time.Unix(0, 0).UTC()}, nil
 			}
 			tm, err := time.Parse("2006-01-02 15:04:05.999999", strings.ReplaceAll(val, "-00", "-01"))
 			if err != nil {
 				return nil, err
 			}
-			return types.QValueTimestamp{Val: tm}, nil
+			return types.QValueTimestamp{Val: tm.UTC()}, nil
 		}
 	}
 	return nil, fmt.Errorf("unexpected type %T for mysql type %d, qkind %s", val, mytype, qkind)