Add integration tests for MariaDB

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

@@ -0,0 +1,37 @@
+name: Integration Tests for MariaDB
+
+on:
+  pull_request:
+    branches: [ "trunk", "0.9.x" ]
+
+jobs:
+  mariadb-integration-tests-pr:
+    runs-on: ubuntu-20.04
+    strategy:
+      matrix:
+        mariadb-version: [ 10.6, 10.11 ]
+    name: Integration test with MariaDB ${{ matrix.mariadb-version }}
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Temurin 8
+        uses: actions/setup-java@v3
+        with:
+          distribution: temurin
+          java-version: 8
+          cache: maven
+      - name: Shutdown the Default MySQL
+        run: sudo service mysql stop
+      - name: Set up MariaDB ${{ matrix.mariadb-version }}
+        env:
+          MYSQL_DATABASE: r2dbc
+          MYSQL_ROOT_PASSWORD: r2dbc-password!@
+          MARIADB_VERSION: ${{ matrix.mariadb-version }}
+        run: docker-compose -f ${{ github.workspace }}/containers/mariadb-compose.yml up -d
+      - name: Integration test with MySQL ${{ matrix.mysql-version }}
+        run: |
+          ./mvnw -B verify -Dmaven.javadoc.skip=true \
+          -Dmaven.surefire.skip=true \
+          -Dtest.mysql.password=r2dbc-password!@ \
+          -Dtest.mysql.version=${{ matrix.mariadb-version }} \
+          -Dtest.db.type=mariadb \
+          -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=WARN

.github/workflows/ci-unit-tests.yml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        java-version: [ 8, 11, 17 , 21]
+        java-version: [ 8, 11, 17, 21 ]
     name: linux-java-${{ matrix.java-version }}
     steps:
       - uses: actions/checkout@v3

README.md

@@ -43,7 +43,8 @@ This project is currently being maintained by [@jchrys](https://github.com/jchry
 ![MySQL 8.0 status](https://img.shields.io/badge/MySQL%208.0-pass-blue)
 ![MySQL 8.1 status](https://img.shields.io/badge/MySQL%208.1-pass-blue)
 ![MySQL 8.2 status](https://img.shields.io/badge/MySQL%208.2-pass-blue)
-
+![MariaDB 10.6 status](https://img.shields.io/badge/MariaDB%2010.6-pass-blue)
+![MariaDB 10.11 status](https://img.shields.io/badge/MariaDB%2010.11-pass-blue)
 
 
 In fact, it supports lower versions, in the theory, such as 4.1, 4.0, etc.
@@ -546,7 +547,9 @@ If you want to raise an issue, please follow the recommendations below:
 - The MySQL server does not **actively** return time zone when query `DATETIME` or `TIMESTAMP`, this driver does not attempt time zone conversion. That means should always use `LocalDateTime` for SQL type `DATETIME` or `TIMESTAMP`. Execute `SHOW VARIABLES LIKE '%time_zone%'` to get more information.
 - Should not turn-on the `trace` log level unless debugging. Otherwise, the security information may be exposed through `ByteBuf` dump.
 - If `Statement` bound `returnGeneratedValues`, the `Result` of the `Statement` can be called both: `getRowsUpdated` to get affected rows, and `map` to get last inserted ID.
-- The MySQL may be not support search rows by a binary field, like `BIT`, `BLOB` and `JSON`, because those data fields maybe just an address of reference in MySQL server, or maybe need strict bit-aligned. (but `VARBINARY` is OK)
+- The MySQL may be not support well for searching rows by a binary field, like `BIT` and `JSON`
+  - `BIT`: cannot select 'BIT(64)' with value greater than 'Long.MAX_VALUE' (or equivalent in binary)
+  - `JSON`: different MySQL may have different serialization formats, e.g. MariaDB and MySQL
 
 ## License
 

containers/mariadb-compose.yml

@@ -0,0 +1,12 @@
+version: "3"
+
+services:
+  mariadb:
+    image: mariadb:${MARIADB_VERSION}
+    container_name: mariadb_${MARIADB_VERSION}
+    environment:
+      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD}
+      MYSQL_DATABASE: ${MYSQL_DATABASE}
+    command: mysqld --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
+    ports:
+      - "3306:3306"

src/main/java/io/asyncer/r2dbc/mysql/ParameterWriter.java

@@ -48,14 +48,24 @@ public abstract class ParameterWriter extends Writer {
 
     /**
      * Writes a value of {@code long} to current parameter. If current mode is string mode, it will write as a
-     * string like {@code write(String.valueOf(value))}. If it write as a numeric, nothing else can be written
-     * before or after this.
+     * string like {@code write(String.valueOf(value))}. If it writes as a numeric, nothing else can be
+     * written before or after this.
      *
      * @param value the value of {@code long}.
      * @throws IllegalStateException if parameters filled, or something was written before that numeric.
      */
     public abstract void writeLong(long value);
 
+    /**
+     * Writes a value as an unsigned {@code long} to current parameter. If current mode is string mode, it
+     * will write as a string like {@code write(String.valueOf(value))}. If it writes as a numeric, nothing
+     * else can be written before or after this.
+     *
+     * @param value the value as an unsigned {@code long}.
+     * @throws IllegalStateException if parameters filled, or something was written before that numeric.
+     */
+    public abstract void writeUnsignedLong(long value);
+
     /**
      * Writes a value of {@link BigInteger} to current parameter. If current mode is string mode, it will
      * write as a string like {@code write(value.toString())}. If it write as a numeric, nothing else can be

src/main/java/io/asyncer/r2dbc/mysql/codec/BitSetCodec.java

@@ -73,7 +73,9 @@ public MySqlParameter encode(Object value, CodecContext context) {
 
         MySqlType type;
 
-        if ((byte) bits == bits) {
+        if (bits < 0) {
+            type = MySqlType.BIGINT;
+        } else if ((byte) bits == bits) {
             type = MySqlType.TINYINT;
         } else if ((short) bits == bits) {
             type = MySqlType.SMALLINT;
@@ -135,17 +137,7 @@ public Mono<ByteBuf> publishBinary() {
 
         @Override
         public Mono<Void> publishText(ParameterWriter writer) {
-            return Mono.fromRunnable(() -> {
-                if (value == 0) {
-                    // Must filled by 0 for MySQL 5.5.x, because MySQL 5.5.x does not clear its buffer on type
-                    // BIT (i.e. unsafe allocate).
-                    // So if we do not fill the buffer, it will use last content which is an undefined
-                    // behavior. A classic bug, right?
-                    writer.writeBinary(false);
-                } else {
-                    writer.writeHex(value);
-                }
-            });
+            return Mono.fromRunnable(() -> writer.writeUnsignedLong(value));
         }
 
         @Override

src/main/java/io/asyncer/r2dbc/mysql/message/client/ParamWriter.java

@@ -79,6 +79,13 @@ public void writeLong(long value) {
         builder.append(value);
     }
 
+    @Override
+    public void writeUnsignedLong(long value) {
+        startAvailable(Mode.NUMERIC);
+
+        builder.append(Long.toUnsignedString(value));
+    }
+
     @Override
     public void writeBigInteger(BigInteger value) {
         requireNonNull(value, "value must not be null");

src/test/java/io/asyncer/r2dbc/mysql/IntegrationTestSupport.java

@@ -102,4 +102,38 @@ static MySqlConnectionConfiguration configuration(
 
         return builder.build();
     }
+
+    boolean envIsLessThanMySql56() {
+        String version = System.getProperty("test.mysql.version");
+
+        if (version == null || version.isEmpty()) {
+            return true;
+        }
+
+        ServerVersion ver = ServerVersion.parse(version);
+        String type = System.getProperty("test.db.type");
+
+        if ("mariadb".equalsIgnoreCase(type)) {
+            return false;
+        }
+
+        return ver.isLessThan(ServerVersion.create(5, 6, 0));
+    }
+
+    boolean envIsLessThanMySql57OrMariaDb102() {
+        String version = System.getProperty("test.mysql.version");
+
+        if (version == null || version.isEmpty()) {
+            return true;
+        }
+
+        ServerVersion ver = ServerVersion.parse(version);
+        String type = System.getProperty("test.db.type");
+
+        if ("mariadb".equalsIgnoreCase(type)) {
+            return ver.isLessThan(ServerVersion.create(10, 2, 0));
+        }
+
+        return ver.isLessThan(ServerVersion.create(5, 7, 0));
+    }
 }

src/test/java/io/asyncer/r2dbc/mysql/JacksonIntegrationTestSupport.java

@@ -22,7 +22,7 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
+import org.junit.jupiter.api.condition.DisabledIf;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -64,7 +64,7 @@ void tearDown() {
         JacksonCodecRegistrar.tearDown();
     }
 
-    @DisabledIfSystemProperty(named = "test.mysql.version", matches = "5\\.[56](\\.\\d+)?")
+    @DisabledIf("envIsLessThanMySql57OrMariaDb102")
     @Test
     void json() {
         create().flatMap(connection -> Mono.from(connection.createStatement(TDL).execute())

src/test/java/io/asyncer/r2dbc/mysql/QueryIntegrationTestSupport.java

@@ -22,12 +22,15 @@
 import io.r2dbc.spi.Result;
 import org.jetbrains.annotations.Nullable;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
+import org.junit.jupiter.api.condition.DisabledIf;
+import org.testcontainers.shaded.com.fasterxml.jackson.databind.JsonNode;
+import org.testcontainers.shaded.com.fasterxml.jackson.databind.ObjectMapper;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
 import reactor.util.function.Tuples;
 
+import java.io.IOException;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.math.BigDecimal;
@@ -206,11 +209,23 @@ void varbinary() {
             ByteBuffer.wrap(new byte[] { 1, 2, 3, 4, 5 }));
     }
 
+    @Test
+    void text() {
+        testType(byte[].class, true, "TEXT", null, new byte[0], new byte[] { 1, 2, 3, 4, 5 });
+        testType(ByteBuffer.class, true, "TEXT", null, ByteBuffer.allocate(0),
+            ByteBuffer.wrap(new byte[] { 1, 2, 3, 4, 5 }));
+    }
+
     @Test
     void bit() {
         testType(Boolean.class, true, "BIT(1)", null, false, true);
         testType(BitSet.class, true, "BIT(16)", null, BitSet.valueOf(new byte[] { (byte) 0xEF, (byte) 0xCD }),
-            BitSet.valueOf(new byte[0]), BitSet.valueOf(new byte[] { 0, 0 }));
+            BitSet.valueOf(new byte[0]), BitSet.valueOf(new byte[] { 0, 0 }), BitSet.valueOf(new byte[] { (byte) 0xCD }));
+        testType(BitSet.class, true, "BIT(64)", null, BitSet.valueOf(new long[0]),
+            BitSet.valueOf(new long[] { 0 }), BitSet.valueOf(new long[] { 0xEFCD }),
+            BitSet.valueOf(new long[] { Long.MAX_VALUE }));
+        testType(BitSet.class, false, "BIT(64)", BitSet.valueOf(new long[] { -1 }),
+            BitSet.valueOf(new long[] { Long.MIN_VALUE }));
         testType(byte[].class, false, "BIT(16)", null, new byte[] { (byte) 0xCD, (byte) 0xEF },
             new byte[] { 0, 0 });
         testType(ByteBuffer.class, false, "BIT(16)", null, ByteBuffer.wrap(new byte[] { 1, 2 }),
@@ -247,11 +262,13 @@ void set() {
             EnumSet.of(EnumData.ONE, EnumData.THREE));
     }
 
-    @DisabledIfSystemProperty(named = "test.mysql.version", matches = "5\\.[56](\\.\\d+)?")
+    @DisabledIf("envIsLessThanMySql57OrMariaDb102")
     @Test
     void json() {
         testType(String.class, false, "JSON", null, "{\"data\": 1}", "[\"data\", 1]", "1", "null",
             "\"R2DBC\"", "2.56");
+
+
     }
 
     @Test
@@ -274,7 +291,7 @@ void time() {
         testType(Duration.class, true, "TIME", null, minDuration, aDuration, maxDuration);
     }
 
-    @DisabledIfSystemProperty(named = "test.mysql.version", matches = "5\\.5(\\.\\d+)?")
+    @DisabledIf("envIsLessThanMySql56")
     @Test
     void time6() {
         LocalTime smallTime = LocalTime.of(0, 0, 0, 1000);
@@ -307,7 +324,7 @@ void timeDuration() {
                 .concatMap(pair -> testTimeDuration(connection, pair.getT1(), pair.getT2()))));
     }
 
-    @DisabledIfSystemProperty(named = "test.mysql.version", matches = "5\\.5(\\.\\d+)?")
+    @DisabledIf("envIsLessThanMySql56")
     @Test
     void timeDuration6() {
         long seconds = TimeUnit.HOURS.toSeconds(8) + TimeUnit.MINUTES.toSeconds(5) + 45;
@@ -337,7 +354,7 @@ void dateTime() {
         testType(LocalDateTime.class, true, "DATETIME", null, minDateTime, aDateTime, maxDateTime);
     }
 
-    @DisabledIfSystemProperty(named = "test.mysql.version", matches = "5\\.5(\\.\\d+)?")
+    @DisabledIf("envIsLessThanMySql56")
     @Test
     void dateTime6() {
         LocalDateTime smallDateTime = LocalDateTime.of(1000, 1, 1, 0, 0, 0, 1000);
@@ -357,7 +374,7 @@ void timestamp() {
         testType(LocalDateTime.class, true, "TIMESTAMP", minTimestamp, aTimestamp, maxTimestamp);
     }
 
-    @DisabledIfSystemProperty(named = "test.mysql.version", matches = "5\\.5(\\.\\d+)?")
+    @DisabledIf("envIsLessThanMySql56")
     @Test
     void timestamp6() {
         LocalDateTime minTimestamp = LocalDateTime.of(1970, 1, 3, 0, 0, 0, 1000);
@@ -553,41 +570,55 @@ void insertOnDuplicate() {
     }
 
     /**
-     * ref: https://github.com/asyncer-io/r2dbc-mysql/issues/91
+     * ref: <a href="https://github.com/asyncer-io/r2dbc-mysql/issues/91">Issue 91</a>
      */
-    @DisabledIfSystemProperty(named = "test.mysql.version", matches = "5\\.[56](\\.\\d+)?")
+    @DisabledIf("envIsLessThanMySql57OrMariaDb102")
     @Test
     void testUnionQueryWithJsonColumnDecodedAsString() {
         complete(connection ->
-                         Flux.from(connection.createStatement(
-                                                     "CREATE TEMPORARY TABLE test1 (id INT PRIMARY KEY AUTO_INCREMENT, value JSON)")
-                                             .execute())
-                             .flatMap(IntegrationTestSupport::extractRowsUpdated)
-                             .thenMany(connection.createStatement("INSERT INTO test1 VALUES(DEFAULT, ?)")
-                                                 .bind(0, "{\"id\":1,\"name\":\"iron man\"}")
-                                                 .execute())
-                             .flatMap(IntegrationTestSupport::extractRowsUpdated)
-                             .doOnNext(it -> assertThat(it).isEqualTo(1))
-                             .thenMany(connection.createStatement(
-                                                         "CREATE TEMPORARY TABLE test2 (id INT PRIMARY KEY AUTO_INCREMENT, value JSON)")
-                                                 .execute())
-                             .flatMap(IntegrationTestSupport::extractRowsUpdated)
-                             .thenMany(connection.createStatement("INSERT INTO test2 VALUES(DEFAULT, ?)")
-                                                 .bind(0,
-                                                       "[{\"id\":2,\"name\":\"bat man\"},{\"id\":3,\"name\":\"super man\"}]")
-                                                 .execute())
-                             .flatMap(IntegrationTestSupport::extractRowsUpdated)
-                             .doOnNext(it -> assertThat(it).isEqualTo(1))
-                             .thenMany(
-                                     connection.createStatement("SELECT value FROM test1 UNION SELECT value FROM test2")
-                                               .execute())
-                             .flatMap(r -> r.map((row, metadata) -> row.get(0, String.class))
-                                            .collectList()
-                                            .doOnNext(it -> assertThat(it).isEqualTo(
-                                                    Arrays.asList("{\"id\": 1, \"name\": \"iron man\"}",
-                                                                  "[{\"id\": 2, \"name\": \"bat man\"}, {\"id\": 3, \"name\": \"super man\"}]"))))
+            Flux.from(connection.createStatement(
+                        "CREATE TEMPORARY TABLE test1 (id INT PRIMARY KEY AUTO_INCREMENT, value JSON)")
+                    .execute())
+                .flatMap(IntegrationTestSupport::extractRowsUpdated)
+                .thenMany(connection.createStatement("INSERT INTO test1 VALUES(DEFAULT, ?)")
+                    .bind(0, "{\"id\":1,\"name\":\"iron man\"}")
+                    .execute())
+                .flatMap(IntegrationTestSupport::extractRowsUpdated)
+                .doOnNext(it -> assertThat(it).isEqualTo(1))
+                .thenMany(connection.createStatement(
+                        "CREATE TEMPORARY TABLE test2 (id INT PRIMARY KEY AUTO_INCREMENT, value JSON)")
+                    .execute())
+                .flatMap(IntegrationTestSupport::extractRowsUpdated)
+                .thenMany(connection.createStatement("INSERT INTO test2 VALUES(DEFAULT, ?)")
+                    .bind(0,
+                        "[{\"id\":2,\"name\":\"bat man\"},{\"id\":3,\"name\":\"super man\"}]")
+                    .execute())
+                .flatMap(IntegrationTestSupport::extractRowsUpdated)
+                .doOnNext(it -> assertThat(it).isEqualTo(1))
+                .thenMany(
+                    connection.createStatement("SELECT value FROM test1 UNION SELECT value FROM test2")
+                        .execute())
+                .flatMap(r -> r.map((row, metadata) -> row.get(0, String.class))
+                    .map(QueryIntegrationTestSupport::parseJson)
+                    .collectList()
+                    .doOnNext(it -> assertThat(it).isEqualTo(
+                        Arrays.asList(
+                            parseJson("{\"id\": 1, \"name\": \"iron man\"}"),
+                            parseJson(
+                                "[{\"id\": 2, \"name\": \"bat man\"}, {\"id\": 3, \"name\": \"super man\"}]"
+                            )
+                        )
+                    )))
         );
+    }
 
+    private static JsonNode parseJson(String json) {
+        ObjectMapper mapper = new ObjectMapper();
+        try {
+            return mapper.readTree(json);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     private static Flux<Integer> extractFirstInteger(Result result) {