Added MariaDB extended metadata

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

@@ -169,13 +169,19 @@ 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_METADATA = 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 |
         NO_SCHEMA | COMPRESS | LOCAL_FILES | IGNORE_SPACE | PROTOCOL_41 | INTERACTIVE | SSL |
         TRANSACTIONS | SECURE_SALT | MULTI_STATEMENTS | MULTI_RESULTS | PS_MULTI_RESULTS |
-        PLUGIN_AUTH | CONNECT_ATTRS | VAR_INT_SIZED_AUTH | SESSION_TRACK | DEPRECATE_EOF | ZSTD_COMPRESS;
+        PLUGIN_AUTH | CONNECT_ATTRS | VAR_INT_SIZED_AUTH | SESSION_TRACK | DEPRECATE_EOF | ZSTD_COMPRESS |
+        MARIADB_CLIENT_EXTENDED_METADATA;
 
     /**
      * The default capabilities for a MySQL connection. It contains all client supported capabilities.
@@ -310,6 +316,15 @@ public boolean isZstdCompression() {
         return (bitmap & ZSTD_COMPRESS) != 0;
     }
 
+    /**
+     * Checks if MariaDB extended metadata enabled.
+     *
+     * @return if MariaDB extended metadata enabled.
+     */
+    public boolean isExtendedMetadata() {
+        return (bitmap & MARIADB_CLIENT_EXTENDED_METADATA) != 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 extendedMetadata) {
         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, extendedMetadata);
 
         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.getExtendedMetadata());
     }
 
     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;
 
@@ -65,10 +69,17 @@ final class MySqlTypeMetadata implements MySqlNativeTypeMetadata {
      */
     private final int collationId;
 
-    MySqlTypeMetadata(int typeId, int definitions, int collationId) {
+    /**
+     * The MariaDB extended metadata field that provides more specific details about column type.
+     */
+    @Nullable
+    private final String extendedMetadata;
+
+    MySqlTypeMetadata(int typeId, int definitions, int collationId, @Nullable String extendedMetadata) {
         this.typeId = typeId;
         this.definitions = (short) (definitions & ALL_USED);
         this.collationId = collationId;
+        this.extendedMetadata = extendedMetadata;
     }
 
     @Override
@@ -106,6 +117,11 @@ public boolean isSet() {
         return (definitions & SET) != 0;
     }
 
+    @Override
+    public boolean isMariaDbJson() {
+        return (extendedMetadata == null ? false : extendedMetadata.equals("json"));
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -117,20 +133,23 @@ 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(extendedMetadata, that.extendedMetadata);
     }
 
     @Override
     public int hashCode() {
         int result = 31 * typeId + (int) definitions;
-        return 31 * result + collationId;
+        result = 31 * result + collationId;
+        return 31 * result + (extendedMetadata == null ? 0 : extendedMetadata.hashCode());
     }
 
     @Override
     public String toString() {
         return "MySqlTypeMetadata{typeId=" + typeId +
             ", definitions=0x" + Integer.toHexString(definitions) +
             ", collationId=" + collationId +
-            '}';
+            ", extendedMetadata='" + extendedMetadata +
+            "'}";
     }
 }

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

@@ -56,9 +56,12 @@ public final class DefinitionMetadataMessage implements ServerMessage {
 
     private final short decimals;
 
+    @Nullable
+    private final String extendedMetadata;
+
     private DefinitionMetadataMessage(@Nullable String database, String table, @Nullable String originTable,
         String column, @Nullable String originColumn, int collationId, long size, short typeId,
-        int definitions, short decimals) {
+        int definitions, short decimals, @Nullable String extendedMetadata) {
         require(size >= 0, "size must not be a negative integer");
 
         this.database = database;
@@ -71,6 +74,7 @@ private DefinitionMetadataMessage(@Nullable String database, String table, @Null
         this.typeId = typeId;
         this.definitions = definitions;
         this.decimals = decimals;
+        this.extendedMetadata = extendedMetadata;
     }
 
     public String getColumn() {
@@ -97,6 +101,11 @@ public short getDecimals() {
         return decimals;
     }
 
+    @Nullable
+    public String getExtendedMetadata() {
+        return extendedMetadata;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -115,21 +124,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(extendedMetadata, that.extendedMetadata);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(database, table, originTable, column, originColumn, collationId, size, typeId,
-            definitions, decimals);
+            definitions, decimals, extendedMetadata);
     }
 
     @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 + '}';
+            ", decimals=" + decimals + ", extendedMetadata='" + extendedMetadata + "'}";
     }
 
     static DefinitionMetadataMessage decode(ByteBuf buf, ConnectionContext context) {
@@ -157,7 +167,7 @@ private static DefinitionMetadataMessage decode320(ByteBuf buf, ConnectionContex
         short decimals = buf.readUnsignedByte();
 
         return new DefinitionMetadataMessage(null, table, null, column, null, 0, size, typeId,
-            definitions, decimals);
+            definitions, decimals, null);
     }
 
     private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext context) {
@@ -171,6 +181,12 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
         String column = readVarIntSizedString(buf, charset);
         String originColumn = readVarIntSizedString(buf, charset);
 
+        String extendMetadata = null;
+        if (context.getCapability().isExtendedMetadata() && buf.readUnsignedByte() != 0) {
+            buf.readUnsignedByte();
+            extendMetadata = readVarIntSizedString(buf, charset);
+        }
+
         // Skip constant 0x0c encoded by var integer
         VarIntUtils.readVarInt(buf);
 
@@ -180,7 +196,7 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
         int definitions = buf.readUnsignedShortLE();
 
         return new DefinitionMetadataMessage(database, table, originTable, column, originColumn, collationId,
-            size, typeId, definitions, buf.readUnsignedByte());
+            size, typeId, definitions, buf.readUnsignedByte(), extendMetadata);
     }
 
     private static String readVarIntSizedString(ByteBuf buf, Charset charset) {

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

@@ -17,8 +17,14 @@
 package io.asyncer.r2dbc.mysql;
 
 import io.asyncer.r2dbc.mysql.api.MySqlConnection;
+import io.asyncer.r2dbc.mysql.api.MySqlRowMetadata;
 import io.r2dbc.spi.Readable;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIf;
+
 import reactor.core.publisher.Mono;
 
 import java.time.Instant;
@@ -141,6 +147,23 @@ void returningGetRowUpdated() {
             .doOnNext(it -> assertThat(it).isEqualTo(2)));
     }
 
+    @Test
+    @EnabledIf("envIsMariaDb10_5_1")
+    void returningExtendedTypeInfoJson() {
+        complete(conn -> conn.createStatement("CREATE TEMPORARY TABLE test(" +
+            "id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, value JSON NOT NULL)")
+            .execute()
+            .flatMap(IntegrationTestSupport::extractRowsUpdated)
+            .thenMany(conn.createStatement("INSERT INTO test(value) VALUES (?)")
+                .bind(0, "{\"abc\": 123}")
+                .returnGeneratedValues()
+                .execute())
+            .flatMap(result -> result.map(DataEntity::readExtendedMetadataResult))
+            .collectList()
+            .doOnNext(list -> assertIfExtendedMetadataEnabled(conn, list))
+            );
+    }
+
     private static Mono<Void> assertWithSelectAll(MySqlConnection conn, Mono<List<DataEntity>> returning) {
         return returning.zipWhen(list -> conn.createStatement("SELECT * FROM test WHERE id IN (?,?,?,?,?)")
                 .bind(0, list.get(0).getId())
@@ -171,6 +194,15 @@ private static Mono<Void> assertWithoutCreatedAt(MySqlConnection conn, Mono<List
             .then();
     }
 
+    private static void assertIfExtendedMetadataEnabled(MySqlConnection conn, List<Boolean> list) {
+        boolean enabled = ((MySqlSimpleConnection)conn).context().getCapability().isExtendedMetadata();
+        if (enabled) {
+            assertThat(list.get(0)).isEqualTo(true);
+        } else {
+            assertThat(list.get(0)).isEqualTo(false);
+        }
+    }
+
     private static final class DataEntity {
 
         private final int id;
@@ -250,5 +282,11 @@ static DataEntity withoutCreatedAt(Readable readable) {
 
             return new DataEntity(id, value, ZonedDateTime.ofInstant(Instant.EPOCH, ZoneOffset.UTC));
         }
+
+        static Boolean readExtendedMetadataResult(Row row, RowMetadata rowMetadata) {
+            Boolean extendedMetadataResult = ((MySqlRowMetadata)rowMetadata)
+                .getColumnMetadata("value").getNativeTypeMetadata().isMariaDbJson();
+            return extendedMetadataResult;
+        }
     }
 }

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);
+    }
 }