Added CompatibilityMode.INVALID_DATE_AND_TIME_AS_NEGATIVE_ONE (EventDeserializer)

Author: shyiko

src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/AbstractRowsEventDataDeserializer.java

@@ -72,7 +72,7 @@ public abstract class AbstractRowsEventDataDeserializer<T extends EventData> imp
     private final Map<Long, TableMapEventData> tableMapEventByTableId;
 
     private boolean deserializeDateAndTimeAsLong;
-    private boolean deserializeInvalidDateAndTimeAsZero;
+    private Long invalidDateAndTimeRepresentation;
     private boolean microsecondsPrecision;
     private boolean deserializeCharAndBinaryAsByteArray;
 
@@ -84,8 +84,9 @@ void setDeserializeDateAndTimeAsLong(boolean value) {
         this.deserializeDateAndTimeAsLong = value;
     }
 
-    void setDeserializeInvalidDateAndTimeAsZero(boolean value) {
-        this.deserializeInvalidDateAndTimeAsZero = value;
+    // value to return in case of 0000-00-00 00:00:00, 0000-00-00, etc.
+    void setInvalidDateAndTimeRepresentation(Long value) {
+        this.invalidDateAndTimeRepresentation = value;
     }
 
     void setMicrosecondsPrecision(boolean value) {
@@ -238,7 +239,7 @@ protected Serializable deserializeNewDecimal(int meta, ByteArrayInputStream inpu
     }
 
     private Long castTimestamp(Long timestamp, int fsp) {
-        if (timestamp != null && microsecondsPrecision) {
+        if (microsecondsPrecision && timestamp != null && timestamp > -1) {
             return timestamp * 1000 + fsp % 1000;
         }
         return timestamp;
@@ -414,7 +415,7 @@ protected byte[] deserializeJson(int meta, ByteArrayInputStream inputStream) thr
     protected Long asUnixTime(int year, int month, int day, int hour, int minute, int second, int millis) {
         // https://dev.mysql.com/doc/refman/5.0/en/datetime.html
         if (year == 0 || month == 0 || day == 0) {
-            return deserializeInvalidDateAndTimeAsZero ? 0L : null;
+            return invalidDateAndTimeRepresentation;
         }
         return UnixTime.from(year, month, day, hour, minute, second, millis);
     }

src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventDeserializer.java

@@ -156,16 +156,23 @@ private void ensureCompatibility(EventDataDeserializer eventDataDeserializer) {
         if (eventDataDeserializer instanceof AbstractRowsEventDataDeserializer) {
             AbstractRowsEventDataDeserializer deserializer =
                 (AbstractRowsEventDataDeserializer) eventDataDeserializer;
-            deserializer.setDeserializeDateAndTimeAsLong(
+            boolean deserializeDateAndTimeAsLong =
                 compatibilitySet.contains(CompatibilityMode.DATE_AND_TIME_AS_LONG) ||
-                compatibilitySet.contains(CompatibilityMode.DATE_AND_TIME_AS_LONG_MICRO)
-            );
+                compatibilitySet.contains(CompatibilityMode.DATE_AND_TIME_AS_LONG_MICRO);
+            deserializer.setDeserializeDateAndTimeAsLong(deserializeDateAndTimeAsLong);
             deserializer.setMicrosecondsPrecision(
                 compatibilitySet.contains(CompatibilityMode.DATE_AND_TIME_AS_LONG_MICRO)
             );
-            deserializer.setDeserializeInvalidDateAndTimeAsZero(
-                compatibilitySet.contains(CompatibilityMode.INVALID_DATE_AND_TIME_AS_ZERO)
-            );
+            if (compatibilitySet.contains(CompatibilityMode.INVALID_DATE_AND_TIME_AS_ZERO)) {
+                deserializer.setInvalidDateAndTimeRepresentation(0L);
+            }
+            if (compatibilitySet.contains(CompatibilityMode.INVALID_DATE_AND_TIME_AS_NEGATIVE_ONE)) {
+                if (!deserializeDateAndTimeAsLong) {
+                    throw new IllegalArgumentException("INVALID_DATE_AND_TIME_AS_NEGATIVE_ONE requires " +
+                        "DATE_AND_TIME_AS_LONG or DATE_AND_TIME_AS_LONG_MICRO");
+                }
+                deserializer.setInvalidDateAndTimeRepresentation(-1L);
+            }
             deserializer.setDeserializeCharAndBinaryAsByteArray(
                 compatibilitySet.contains(CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY)
             );
@@ -248,10 +255,15 @@ public enum CompatibilityMode {
         /**
          * Return 0 instead of null if year/month/day is 0.
          * Affects DATETIME/DATETIME_V2/DATE/TIME/TIME_V2.
+         */
+        INVALID_DATE_AND_TIME_AS_ZERO,
+        /**
+         * Return -1 instead of null if year/month/day is 0.
+         * Affects DATETIME/DATETIME_V2/DATE/TIME/TIME_V2.
          *
          * <p>This option is going to be enabled by default starting from mysql-binlog-connector-java@1.0.0.
          */
-        INVALID_DATE_AND_TIME_AS_ZERO,
+        INVALID_DATE_AND_TIME_AS_NEGATIVE_ONE,
         /**
          * Return CHAR/VARCHAR/BINARY/VARBINARY values as byte[]|s (instead of String|s).
          *