Add LOCAL INFILE support

README.md

@@ -141,6 +141,7 @@ ConnectionFactoryOptions options = ConnectionFactoryOptions.builder()
     .option(Option.valueOf("sslContextBuilderCustomizer"), "com.example.demo.MyCustomizer") // optional, default is no-op customizer
     .option(Option.valueOf("zeroDate"), "use_null") // optional, default "use_null"
     .option(Option.valueOf("useServerPrepareStatement"), true) // optional, default false
+    .option(Option.valueOf("allowLoadLocalInfileInPath"), "/opt") // optional, default null, null means LOCAL INFILE not be allowed
     .option(Option.valueOf("tcpKeepAlive"), true) // optional, default false
     .option(Option.valueOf("tcpNoDelay"), true) // optional, default false
     .option(Option.valueOf("autodetectExtensions"), false) // optional, default false
@@ -189,6 +190,7 @@ MySqlConnectionConfiguration configuration = MySqlConnectionConfiguration.builde
     .sslContextBuilderCustomizer(MyCustomizer.INSTANCE) // optional, default is no-op customizer
     .zeroDateOption(ZeroDateOption.USE_NULL) // optional, default ZeroDateOption.USE_NULL
     .useServerPrepareStatement() // Use server-preparing statements, default use client-preparing statements
+    .allowLoadLocalInfileInPath("/opt") // optional, default null, null means LOCAL INFILE not be allowed
     .tcpKeepAlive(true) // optional, controls TCP Keep Alive, default is false
     .tcpNoDelay(true) // optional, controls TCP No Delay, default is false
     .autodetectExtensions(false) // optional, controls extension auto-detect, default is true
@@ -242,6 +244,7 @@ Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
 | zeroDateOption | Any value of `ZeroDateOption` | Optional, default `USE_NULL` | The option indicates "zero date" handling, see following notice |
 | autodetectExtensions | `true` or `false` | Optional, default is `true` | Controls auto-detect `Extension`s |
 | useServerPrepareStatement | `true`, `false` or `Predicate<String>` | Optional, default is `false` | See following notice |
+| allowLoadLocalInfileInPath | A path | Optional, default is `null` | The path that allows `LOAD DATA LOCAL INFILE` to load file data |
 | passwordPublisher | A `Publisher<String>` | Optional, default is `null` | The password publisher, see following notice |
 
 - `SslMode` Considers security level and verification for SSL, make sure the database server supports SSL before you want change SSL mode to `REQUIRED` or higher. **The Unix Domain Socket only offers "DISABLED" available**
@@ -584,6 +587,7 @@ If you want to raise an issue, please follow the recommendations below:
 - 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
+- MySQL 8.0+ disables `@@global.local_infile` by default, make sure `@@local_infile` is `ON` before enable `allowLoadLocalInfileInPath` of the driver. e.g. run `SET GLOBAL local_infile=ON`, or set it in `mysql.cnf`.
 
 ## License
 

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

@@ -361,7 +361,7 @@ void disableCompression() {
             this.bitmap &= ~COMPRESS;
         }
 
-        void disableLoadDataInfile() {
+        void disableLoadDataLocalInfile() {
             this.bitmap &= ~LOCAL_FILES;
         }
 

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

@@ -22,6 +22,7 @@
 import io.asyncer.r2dbc.mysql.constant.ZeroDateOption;
 import org.jetbrains.annotations.Nullable;
 
+import java.nio.file.Path;
 import java.time.ZoneId;
 
 import static io.asyncer.r2dbc.mysql.internal.util.AssertUtils.requireNonNull;
@@ -42,6 +43,11 @@ public final class ConnectionContext implements CodecContext {
 
     private final ZeroDateOption zeroDateOption;
 
+    @Nullable
+    private final Path localInfilePath;
+
+    private final int localInfileBufferSize;
+
     @Nullable
     private ZoneId serverZoneId;
 
@@ -53,8 +59,11 @@ public final class ConnectionContext implements CodecContext {
 
     private volatile Capability capability = null;
 
-    ConnectionContext(ZeroDateOption zeroDateOption, @Nullable ZoneId serverZoneId) {
+    ConnectionContext(ZeroDateOption zeroDateOption, @Nullable Path localInfilePath,
+        int localInfileBufferSize, @Nullable ZoneId serverZoneId) {
         this.zeroDateOption = requireNonNull(zeroDateOption, "zeroDateOption must not be null");
+        this.localInfilePath = localInfilePath;
+        this.localInfileBufferSize = localInfileBufferSize;
         this.serverZoneId = serverZoneId;
     }
 
@@ -119,6 +128,25 @@ public ZeroDateOption getZeroDateOption() {
         return zeroDateOption;
     }
 
+    /**
+     * Gets the allowed local infile path.
+     *
+     * @return the path.
+     */
+    @Nullable
+    public Path getLocalInfilePath() {
+        return localInfilePath;
+    }
+
+    /**
+     * Gets the local infile buffer size.
+     *
+     * @return the buffer size.
+     */
+    public int getLocalInfileBufferSize() {
+        return localInfileBufferSize;
+    }
+
     /**
      * Get the bitmap of server statuses.
      *

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

@@ -25,6 +25,8 @@
 
 import javax.net.ssl.HostnameVerifier;
 import java.net.Socket;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.time.Duration;
 import java.time.ZoneId;
 import java.util.ArrayList;
@@ -89,6 +91,11 @@ public final class MySqlConnectionConfiguration {
     @Nullable
     private final Predicate<String> preferPrepareStatement;
 
+    @Nullable
+    private final Path loadLocalInfilePath;
+
+    private final int localInfileBufferSize;
+
     private final int queryCacheSize;
 
     private final int prepareCacheSize;
@@ -104,6 +111,7 @@ private MySqlConnectionConfiguration(
         @Nullable Duration socketTimeout, ZeroDateOption zeroDateOption, @Nullable ZoneId serverZoneId,
         String user, @Nullable CharSequence password, @Nullable String database,
         boolean createDatabaseIfNotExist, @Nullable Predicate<String> preferPrepareStatement,
+        @Nullable Path loadLocalInfilePath, int localInfileBufferSize,
         int queryCacheSize, int prepareCacheSize, Extensions extensions,
         @Nullable Publisher<String> passwordPublisher
     ) {
@@ -122,6 +130,8 @@ private MySqlConnectionConfiguration(
         this.database = database == null || database.isEmpty() ? "" : database;
         this.createDatabaseIfNotExist = createDatabaseIfNotExist;
         this.preferPrepareStatement = preferPrepareStatement;
+        this.loadLocalInfilePath = loadLocalInfilePath;
+        this.localInfileBufferSize = localInfileBufferSize;
         this.queryCacheSize = queryCacheSize;
         this.prepareCacheSize = prepareCacheSize;
         this.extensions = extensions;
@@ -207,6 +217,15 @@ Predicate<String> getPreferPrepareStatement() {
         return preferPrepareStatement;
     }
 
+    @Nullable
+    Path getLoadLocalInfilePath() {
+        return loadLocalInfilePath;
+    }
+
+    int getLocalInfileBufferSize() {
+        return localInfileBufferSize;
+    }
+
     int getQueryCacheSize() {
         return queryCacheSize;
     }
@@ -248,6 +267,8 @@ public boolean equals(Object o) {
             database.equals(that.database) &&
             createDatabaseIfNotExist == that.createDatabaseIfNotExist &&
             Objects.equals(preferPrepareStatement, that.preferPrepareStatement) &&
+            Objects.equals(loadLocalInfilePath, that.loadLocalInfilePath) &&
+            localInfileBufferSize == that.localInfileBufferSize &&
             queryCacheSize == that.queryCacheSize &&
             prepareCacheSize == that.prepareCacheSize &&
             extensions.equals(that.extensions) &&
@@ -258,7 +279,8 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hash(isHost, domain, port, ssl, tcpKeepAlive, tcpNoDelay, connectTimeout,
             socketTimeout, serverZoneId, zeroDateOption, user, password, database, createDatabaseIfNotExist,
-            preferPrepareStatement, queryCacheSize, prepareCacheSize, extensions, passwordPublisher);
+            preferPrepareStatement, loadLocalInfilePath, localInfileBufferSize, queryCacheSize,
+            prepareCacheSize, extensions, passwordPublisher);
     }
 
     @Override
@@ -269,16 +291,21 @@ public String toString() {
                 connectTimeout + ", socketTimeout=" + socketTimeout + ", serverZoneId=" + serverZoneId +
                 ", zeroDateOption=" + zeroDateOption + ", user='" + user + "', password=" + password +
                 ", database='" + database + "', createDatabaseIfNotExist=" + createDatabaseIfNotExist +
-                ", preferPrepareStatement=" + preferPrepareStatement + ", queryCacheSize=" + queryCacheSize +
-                ", prepareCacheSize=" + prepareCacheSize + ", extensions=" + extensions +
-                ", passwordPublisher=" + passwordPublisher + '}';
+                ", preferPrepareStatement=" + preferPrepareStatement +
+                ", loadLocalInfilePath=" + loadLocalInfilePath +
+                ", localInfileBufferSize=" + localInfileBufferSize +
+                ", queryCacheSize=" + queryCacheSize + ", prepareCacheSize=" + prepareCacheSize +
+                ", extensions=" + extensions + ", passwordPublisher=" + passwordPublisher + '}';
         }
 
         return "MySqlConnectionConfiguration{unixSocket='" + domain + "', connectTimeout=" +
             connectTimeout + ", socketTimeout=" + socketTimeout + ", serverZoneId=" + serverZoneId +
             ", zeroDateOption=" + zeroDateOption + ", user='" + user + "', password=" + password +
             ", database='" + database + "', createDatabaseIfNotExist=" + createDatabaseIfNotExist +
-            ", preferPrepareStatement=" + preferPrepareStatement + ", queryCacheSize=" + queryCacheSize +
+            ", preferPrepareStatement=" + preferPrepareStatement +
+            ", loadLocalInfilePath=" + loadLocalInfilePath +
+            ", localInfileBufferSize=" + localInfileBufferSize +
+            ", queryCacheSize=" + queryCacheSize +
             ", prepareCacheSize=" + prepareCacheSize + ", extensions=" + extensions +
             ", passwordPublisher=" + passwordPublisher + '}';
     }
@@ -345,6 +372,11 @@ public static final class Builder {
         @Nullable
         private Predicate<String> preferPrepareStatement;
 
+        @Nullable
+        private Path loadLocalInfilePath;
+
+        private int localInfileBufferSize = 8192;
+
         private int queryCacheSize = 0;
 
         private int prepareCacheSize = 256;
@@ -379,7 +411,8 @@ public MySqlConnectionConfiguration build() {
                 sslCa, sslKey, sslKeyPassword, sslCert, sslContextBuilderCustomizer);
             return new MySqlConnectionConfiguration(isHost, domain, port, ssl, tcpKeepAlive, tcpNoDelay,
                 connectTimeout, socketTimeout, zeroDateOption, serverZoneId, user, password, database,
-                createDatabaseIfNotExist, preferPrepareStatement, queryCacheSize, prepareCacheSize,
+                createDatabaseIfNotExist, preferPrepareStatement, loadLocalInfilePath,
+                localInfileBufferSize, queryCacheSize, prepareCacheSize,
                 Extensions.from(extensions, autodetectExtensions), passwordPublisher);
         }
 
@@ -754,6 +787,38 @@ public Builder useServerPrepareStatement(Predicate<String> preferPrepareStatemen
             return this;
         }
 
+        /**
+         * Configures to allow the {@code LOAD DATA LOCAL INFILE} statement in the given {@code path} or
+         * disallow the statement.  Default to {@code null} which means not allow the statement.
+         *
+         * @param path which parent path are allowed to load file data, {@code null} means not be allowed.
+         * @return {@link Builder this}.
+         * @throws java.nio.file.InvalidPathException if the string cannot be converted to a {@link Path}.
+         * @since 1.1.0
+         */
+        public Builder allowLoadLocalInfileInPath(@Nullable String path) {
+            this.loadLocalInfilePath = path == null ? null : Paths.get(path);
+
+            return this;
+        }
+
+        /**
+         * Configures the buffer size for {@code LOAD DATA LOCAL INFILE} statement.  Default to {@code 8192}.
+         * <p>
+         * It is used only if {@link #allowLoadLocalInfileInPath(String)} is set.
+         *
+         * @param localInfileBufferSize the buffer size.
+         * @return {@link Builder this}.
+         * @throws IllegalArgumentException if {@code localInfileBufferSize} is not positive.
+         * @since 1.1.0
+         */
+        public Builder localInfileBufferSize(int localInfileBufferSize) {
+            require(localInfileBufferSize > 0, "localInfileBufferSize must be positive");
+
+            this.localInfileBufferSize = localInfileBufferSize;
+            return this;
+        }
+
         /**
          * Configures the maximum size of the {@link Query} parsing cache. Usually it should be power of two.
          * Default to {@code 0}. Driver will use unbounded cache if size is less than {@code 0}.
@@ -823,6 +888,7 @@ public Builder extendWith(Extension extension) {
 
         /**
          * Registers a password publisher function.
+         *
          * @param passwordPublisher function to retrieve password before making connection.
          * @return this {@link Builder}.
          */

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

@@ -90,8 +90,12 @@ public static MySqlConnectionFactory from(MySqlConnectionConfiguration configura
             String user = configuration.getUser();
             CharSequence password = configuration.getPassword();
             SslMode sslMode = ssl.getSslMode();
-            ConnectionContext context = new ConnectionContext(configuration.getZeroDateOption(),
-                configuration.getServerZoneId());
+            ConnectionContext context = new ConnectionContext(
+                configuration.getZeroDateOption(),
+                configuration.getLoadLocalInfilePath(),
+                configuration.getLocalInfileBufferSize(),
+                configuration.getServerZoneId()
+            );
             Extensions extensions = configuration.getExtensions();
             Predicate<String> prepare = configuration.getPreferPrepareStatement();
             int prepareCacheSize = configuration.getPrepareCacheSize();

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

@@ -191,6 +191,9 @@ public final class MySqlConnectionFactoryProvider implements ConnectionFactoryPr
     public static final Option<Object> USE_SERVER_PREPARE_STATEMENT =
         Option.valueOf("useServerPrepareStatement");
 
+    public static final Option<String> ALLOW_LOAD_LOCAL_INFILE_IN_PATH =
+        Option.valueOf("allowLoadLocalInfileInPath");
+
     /**
      * Option to set the maximum size of the {@link Query} parsing cache.  Default to {@code 256}.
      *
@@ -266,6 +269,8 @@ static MySqlConnectionConfiguration setup(ConnectionFactoryOptions options) {
             .to(builder::zeroDateOption);
         mapper.optional(USE_SERVER_PREPARE_STATEMENT).prepare(builder::useClientPrepareStatement,
             builder::useServerPrepareStatement, builder::useServerPrepareStatement);
+        mapper.optional(ALLOW_LOAD_LOCAL_INFILE_IN_PATH).asString()
+            .to(builder::allowLoadLocalInfileInPath);
         mapper.optional(QUERY_CACHE_SIZE).asInt()
             .to(builder::queryCacheSize);
         mapper.optional(PREPARE_CACHE_SIZE).asInt()