diff --git a/replication/row_event.go b/replication/row_event.go
index a83a73ca5..1972cb217 100644
--- a/replication/row_event.go
+++ b/replication/row_event.go
@@ -1316,7 +1316,7 @@ func (e *RowsEvent) decodeValue(data []byte, tp byte, meta uint16, isPartial boo
 		n = 4
 		t := binary.LittleEndian.Uint32(data)
 		if t == 0 {
-			v = formatZeroTime(0, 0)
+			v = "0000-00-00 00:00:00"
 		} else {
 			v = e.parseFracTime(fracTime{
 				Time:                    time.Unix(int64(t), 0),
@@ -1331,26 +1331,37 @@ func (e *RowsEvent) decodeValue(data []byte, tp byte, meta uint16, isPartial boo
 		n = 8
 		i64 := binary.LittleEndian.Uint64(data)
 		if i64 == 0 {
-			v = formatZeroTime(0, 0)
+			v = "0000-00-00 00:00:00"
 		} else {
 			d := i64 / 1000000
 			t := i64 % 1000000
-			v = e.parseFracTime(fracTime{
-				Time: time.Date(
-					int(d/10000),
-					time.Month((d%10000)/100),
-					int(d%100),
-					int(t/10000),
-					int((t%10000)/100),
-					int(t%100),
-					0,
-					time.UTC,
-				),
-				Dec: 0,
-			})
+			years := int(d / 10000)
+			months := int(d%10000) / 100
+			days := int(d % 100)
+			hours := int(t / 10000)
+			minutes := int(t%10000) / 100
+			seconds := int(t % 100)
+			if !e.parseTime || months == 0 || days == 0 {
+				v = fmt.Sprintf("%04d-%02d-%02d %02d:%02d:%02d",
+					years, months, days, hours, minutes, seconds)
+			} else {
+				v = e.parseFracTime(fracTime{
+					Time: time.Date(
+						years,
+						time.Month(months),
+						days,
+						hours,
+						minutes,
+						seconds,
+						0,
+						time.UTC,
+					),
+					Dec: 0,
+				})
+			}
 		}
 	case mysql.MYSQL_TYPE_DATETIME2:
-		v, n, err = decodeDatetime2(data, meta)
+		v, n, err = decodeDatetime2(data, meta, e.parseTime)
 		v = e.parseFracTime(v)
 	case mysql.MYSQL_TYPE_TIME:
 		n = 3
@@ -1675,7 +1686,7 @@ func decodeTimestamp2(data []byte, dec uint16, timestampStringLocation *time.Loc
 
 const DATETIMEF_INT_OFS int64 = 0x8000000000
 
-func decodeDatetime2(data []byte, dec uint16) (interface{}, int, error) {
+func decodeDatetime2(data []byte, dec uint16, parseTime bool) (interface{}, int, error) {
 	// get datetime binary length
 	n := int(5 + (dec+1)/2)
 
@@ -1725,8 +1736,8 @@ func decodeDatetime2(data []byte, dec uint16) (interface{}, int, error) {
 	// minute = 0 = 0b000000
 	// second = 0 = 0b000000
 	// integer value = 0b1100100000010110000100000000000000000 = 107420450816
-	if intPart < 107420450816 {
-		return formatBeforeUnixZeroTime(year, month, day, hour, minute, second, int(frac), int(dec)), n, nil
+	if !parseTime || intPart < 107420450816 || month == 0 || day == 0 {
+		return formatDatetime(year, month, day, hour, minute, second, int(frac), int(dec)), n, nil
 	}
 
 	return fracTime{
diff --git a/replication/row_event_test.go b/replication/row_event_test.go
index 7b004c50a..60e5cbb9b 100644
--- a/replication/row_event_test.go
+++ b/replication/row_event_test.go
@@ -675,6 +675,8 @@ func TestDecodeDatetime2(t *testing.T) {
 	}{
 		{[]byte("\xfe\xf3\xff\x7e\xfb"), 0, true, "9999-12-31 23:59:59"},
 		{[]byte("\x99\x9a\xb8\xf7\xaa"), 0, true, "2016-10-28 15:30:42"},
+		{[]byte("\x99\x98\x38\xf7\xaa"), 0, false, "2016-00-28 15:30:42"},
+		{[]byte("\x99\x9a\x80\xf7\xaa"), 0, false, "2016-10-00 15:30:42"},
 		{[]byte("\x99\x02\xc2\x00\x00"), 0, true, "1970-01-01 00:00:00"},
 		{[]byte("\x80\x00\x00\x00\x00"), 0, false, "0000-00-00 00:00:00"},
 		{[]byte("\x80\x00\x02\xf1\x05"), 0, false, "0000-00-01 15:04:05"},
@@ -684,17 +686,20 @@ func TestDecodeDatetime2(t *testing.T) {
 		{[]byte("\x80\x03\x82\x00\x00\x01\xe2\x40"), uint16(6), false, "0001-01-01 00:00:00.123456"},
 	}
 	for _, tc := range testcases {
-		value, _, err := decodeDatetime2(tc.data, tc.dec)
-		require.NoError(t, err)
-		switch v := value.(type) {
-		case fracTime:
-			require.True(t, tc.getFracTime)
-			require.Equal(t, tc.expected, v.String())
-		case string:
-			require.False(t, tc.getFracTime)
-			require.Equal(t, tc.expected, v)
-		default:
-			require.FailNow(t, "invalid value type: %T", value)
+		for _, parseTime := range []bool{true, false} {
+			value, _, err := decodeDatetime2(tc.data, tc.dec, parseTime)
+			require.NoError(t, err)
+			switch v := value.(type) {
+			case fracTime:
+				require.True(t, parseTime)
+				require.True(t, tc.getFracTime)
+				require.Equal(t, tc.expected, v.String())
+			case string:
+				require.False(t, parseTime && tc.getFracTime)
+				require.Equal(t, tc.expected, v)
+			default:
+				require.FailNow(t, "invalid value type: %T", value)
+			}
 		}
 	}
 }
diff --git a/replication/time.go b/replication/time.go
index 948235666..cfc0a8585 100644
--- a/replication/time.go
+++ b/replication/time.go
@@ -44,7 +44,7 @@ func formatZeroTime(frac int, dec int) string {
 	return s[0 : len(s)-(6-dec)]
 }
 
-func formatBeforeUnixZeroTime(year, month, day, hour, minute, second, frac, dec int) string {
+func formatDatetime(year, month, day, hour, minute, second, frac, dec int) string {
 	if dec == 0 {
 		return fmt.Sprintf("%04d-%02d-%02d %02d:%02d:%02d", year, month, day, hour, minute, second)
 	}