Added support for MariaDb extend type info

.github/workflows/ci-mariadb-intergration-tests.yml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        mariadb-version: [ 10.0, 10.1, 10.2.15, 10.2, 10.3.7, 10.3, 10.5.1, 10.5, 10.6, 10.11]
+        mariadb-version: [ 10.0, 10.1, 10.2.15, 10.2, 10.3.7, 10.3, 10.5.1, 10.5, 10.6, 10.11 ]
     name: Integration test with MariaDB ${{ matrix.mariadb-version }}
     steps:
       - uses: actions/checkout@v3

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

@@ -169,13 +169,18 @@ 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 |
         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_TYPE_INFO;
 
     /**
      * The default capabilities for a MySQL connection. It contains all client supported capabilities.
@@ -310,6 +315,15 @@ 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;
 
@@ -65,10 +69,17 @@ final class MySqlTypeMetadata implements MySqlNativeTypeMetadata {
      */
     private final int collationId;
 
-    MySqlTypeMetadata(int typeId, int definitions, 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, @Nullable String extendedTypeInfo) {
         this.typeId = typeId;
         this.definitions = (short) (definitions & ALL_USED);
         this.collationId = collationId;
+        this.extendedTypeInfo = extendedTypeInfo;
     }
 
     @Override
@@ -106,6 +117,11 @@ public boolean isSet() {
         return (definitions & SET) != 0;
     }
 
+    @Override
+    public boolean isMariaDbJson() {
+        return (extendedTypeInfo == null ? false : extendedTypeInfo.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(extendedTypeInfo, that.extendedTypeInfo);
     }
 
     @Override
     public int hashCode() {
         int result = 31 * typeId + (int) definitions;
-        return 31 * result + collationId;
+        result = 31 * result + collationId;
+        return 31 * result + (extendedTypeInfo == null ? 0 : extendedTypeInfo.hashCode());
     }
 
     @Override
     public String toString() {
         return "MySqlTypeMetadata{typeId=" + typeId +
             ", definitions=0x" + Integer.toHexString(definitions) +
             ", 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

@@ -46,6 +46,9 @@ public final class DefinitionMetadataMessage implements ServerMessage {
     @Nullable
     private final String originColumn;
 
+    @Nullable
+    private final String extendedTypeInfo;
+
     private final int collationId;
 
     private final long size;
@@ -57,15 +60,16 @@ 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,
-        int definitions, short decimals) {
+        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");
 
         this.database = database;
         this.table = requireNonNull(table, "table must not be 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;
@@ -97,6 +101,11 @@ public short getDecimals() {
         return decimals;
     }
 
+    @Nullable
+    public String getExtendedTypeInfo() {
+        return extendedTypeInfo;
+    }
+
     @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(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 +166,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);
     }
 
@@ -171,6 +181,12 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
         String column = readVarIntSizedString(buf, charset);
         String originColumn = readVarIntSizedString(buf, charset);
 
+        String extendTypeInfo = null;
+        if (context.getCapability().isMariaDb() &&  context.getCapability().isExtendedTypeInfo()) {
+            buf.readUnsignedByte();
+            extendTypeInfo = readVarIntSizedString(buf, charset);
+        }
+
         // Skip constant 0x0c encoded by var integer
         VarIntUtils.readVarInt(buf);
 
@@ -179,8 +195,8 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
         short typeId = buf.readUnsignedByte();
         int definitions = buf.readUnsignedShortLE();
 
-        return new DefinitionMetadataMessage(database, table, originTable, column, originColumn, collationId,
-            size, typeId, definitions, buf.readUnsignedByte());
+        return new DefinitionMetadataMessage(database, table, originTable, column, originColumn,
+            extendTypeInfo, collationId, size, typeId, definitions, buf.readUnsignedByte());
     }
 
     private static String readVarIntSizedString(ByteBuf buf, Charset charset) {

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

@@ -16,9 +16,17 @@
 
 package io.asyncer.r2dbc.mysql;
 
+import io.asyncer.r2dbc.mysql.api.MySqlColumnMetadata;
 import io.asyncer.r2dbc.mysql.api.MySqlConnection;
+import io.asyncer.r2dbc.mysql.api.MySqlRow;
+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 +149,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::readExtendedTypeInfoResult))
+            .collectList()
+            .doOnNext(list -> assertIfExtendedTypeInfoEnabled(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 +196,15 @@ private static Mono<Void> assertWithoutCreatedAt(MySqlConnection conn, Mono<List
             .then();
     }
 
+    private static void assertIfExtendedTypeInfoEnabled(MySqlConnection conn, List<Boolean> list) {
+        boolean enabled = ((MySqlSimpleConnection)conn).context().getCapability().isExtendedTypeInfo();
+        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 +284,11 @@ static DataEntity withoutCreatedAt(Readable readable) {
 
             return new DataEntity(id, value, ZonedDateTime.ofInstant(Instant.EPOCH, ZoneOffset.UTC));
         }
+
+        static Boolean readExtendedTypeInfoResult(Row row, RowMetadata rowMetadata) {
+            Boolean extendedTypeInfoResult = ((MySqlRowMetadata)rowMetadata)
+                .getColumnMetadata("value").getNativeTypeMetadata().isMariaDbJson();
+            return extendedTypeInfoResult;
+        }
     }
 }

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