MariaDb extended type info changes

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

@@ -169,7 +169,7 @@ 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;
-    
+
     /**
      * Receive extended column type information from MariaDB to find out more specific details about column type.
      */
@@ -181,11 +181,22 @@ public final class Capability {
         TRANSACTIONS | SECURE_SALT | MULTI_STATEMENTS | MULTI_RESULTS | PS_MULTI_RESULTS |
         PLUGIN_AUTH | CONNECT_ATTRS | VAR_INT_SIZED_AUTH | SESSION_TRACK | DEPRECATE_EOF | ZSTD_COMPRESS;
 
+    private static final long ALL_SUPPORTED_MARIADB = 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 |
+        MARIADB_CLIENT_EXTENDED_TYPE_INFO;
+
     /**
      * The default capabilities for a MySQL connection. It contains all client supported capabilities.
      */
     public static final Capability DEFAULT = new Capability(ALL_SUPPORTED);
 
+    /**
+     * The default capabilities for a MariaDB connection. It contains all client supported capabilities.
+     */
+    public static final Capability DEFAULT_MARIADB = new Capability(ALL_SUPPORTED_MARIADB);
+
     private final long bitmap;
 
     /**
@@ -313,14 +324,14 @@ 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;
+        return (bitmap & MARIADB_CLIENT_EXTENDED_TYPE_INFO) != 0;
     }
 
     /**

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

@@ -68,7 +68,7 @@ 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.
      */
@@ -116,10 +116,10 @@ public boolean isEnum() {
     public boolean isSet() {
         return (definitions & SET) != 0;
     }
-    
+
     @Override
     public boolean isMariaDbJson() {
-    	return extendedTypeInfo.equals("json");
+        return (extendedTypeInfo == null ? false : extendedTypeInfo.equals("json"));
     }
 
     @Override
@@ -133,22 +133,23 @@ public boolean equals(Object o) {
 
         MySqlTypeMetadata that = (MySqlTypeMetadata) o;
 
-        return typeId == that.typeId && definitions == that.definitions && collationId == that.collationId && 
-        		Objects.equals(extendedTypeInfo, that.extendedTypeInfo);
+        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 + extendedTypeInfo.hashCode();
+        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 + 
-            '}';
+            ", collationId=" + collationId +
+            ", extendedTypeInfo='" + extendedTypeInfo +
+            "'}";
     }
 }

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

@@ -71,10 +71,10 @@ 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/message/server/DefinitionMetadataMessage.java

@@ -45,7 +45,7 @@ public final class DefinitionMetadataMessage implements ServerMessage {
 
     @Nullable
     private final String originColumn;
-    
+
     @Nullable
     private final String extendedTypeInfo;
 
@@ -60,8 +60,8 @@ 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, @Nullable String extendedTypeInfo, 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;
@@ -100,9 +100,10 @@ public int getDefinitions() {
     public short getDecimals() {
         return decimals;
     }
-
+
+    @Nullable
     public String getExtendedTypeInfo() {
-    	return extendedTypeInfo;
+        return extendedTypeInfo;
     }
 
     @Override
@@ -136,9 +137,9 @@ public int hashCode() {
     @Override
     public String toString() {
         return "DefinitionMetadataMessage{database='" + database + "', table='" + table + "' (origin:'" +
-            originTable + "'), column='" + column + "' (origin:'" + originColumn + "'), extendedTypeInfo=" +
-        	extendedTypeInfo + ", 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) {
@@ -179,10 +180,10 @@ 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);
+            extendTypeInfo = readVarIntSizedString(buf, charset);
         }
 
         // Skip constant 0x0c encoded by var integer
@@ -193,8 +194,8 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext
         short typeId = buf.readUnsignedByte();
         int definitions = buf.readUnsignedShortLE();
 
-        return new DefinitionMetadataMessage(database, table, originTable, column, originColumn, extendTypeInfo, 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,8 +16,14 @@
 
 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 reactor.core.publisher.Mono;
 
@@ -141,6 +147,22 @@ void returningGetRowUpdated() {
             .doOnNext(it -> assertThat(it).isEqualTo(2)));
     }
 
+    @Test
+    void returningExtendedTypeInfoJson() {
+        complete(conn -> changeCapability(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 -> assertThat(list.get(0)).isEqualTo(true))
+            );
+    }
+
     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 +193,12 @@ private static Mono<Void> assertWithoutCreatedAt(MySqlConnection conn, Mono<List
             .then();
     }
 
+    private static MySqlConnection changeCapability(MySqlConnection conn) {
+        ConnectionContext ctxt = ((MySqlSimpleConnection) conn).context();
+        ctxt.initHandshake(ctxt.getConnectionId(), ctxt.getServerVersion(), Capability.DEFAULT_MARIADB);
+        return conn;
+    }
+
     private static final class DataEntity {
 
         private final int id;
@@ -250,5 +278,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;
+        }
     }
 }