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
 			}
@@ -471,7 +471,7 @@ func QValueFromMysqlRowEvent(
 			if val == "0000-00-00" {
 				return types.QValueDate{Val: time.Unix(0, 0)}, 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
 			}
@@ -487,7 +487,7 @@ func QValueFromMysqlRowEvent(
 			if strings.HasPrefix(val, "0000-00-00") {
 				return types.QValueTimestamp{Val: time.Unix(0, 0)}, 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
 			}

flow/e2e/clickhouse/mysql_test.go

@@ -85,7 +85,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-00-00 00:00:00.000','-800:0:1')`,
 		quotedSrcFullName)))
 
 	connectionGen := e2e.FlowConnectionGenerationConfig{
@@ -105,7 +105,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-00-00 00:00:00.000','-800:0:1')`,
 		quotedSrcFullName)))
 
 	e2e.EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc", srcTableName, dstTableName, "id,\"key\",d,dt,tm,t")