fix bug signedness parse All Column Order

Author: sean-k1

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

@@ -20,6 +20,8 @@
 import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * @author <a href="mailto:stanley.shyiko@gmail.com">Stanley Shyiko</a>
@@ -47,13 +49,35 @@ public TableMapEventData deserialize(ByteArrayInputStream inputStream) throws IO
             metadata = metadataDeserializer.deserialize(
                 new ByteArrayInputStream(inputStream.read(metadataLength)),
                 eventData.getColumnTypes().length,
-                numericColumnCount(eventData.getColumnTypes())
+                numericColumnCount(eventData.getColumnTypes()),
+                numericColumnIndex(eventData.getColumnTypes())
             );
         }
         eventData.setEventMetadata(metadata);
         return eventData;
     }
 
+    private List<Integer> numericColumnIndex(byte[] types){
+        ArrayList<Integer> numericColumnIndexList = new ArrayList<>();
+        for (int i = 0; i < types.length; i++) {
+            switch (ColumnType.byCode(types[i] & 0xff)) {
+                case TINY:
+                case SHORT:
+                case INT24:
+                case LONG:
+                case LONGLONG:
+                case NEWDECIMAL:
+                case FLOAT:
+                case DOUBLE:
+                case YEAR:
+                    numericColumnIndexList.add(i);
+                    break;
+                default:
+                    break;
+            }
+        }
+        return numericColumnIndexList;
+    }
     private int numericColumnCount(byte[] types) {
         int count = 0;
         for (int i = 0; i < types.length; i++) {

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

@@ -36,7 +36,10 @@ public class TableMapEventMetadataDeserializer {
 
     private final Logger logger = Logger.getLogger(getClass().getName());
 
-    public TableMapEventMetadata deserialize(ByteArrayInputStream inputStream, int nColumns, int nNumericColumns) throws IOException {
+    public TableMapEventMetadata deserialize(ByteArrayInputStream inputStream, int nColumns,
+                                             int nNumericColumns, List<Integer> numericColumIdxList)
+        throws IOException {
+
         int remainingBytes = inputStream.available();
         if (remainingBytes <= 0) {
             return null;
@@ -65,7 +68,8 @@ public TableMapEventMetadata deserialize(ByteArrayInputStream inputStream, int n
 
             switch (fieldType) {
                 case SIGNEDNESS:
-                    result.setSignedness(readBooleanList(inputStream, nNumericColumns));
+                    result.setSignedness(
+                        convertColumnOrder(readBooleanList(inputStream, nNumericColumns), numericColumIdxList));
                     break;
                 case DEFAULT_CHARSET:
                     result.setDefaultCharset(readDefaultCharset(inputStream));
@@ -108,6 +112,26 @@ public TableMapEventMetadata deserialize(ByteArrayInputStream inputStream, int n
         return result;
     }
 
+    private static BitSet convertColumnOrder(BitSet numericOrderBitSet, List<Integer> numericColumIdxList) {
+        // case SIGNEDNESS The order of the index in the packet is only the index between the numeric_columns.
+        // So we need to map the index to the index in the all columns.
+        Map<Integer, Integer> mappingColumnOrderMap = new HashMap<>();
+
+        for (int numericColumnOrder = 0; numericColumnOrder < numericColumIdxList.size();
+             numericColumnOrder++) {
+            int allColumnIndex = numericColumIdxList.get(numericColumnOrder);
+            mappingColumnOrderMap.put(numericColumnOrder, allColumnIndex);
+        }
+
+        BitSet columnOrderBitSet = new BitSet();
+        for (int i = 0; i < numericOrderBitSet.length(); i++) {
+            int numericColumnOrder = numericOrderBitSet.nextSetBit(i);
+            int allColumnIndex = mappingColumnOrderMap.get(numericColumnOrder);
+            columnOrderBitSet.set(allColumnIndex);
+        }
+        return columnOrderBitSet;
+    }
+
     private static BitSet readBooleanList(ByteArrayInputStream inputStream, int length) throws IOException {
         BitSet result = new BitSet();
         // according to MySQL internals the amount of storage required for N columns is INT((N+7)/8) bytes

src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/TableMapEventMetadataDeserializerTest.java

@@ -5,6 +5,7 @@
 import org.testng.annotations.Test;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
@@ -29,7 +30,7 @@ public void deserialize() throws IOException {
         byte[] metadataIncludingUnknownFieldType = {1, 2, 0, -128, 2, 9, 83, 6, 63, 7, 63, 8, 63, 9, 63};
         TableMapEventMetadataDeserializer deserializer = new TableMapEventMetadataDeserializer();
         TableMapEventMetadata tableMapEventMetadata =
-            deserializer.deserialize(new ByteArrayInputStream(metadataIncludingUnknownFieldType), 23, 8);
+            deserializer.deserialize(new ByteArrayInputStream(metadataIncludingUnknownFieldType), 23, 8, Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7)); // suppose there numeric Columns Idx are 0, 1, 2, 3, 4, 5, 6, 7
 
         Map<Integer, Integer> expectedCharsetCollations = new LinkedHashMap<>();
         expectedCharsetCollations.put(6, 63);