mysql: map 00 day/month to 01

Author: serprex

flow/connectors/mysql/qvalue_convert.go

@@ -316,7 +316,7 @@ func QValueFromMysqlFieldValue(qkind types.QValueKind, mytype byte, fv mysql.Fie
 			if strings.HasPrefix(unsafeString, "0000-00-00") {
 				return types.QValueTimestamp{Val: time.Unix(0, 0)}, nil
 			}
-			val, err := time.Parse("2006-01-02 15:04:05.999999", unsafeString)
+			val, err := time.Parse("2006-01-02 15:04:05.999999", strings.ReplaceAll(unsafeString, "-00", "-01"))
 			if err != nil {
 				return nil, err
 			}
@@ -334,7 +334,7 @@ func QValueFromMysqlFieldValue(qkind types.QValueKind, mytype byte, fv mysql.Fie
 			if unsafeString == "0000-00-00" {
 				return types.QValueDate{Val: time.Unix(0, 0)}, nil
 			}
-			val, err := time.Parse(time.DateOnly, unsafeString)
+			val, err := time.Parse(time.DateOnly, strings.ReplaceAll(unsafeString, "-00", "-01"))
 			if err != nil {
 				return nil, err
 			}
@@ -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, val)
+			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", val)
+			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)

flow/e2e/clickhouse/mysql_test.go

@@ -88,7 +88,7 @@ func (s ClickHouseSuite) Test_MySQL_Time() {
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`INSERT INTO %s ("key",d,dt,tm,t) VALUES
 		('init','1935-01-01','1953-02-02 12:01:02','1973-02-02 13:01:02.123','14:21.654321'),
 		('init','0000-00-00','0000-00-00 00:00:00','0000-00-00 00:00:00.000','00:00'),
-		('init','0000-00-00','0000-00-00 00:00:00','0000-00-00 00:00:00.000','-800:0:1')`,
+		('init','2000-01-00','2000-00-01 00:00:00','2000-01-01 00:00:00.000','-800:0:1')`,
 		quotedSrcFullName)))
 
 	connectionGen := e2e.FlowConnectionGenerationConfig{
@@ -108,7 +108,7 @@ func (s ClickHouseSuite) Test_MySQL_Time() {
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`INSERT INTO %s ("key",d,dt,tm,t) VALUES
 		('cdc','1935-01-01','1953-02-02 12:01:02','1973-02-02 13:01:02.123','14:21.654321'),
 		('cdc','0000-00-00','0000-00-00 00:00:00','0000-00-00 00:00:00.000','00:00'),
-		('cdc','0000-00-00','0000-00-00 00:00:00','0000-00-00 00:00:00.000','-800:0:1')`,
+		('cdc','2000-01-00','2000-00-01 00:00:00','2000-01-01 00:00:00.000','-800:0:1')`,
 		quotedSrcFullName)))
 
 	e2e.EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc", srcTableName, dstTableName, "id,\"key\",d,dt,tm,t")