Added support for MariaDb extend type info

r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/Capability.java

@@ -169,7 +169,11 @@ public final class Capability {
 //    private static final long MARIADB_CLIENT_PROGRESS = 1L << 32;
 //    private static final long MARIADB_CLIENT_COM_MULTI = 1L << 33;
 //    private static final long MARIADB_CLIENT_STMT_BULK_OPERATIONS = 1L << 34;
-//    private static final long MARIADB_CLIENT_EXTENDED_TYPE_INFO = 1L << 35;
+
+    /**
+     * Receive extended column type information from MariaDB to find out more specific details about column type.
+     */
+    private static final long MARIADB_CLIENT_EXTENDED_TYPE_INFO = 1L << 35;
 //    private static final long MARIADB_CLIENT_CACHE_METADATA = 1L << 36;
 
     private static final long ALL_SUPPORTED = CLIENT_MYSQL | FOUND_ROWS | LONG_FLAG | CONNECT_WITH_DB |
@@ -309,6 +313,15 @@ public boolean isZlibCompression() {
     public boolean isZstdCompression() {
         return (bitmap & ZSTD_COMPRESS) != 0;
     }
+
+    /**
+     * Checks if MariaDB extended type info enabled.
+     * 
+     * @return if MariaDB extended type info enabled.
+     */
+    public boolean isExtendedTypeInfo() {
+    	return (bitmap & MARIADB_CLIENT_EXTENDED_TYPE_INFO) != 0;
+    }
 
     /**
      * Extends MariaDB capabilities.

r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/MySqlColumnDescriptor.java

@@ -23,6 +23,8 @@
 import io.asyncer.r2dbc.mysql.constant.MySqlType;
 import io.asyncer.r2dbc.mysql.message.server.DefinitionMetadataMessage;
 import io.r2dbc.spi.Nullability;
+
+import org.jetbrains.annotations.Nullable;
 import org.jetbrains.annotations.VisibleForTesting;
 
 import static io.asyncer.r2dbc.mysql.internal.util.AssertUtils.require;
@@ -53,13 +55,13 @@ final class MySqlColumnDescriptor implements MySqlColumnMetadata {
 
     @VisibleForTesting
     MySqlColumnDescriptor(int index, short typeId, String name, int definitions,
-        long size, int decimals, int collationId) {
+        long size, int decimals, int collationId, @Nullable String extendedTypeInfo) {
         require(index >= 0, "index must not be a negative integer");
         require(size >= 0, "size must not be a negative integer");
         require(decimals >= 0, "decimals must not be a negative integer");
         requireNonNull(name, "name must not be null");
 
-        MySqlTypeMetadata typeMetadata = new MySqlTypeMetadata(typeId, definitions, collationId);
+        MySqlTypeMetadata typeMetadata = new MySqlTypeMetadata(typeId, definitions, collationId, extendedTypeInfo);
 
         this.index = index;
         this.typeMetadata = typeMetadata;
@@ -74,7 +76,7 @@ final class MySqlColumnDescriptor implements MySqlColumnMetadata {
     static MySqlColumnDescriptor create(int index, DefinitionMetadataMessage message) {
         int definitions = message.getDefinitions();
         return new MySqlColumnDescriptor(index, message.getTypeId(), message.getColumn(), definitions,
-            message.getSize(), message.getDecimals(), message.getCollationId());
+            message.getSize(), message.getDecimals(), message.getCollationId(), message.getExtendedTypeInfo());
     }
 
     int getIndex() {

r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/MySqlTypeMetadata.java

@@ -16,6 +16,10 @@
 
 package io.asyncer.r2dbc.mysql;
 
+import java.util.Objects;
+
+import org.jetbrains.annotations.Nullable;
+
 import io.asyncer.r2dbc.mysql.api.MySqlNativeTypeMetadata;
 import io.asyncer.r2dbc.mysql.collation.CharCollation;
 
@@ -64,11 +68,18 @@ final class MySqlTypeMetadata implements MySqlNativeTypeMetadata {
      * collationId > 0 when protocol version == 4.1, 0 otherwise.
      */
     private final int collationId;
+
+    /**
+     * The MariaDB extended type info field that provides more specific details about column type.
+     */
+    @Nullable
+    private final String extendedTypeInfo;
 
-    MySqlTypeMetadata(int typeId, int definitions, int collationId) {
+    MySqlTypeMetadata(int typeId, int definitions, int collationId, @Nullable String extendedTypeInfo) {
         this.typeId = typeId;
         this.definitions = (short) (definitions & ALL_USED);
         this.collationId = collationId;
+        this.extendedTypeInfo = extendedTypeInfo;
     }
 
     @Override
@@ -105,6 +116,11 @@ public boolean isEnum() {
     public boolean isSet() {
         return (definitions & SET) != 0;
     }
+
+    @Override
+    public boolean isMariaDbJson() {
+    	return extendedTypeInfo.equals("json");
+    }
 
     @Override
     public boolean equals(Object o) {
@@ -117,20 +133,22 @@ public boolean equals(Object o) {
 
         MySqlTypeMetadata that = (MySqlTypeMetadata) o;
 
-        return typeId == that.typeId && definitions == that.definitions && collationId == that.collationId;
+        return typeId == that.typeId && definitions == that.definitions && collationId == that.collationId && 
+        		Objects.equals(extendedTypeInfo, that.extendedTypeInfo);
     }
 
     @Override
     public int hashCode() {
         int result = 31 * typeId + (int) definitions;
-        return 31 * result + collationId;
+        return 31 * result + collationId + extendedTypeInfo.hashCode();
     }
 
     @Override
     public String toString() {
         return "MySqlTypeMetadata{typeId=" + typeId +
             ", definitions=0x" + Integer.toHexString(definitions) +
-            ", collationId=" + collationId +
+            ", collationId=" + collationId + 
+            ", extendedTypeInfo=" + extendedTypeInfo + 
             '}';
     }
 }

r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/api/MySqlNativeTypeMetadata.java

@@ -71,4 +71,11 @@ public interface MySqlNativeTypeMetadata {
      * @return if value is a set
      */
     boolean isSet();
+
+    /**
+     * Checks if value is JSON for MariaDb.
+     * 
+     * @return if value is a JSON for MariaDb
+     */
+    boolean isMariaDbJson();
 }

r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/constant/MySqlType.java

@@ -712,7 +712,7 @@ public static MySqlType of(MySqlNativeTypeMetadata metadata) {
             case ID_VARCHAR:
             case ID_VAR_STRING:
             case ID_STRING:
-                return metadata.isBinary() ? VARBINARY : VARCHAR;
+                return metadata.isBinary() ? VARBINARY : (metadata.isMariaDbJson() ? JSON : VARCHAR);
             case ID_BIT:
                 return BIT;
             case ID_JSON:

r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/message/server/DefinitionMetadataMessage.java

@@ -45,6 +45,9 @@ public final class DefinitionMetadataMessage implements ServerMessage {
 
     @Nullable
     private final String originColumn;
+
+    @Nullable
+    private final String extendedTypeInfo;
 
     private final int collationId;
 
@@ -57,7 +60,7 @@ public final class DefinitionMetadataMessage implements ServerMessage {
     private final short decimals;
 
     private DefinitionMetadataMessage(@Nullable String database, String table, @Nullable String originTable,
-        String column, @Nullable String originColumn, int collationId, long size, short typeId,
+        String column, @Nullable String originColumn, @Nullable String extendedTypeInfo, int collationId, long size, short typeId,
         int definitions, short decimals) {
         require(size >= 0, "size must not be a negative integer");
 
@@ -66,6 +69,7 @@ private DefinitionMetadataMessage(@Nullable String database, String table, @Null
         this.originTable = originTable;
         this.column = requireNonNull(column, "column must not be null");
         this.originColumn = originColumn;
+        this.extendedTypeInfo = extendedTypeInfo;
         this.collationId = collationId;
         this.size = size;
         this.typeId = typeId;
@@ -96,6 +100,10 @@ public int getDefinitions() {
     public short getDecimals() {
         return decimals;
     }
+
+    public String getExtendedTypeInfo() {
+    	return extendedTypeInfo;
+    }
 
     @Override
     public boolean equals(Object o) {
@@ -115,21 +123,22 @@ public boolean equals(Object o) {
             table.equals(that.table) &&
             Objects.equals(originTable, that.originTable) &&
             column.equals(that.column) &&
-            Objects.equals(originColumn, that.originColumn);
+            Objects.equals(originColumn, that.originColumn) &&
+            Objects.equals(extendedTypeInfo, that.extendedTypeInfo);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(database, table, originTable, column, originColumn, collationId, size, typeId,
-            definitions, decimals);
+            definitions, decimals, extendedTypeInfo);
     }
 
     @Override
     public String toString() {
         return "DefinitionMetadataMessage{database='" + database + "', table='" + table + "' (origin:'" +
-            originTable + "'), column='" + column + "' (origin:'" + originColumn + "'), collationId=" +
-            collationId + ", size=" + size + ", type=" + typeId + ", definitions=" + definitions +
-            ", decimals=" + decimals + '}';
+            originTable + "'), column='" + column + "' (origin:'" + originColumn + "'), extendedTypeInfo=" +
+        	extendedTypeInfo + ", collationId=" +collationId + ", size=" + size + ", type=" + typeId + 
+        	", definitions=" + definitions + ", decimals=" + decimals + '}';
     }
 
     static DefinitionMetadataMessage decode(ByteBuf buf, ConnectionContext context) {
@@ -156,7 +165,7 @@ private static DefinitionMetadataMessage decode320(ByteBuf buf, ConnectionContex
         int definitions = buf.readUnsignedShortLE();
         short decimals = buf.readUnsignedByte();
 
-        return new DefinitionMetadataMessage(null, table, null, column, null, 0, size, typeId,
+        return new DefinitionMetadataMessage(null, table, null, column, null, null, 0, size, typeId,
             definitions, decimals);
     }
 
@@ -170,6 +179,11 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
         String originTable = readVarIntSizedString(buf, charset);
         String column = readVarIntSizedString(buf, charset);
         String originColumn = readVarIntSizedString(buf, charset);
+
+        String extendTypeInfo = null;
+        if (context.getCapability().isMariaDb() &&  context.getCapability().isExtendedTypeInfo()) {
+        	extendTypeInfo = readVarIntSizedString(buf, charset);
+        }
 
         // Skip constant 0x0c encoded by var integer
         VarIntUtils.readVarInt(buf);
@@ -179,7 +193,7 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
         short typeId = buf.readUnsignedByte();
         int definitions = buf.readUnsignedShortLE();
 
-        return new DefinitionMetadataMessage(database, table, originTable, column, originColumn, collationId,
+        return new DefinitionMetadataMessage(database, table, originTable, column, originColumn, extendTypeInfo, collationId,
             size, typeId, definitions, buf.readUnsignedByte());
     }
 

r2dbc-mysql/src/test/java/io/asyncer/r2dbc/mysql/MySqlRowDescriptorTest.java

@@ -56,7 +56,7 @@ private static MySqlRowDescriptor create(final String... names) {
         MySqlColumnDescriptor[] metadata = new MySqlColumnDescriptor[names.length];
         for (int i = 0; i < names.length; ++i) {
             metadata[i] =
-                    new MySqlColumnDescriptor(i, (short) 0, names[i], 0, 0, 0, 1);
+                    new MySqlColumnDescriptor(i, (short) 0, names[i], 0, 0, 0, 1, null);
         }
         return new MySqlRowDescriptor(metadata);
     }

r2dbc-mysql/src/test/java/io/asyncer/r2dbc/mysql/MySqlTypeMetadataTest.java

@@ -17,6 +17,8 @@
 package io.asyncer.r2dbc.mysql;
 
 import io.asyncer.r2dbc.mysql.collation.CharCollation;
+import io.asyncer.r2dbc.mysql.constant.MySqlType;
+
 import org.junit.jupiter.api.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -28,7 +30,7 @@ class MySqlTypeMetadataTest {
 
     @Test
     void allSet() {
-        MySqlTypeMetadata metadata = new MySqlTypeMetadata(0, -1, 0);
+        MySqlTypeMetadata metadata = new MySqlTypeMetadata(0, -1, 0, null);
 
         assertThat(metadata.isBinary()).isTrue();
         assertThat(metadata.isSet()).isTrue();
@@ -39,7 +41,7 @@ void allSet() {
 
     @Test
     void noSet() {
-        MySqlTypeMetadata metadata = new MySqlTypeMetadata(0, 0, 0);
+        MySqlTypeMetadata metadata = new MySqlTypeMetadata(0, 0, 0, null);
 
         assertThat(metadata.isBinary()).isFalse();
         assertThat(metadata.isSet()).isFalse();
@@ -50,11 +52,24 @@ void noSet() {
 
     @Test
     void isBinaryUsesCollationId() {
-        MySqlTypeMetadata metadata = new MySqlTypeMetadata(0, -1, CharCollation.BINARY_ID);
+        MySqlTypeMetadata metadata = new MySqlTypeMetadata(0, -1, CharCollation.BINARY_ID, null);
 
         assertThat(metadata.isBinary()).isTrue();
 
-        metadata = new MySqlTypeMetadata(0, -1, 33);
+        metadata = new MySqlTypeMetadata(0, -1, 33, null);
         assertThat(metadata.isBinary()).isFalse();
     }
+
+    @Test
+    void mariaDbJsonReturnsCorrectMySqlType() {
+    	MySqlTypeMetadata metadata = new MySqlTypeMetadata(254, 0, 0, "json");
+
+    	assertThat(metadata.isMariaDbJson()).isTrue();
+    	assertThat(MySqlType.of(metadata)).isEqualTo(MySqlType.JSON);
+
+    	metadata = new MySqlTypeMetadata(254, 0 ,0 , null);
+
+    	assertThat(metadata.isMariaDbJson()).isFalse();
+    	assertThat(MySqlType.of(metadata)).isEqualTo(MySqlType.VARCHAR);
+    }
 }