fixed issue #112 , add mysql 5.6 useconds parse

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java

@@ -24,6 +24,7 @@ public final class RowsLogBuffer {
 
     public static final long   DATETIMEF_INT_OFS = 0x8000000000L;
     public static final long   TIMEF_INT_OFS     = 0x800000L;
+    public static final long   TIMEF_OFS         = 0x800000000000L;
     private final LogBuffer    buffer;
     private final int          columnLen;
     private final String       charsetName;
@@ -437,14 +438,23 @@ public final class RowsLogBuffer {
                         break;
                 }
 
+                String second = null;
                 if (tv_sec == 0) {
-                    value = "0000-00-00 00:00:00";
+                    second = "0000-00-00 00:00:00";
                 } else {
                     Timestamp time = new Timestamp(tv_sec * 1000);
-                    time.setNanos(tv_usec * 1000);
+                    second = time.toString();
-                    String v = time.toString();
+                    second = second.substring(0, second.length() - 2);// 去掉毫秒精度.0
-                    value = v.substring(0, v.length() - 2);
                 }
+
+                if (meta >= 1) {
+                    String microSecond = usecondsToStr(tv_usec, meta);
+                    microSecond = microSecond.substring(0, meta);
+                    value = second + '.' + microSecond;
+                } else {
+                    value = second;
+                }
+
                 javaType = Types.TIMESTAMP;
                 length = 4 + (meta + 1) / 2;
                 break;
@@ -487,7 +497,6 @@ public final class RowsLogBuffer {
                  * .YYdddddh.hhhhmmmm.mmssssss.ffffffff.ffffffff.ffffffff
                  */
                 long intpart = buffer.getBeUlong40() - DATETIMEF_INT_OFS; // big-endian
-                @SuppressWarnings("unused")
                 int frac = 0;
                 switch (meta) {
                     case 0:
@@ -510,8 +519,9 @@ public final class RowsLogBuffer {
                         break;
                 }
 
+                String second = null;
                 if (intpart == 0) {
-                    value = "0000-00-00 00:00:00";
+                    second = "0000-00-00 00:00:00";
                 } else {
                     // 构造TimeStamp只处理到秒
                     long ymd = intpart >> 17;
@@ -524,7 +534,7 @@ public final class RowsLogBuffer {
                     // % (1 << 5)), (int) (hms >> 12),
                     // (int) ((hms >> 6) % (1 << 6)), (int) (hms % (1 << 6)));
                     // value = new Timestamp(cal.getTimeInMillis());
-                    value = String.format("%04d-%02d-%02d %02d:%02d:%02d",
+                    second = String.format("%04d-%02d-%02d %02d:%02d:%02d",
                         (int) (ym / 13),
                         (int) (ym % 13),
                         (int) (ymd % (1 << 5)),
@@ -532,6 +542,15 @@ public final class RowsLogBuffer {
                         (int) ((hms >> 6) % (1 << 6)),
                         (int) (hms % (1 << 6)));
                 }
+
+                if (meta >= 1) {
+                    String microSecond = usecondsToStr(frac, meta);
+                    microSecond = microSecond.substring(0, meta);
+                    value = second + '.' + microSecond;
+                } else {
+                    value = second;
+                }
+
                 javaType = Types.TIMESTAMP;
                 length = 5 + (meta + 1) / 2;
                 break;
@@ -603,7 +622,8 @@ public final class RowsLogBuffer {
                             frac -= 0x100; /* -(0x100 - frac) */
                             // fraclong = frac * 10000;
                         }
-                        ltime = intpart << 24 + frac * 10000;
+                        frac = frac * 10000;
+                        ltime = intpart << 24;
                         break;
                     case 3:
                     case 4:
@@ -619,12 +639,14 @@ public final class RowsLogBuffer {
                             frac -= 0x10000; /* -(0x10000-frac) */
                             // fraclong = frac * 100;
                         }
-                        ltime = intpart << 24 + frac * 100;
+                        frac = frac * 100;
+                        ltime = intpart << 24;
                         break;
                     case 5:
                     case 6:
-                        intpart = buffer.getBeUlong48() - TIMEF_INT_OFS;
+                        intpart = buffer.getBeUlong48() - TIMEF_OFS;
                         ltime = intpart;
+                        frac = (int) (intpart % (1L << 24));
                         break;
                     default:
                         intpart = buffer.getBeUint24() - TIMEF_INT_OFS;
@@ -632,8 +654,9 @@ public final class RowsLogBuffer {
                         break;
                 }
 
+                String second = null;
                 if (intpart == 0) {
-                    value = "00:00:00";
+                    second = "00:00:00";
                 } else {
                     // 目前只记录秒，不处理us frac
                     // if (cal == null) cal = Calendar.getInstance();
@@ -644,13 +667,21 @@ public final class RowsLogBuffer {
                     // value = new Time(cal.getTimeInMillis());
                     long ultime = Math.abs(ltime);
                     intpart = ultime >> 24;
-                    value = String.format("%s%02d:%02d:%02d",
+                    second = String.format("%s%02d:%02d:%02d",
                         ltime >= 0 ? "" : "-",
                         (int) ((intpart >> 12) % (1 << 10)),
                         (int) ((intpart >> 6) % (1 << 6)),
                         (int) (intpart % (1 << 6)));
                 }
 
+                if (meta >= 1) {
+                    String microSecond = usecondsToStr(Math.abs(frac), meta);
+                    microSecond = microSecond.substring(0, meta);
+                    value = second + '.' + microSecond;
+                } else {
+                    value = second;
+                }
+
                 javaType = Types.TIME;
                 length = 3 + (meta + 1) / 2;
                 break;
@@ -973,4 +1004,23 @@ public final class RowsLogBuffer {
     public final int getLength() {
         return length;
     }
+
+    private String usecondsToStr(int frac, int meta) {
+        String sec = String.valueOf(frac);
+        if (meta > 6) {
+            throw new IllegalArgumentException("unknow useconds meta : " + meta);
+        }
+
+        if (sec.length() < 6) {
+            StringBuilder result = new StringBuilder(6);
+            int len = 6 - sec.length();
+            for (; len > 0; len--) {
+                result.append('0');
+            }
+            result.append(sec);
+            sec = result.toString();
+        }
+
+        return sec.substring(0, meta);
+    }
 }

dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java

@@ -25,12 +25,12 @@ public class DirectLogFetcherTest extends BaseLogFetcherTest {
         DirectLogFetcher fecther = new DirectLogFetcher();
         try {
             Class.forName("com.mysql.jdbc.Driver");
-            Connection connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306", "xxxxx", "xxxxx");
+            Connection connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306", "root", "hello");
             Statement statement = connection.createStatement();
             statement.execute("SET @master_binlog_checksum='@@global.binlog_checksum'");
             statement.execute("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'");
 
-            fecther.open(connection, "mysql-bin.000003", 4L, 2);
+            fecther.open(connection, "mysql-bin.000010", 4L, 2);
 
             LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
             LogContext context = new LogContext();
@@ -39,7 +39,8 @@ public class DirectLogFetcherTest extends BaseLogFetcherTest {
                 event = decoder.decode(fecther, context);
 
                 if (event == null) {
-                    throw new RuntimeException("parse failed");
+                    continue;
+                    // throw new RuntimeException("parse failed");
                 }
 
                 int eventType = event.getHeader().getType();