Repository: apache/shardingsphere
Issue: The result of `getBytes` is wrong.

## Bug Report

### Which version of Sharding-Sphere did you use?

3.0.0

### Which project did you use? Sharding-JDBC or Sharding-Proxy?

Sharding-Proxy

### Expected behavior

The result of bytes should be:
[88, -99, -9, 80, 65, -64, 65, -70, -83, -98, 49, 15, -115, -103, 13, 116]

### Actual behavior

The result change every time:
[91, 66, 64, 49, 50, 57, 52, 99, 50, 51, 52]
[91, 66, 64, 55, 50, 101, 49, 56, 57, 100, 54]
[91, 66, 64, 52, 54, 55, 54, 98, 48, 53, 56]
[91, 66, 64, 51, 48, 52, 51, 98, 49, 99, 99]
[91, 66, 64, 49, 97, 99, 102, 97, 99, 53, 56]

### Reason analyze (If you can)

### Steps to reproduce the behavior, such as: SQL to execute, sharding rule configuration, when exception occur etc.

```sql
CREATE TABLE `binary_test` (
  `id` binary(16) NOT NULL
) ENGINE=InnoDB;

insert into binary_test (id) value (0x589df75041c041baad9e310f8d990d74);
```

### Example codes for reproduce this issue (such as a github link).

```java
        Connection connection = ds.getConnection();
        PreparedStatement ps = connection.prepareStatement("SELECT * FROM binary_test");
        ps.execute();
        ResultSet rs = ps.getResultSet();
        rs.next();
        byte[] result = rs.getBytes("id");
        System.out.println(Arrays.toString(result));
```

Changed files:
sharding-proxy/src/test/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnTypeTest.java
sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/binary/execute/protocol/StringLenencBinaryProtocolValue.java
sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/MySQLPacketPayload.java
sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/text/TextResultSetRowPacket.java
sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnType.java

diff --git a/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnType.java b/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnType.java
index e65b43894bc..fdd19647ffd 100644
--- a/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnType.java
+++ b/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnType.java
@@ -116,9 +116,9 @@ public enum ColumnType {
         JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.DATE, MYSQL_TYPE_DATE);
         JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.TIME, MYSQL_TYPE_TIME);
         JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.TIMESTAMP, MYSQL_TYPE_TIMESTAMP);
-        JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.BINARY, MYSQL_TYPE_BLOB);
-        JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.VARBINARY, MYSQL_TYPE_MEDIUM_BLOB);
-        JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.LONGVARBINARY, MYSQL_TYPE_LONG_BLOB);
+        JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.BINARY, MYSQL_TYPE_VARCHAR);
+        JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.VARBINARY, MYSQL_TYPE_VARCHAR);
+        JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.LONGVARBINARY, MYSQL_TYPE_VARCHAR);
         JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.NULL, MYSQL_TYPE_NULL);
         JDBC_TYPE_AND_COLUMN_TYPE_MAP.put(Types.BLOB, MYSQL_TYPE_BLOB);
     }
diff --git a/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/MySQLPacketPayload.java b/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/MySQLPacketPayload.java
index 717b08945c9..5d5f79d0318 100644
--- a/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/MySQLPacketPayload.java
+++ b/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/MySQLPacketPayload.java
@@ -266,6 +266,20 @@ public final class MySQLPacketPayload implements AutoCloseable {
         byteBuf.writeBytes(value.getBytes());
     }
     
+    /**
+     * Write lenenc bytes to byte buffers.
+     *
+     * @param value fixed length bytes
+     */
+    public void writeBytesLenenc(final byte[] value) {
+        if (0 == value.length) {
+            byteBuf.writeByte(0);
+            return;
+        }
+        writeIntLenenc(value.length);
+        byteBuf.writeBytes(value);
+    }
+    
     /**
      * Read fixed length string from byte buffers.
      * 
diff --git a/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/binary/execute/protocol/StringLenencBinaryProtocolValue.java b/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/binary/execute/protocol/StringLenencBinaryProtocolValue.java
index fdb23893c11..15119066d81 100644
--- a/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/binary/execute/protocol/StringLenencBinaryProtocolValue.java
+++ b/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/binary/execute/protocol/StringLenencBinaryProtocolValue.java
@@ -34,6 +34,10 @@ public final class StringLenencBinaryProtocolValue implements BinaryProtocolValu
     
     @Override
     public void write(final MySQLPacketPayload payload, final Object value) {
-        payload.writeStringLenenc(value.toString());
+        if (value instanceof byte[]) {
+            payload.writeBytesLenenc((byte[]) value);
+        } else {
+            payload.writeStringLenenc(value.toString());
+        }
     }
 }
diff --git a/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/text/TextResultSetRowPacket.java b/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/text/TextResultSetRowPacket.java
index cfe1df2aba7..11accce219a 100644
--- a/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/text/TextResultSetRowPacket.java
+++ b/sharding-proxy/src/main/java/io/shardingsphere/shardingproxy/transport/mysql/packet/command/query/text/TextResultSetRowPacket.java
@@ -56,7 +56,11 @@ public final class TextResultSetRowPacket implements MySQLPacket {
             if (null == each) {
                 payload.writeInt1(NULL);
             } else {
-                payload.writeStringLenenc(each.toString());
+                if (each instanceof byte[]) {
+                    payload.writeBytesLenenc((byte[]) each);
+                } else {
+                    payload.writeStringLenenc(each.toString());
+                }
             }
         }
     }
diff --git a/sharding-proxy/src/test/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnTypeTest.java b/sharding-proxy/src/test/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnTypeTest.java
index 3b19a49ccf2..a13dfad5e6f 100644
--- a/sharding-proxy/src/test/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnTypeTest.java
+++ b/sharding-proxy/src/test/java/io/shardingsphere/shardingproxy/transport/mysql/constant/ColumnTypeTest.java
@@ -44,9 +44,9 @@ public final class ColumnTypeTest {
         assertThat(ColumnType.valueOfJDBCType(Types.DATE), is(ColumnType.MYSQL_TYPE_DATE));
         assertThat(ColumnType.valueOfJDBCType(Types.TIME), is(ColumnType.MYSQL_TYPE_TIME));
         assertThat(ColumnType.valueOfJDBCType(Types.TIMESTAMP), is(ColumnType.MYSQL_TYPE_TIMESTAMP));
-        assertThat(ColumnType.valueOfJDBCType(Types.BINARY), is(ColumnType.MYSQL_TYPE_BLOB));
-        assertThat(ColumnType.valueOfJDBCType(Types.VARBINARY), is(ColumnType.MYSQL_TYPE_MEDIUM_BLOB));
-        assertThat(ColumnType.valueOfJDBCType(Types.LONGVARBINARY), is(ColumnType.MYSQL_TYPE_LONG_BLOB));
+        assertThat(ColumnType.valueOfJDBCType(Types.BINARY), is(ColumnType.MYSQL_TYPE_VARCHAR));
+        assertThat(ColumnType.valueOfJDBCType(Types.VARBINARY), is(ColumnType.MYSQL_TYPE_VARCHAR));
+        assertThat(ColumnType.valueOfJDBCType(Types.LONGVARBINARY), is(ColumnType.MYSQL_TYPE_VARCHAR));
         assertThat(ColumnType.valueOfJDBCType(Types.NULL), is(ColumnType.MYSQL_TYPE_NULL));
         assertThat(ColumnType.valueOfJDBCType(Types.BLOB), is(ColumnType.MYSQL_TYPE_BLOB));
     }