use "schema" as "search_path" connection option

README.md

@@ -82,7 +82,7 @@ Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
 | `forceBinary`     | Whether to force binary transfer.  Defaults to `false`. _(Optional)_
 | `preparedStatementCacheQueries` | Determine the number of queries that are cached in each connection. The default is `-1`, meaning there's no limit. The value of `0` disables the cache. Any other value specifies the cache size.
 | `options`         | A `Map<String, String>` of connection parameters. These are applied to each database connection created by the `ConnectionFactory`. Useful for setting generic [PostgreSQL connection parameters][psql-runtime-config]. _(Optional)_
-| `schema`          | The schema to set. _(Optional)_
+| `schema`          | The search path to set. _(Optional)_
 | `sslMode`         | SSL mode to use, see `SSLMode` enum. Supported values: `DISABLE`, `ALLOW`, `PREFER`, `REQUIRE`, `VERIFY_CA`, `VERIFY_FULL`. _(Optional)_
 | `sslRootCert`     | Path to SSL CA certificate in PEM format. _(Optional)_
 | `sslKey`          | Path to SSL key for TLS authentication in PEM format. _(Optional)_

src/main/java/io/r2dbc/postgresql/PostgresqlConnectionConfiguration.java

@@ -34,6 +34,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.HashMap;
 import java.util.ServiceLoader;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -74,8 +75,6 @@ public final class PostgresqlConnectionConfiguration {
 
     private final int port;
 
-    private final String schema;
-
     private final String socket;
 
     private final String username;
@@ -97,16 +96,30 @@ private PostgresqlConnectionConfiguration(String applicationName, boolean autode
         this.fetchSize = fetchSize;
         this.forceBinary = forceBinary;
         this.host = host;
-        this.options = options;
+        this.options = initOptions(options);
         this.password = password;
         this.port = port;
-        this.schema = schema;
+        addToOptions(schema);
         this.socket = socket;
         this.username = Assert.requireNonNull(username, "username must not be null");
         this.sslConfig = sslConfig;
         this.preparedStatementCacheQueries = preparedStatementCacheQueries;
     }
 
+    private Map<String, String> initOptions(@Nullable Map<String, String> options) {
+        if (options == null) {
+            return new HashMap<>();
+        } else {
+            return options;
+        }
+    }
+
+    private void addToOptions(String schema) {
+        if (schema != null && !schema.isEmpty()) {
+            options.put("search_path", schema);
+        }
+    }
+
     /**
      * Returns a new {@link Builder}.
      *
@@ -131,7 +144,6 @@ public String toString() {
             ", options='" + this.options + '\'' +
             ", password='" + obfuscate(this.password != null ? this.password.length() : 0) + '\'' +
             ", port=" + this.port +
-            ", schema='" + this.schema + '\'' +
             ", username='" + this.username + '\'' +
             '}';
     }
@@ -192,11 +204,6 @@ int getPort() {
         return this.port;
     }
 
-    @Nullable
-    String getSchema() {
-        return this.schema;
-    }
-
     @Nullable
     String getSocket() {
         return this.socket;

src/main/java/io/r2dbc/postgresql/PostgresqlConnectionFactory.java

@@ -201,7 +201,6 @@ private Mono<Client> tryConnectWithConfig(SSLConfig sslConfig, @Nullable Map<Str
     private Publisher<?> prepareConnection(PostgresqlConnection connection, ByteBufAllocator byteBufAllocator, DefaultCodecs codecs) {
 
         List<Publisher<?>> publishers = new ArrayList<>();
-        publishers.add(setSchema(connection));
 
         this.extensions.forEach(CodecRegistrar.class, it -> {
             publishers.add(it.register(connection, byteBufAllocator, codecs));
@@ -269,16 +268,6 @@ private Mono<IsolationLevel> getIsolationLevel(io.r2dbc.postgresql.api.Postgresq
             })).defaultIfEmpty(IsolationLevel.READ_COMMITTED).last();
     }
 
-    private Mono<Void> setSchema(PostgresqlConnection connection) {
-        if (this.configuration.getSchema() == null) {
-            return Mono.empty();
-        }
-
-        return connection.createStatement(String.format("SET SCHEMA '%s'", this.configuration.getSchema()))
-            .execute()
-            .then();
-    }
-
     static class PostgresConnectionException extends R2dbcNonTransientResourceException {
 
         public PostgresConnectionException(String msg, @Nullable Throwable cause) {

src/main/java/io/r2dbc/postgresql/PostgresqlConnectionFactoryProvider.java

@@ -73,10 +73,15 @@ public final class PostgresqlConnectionFactoryProvider implements ConnectionFact
     public static final String LEGACY_POSTGRESQL_DRIVER = "postgres";
 
     /**
-     * Schema.
+     * Schema search path (alias for "currentSchema").
      */
     public static final Option<String> SCHEMA = Option.valueOf("schema");
 
+    /**
+     * Schema search path.
+     */
+    public static final Option<String> CURRENT_SCHEMA = Option.valueOf("currentSchema");
+
     /**
      * Unix domain socket.
      */
@@ -247,7 +252,13 @@ private static PostgresqlConnectionConfiguration.Builder fromConnectionFactoryOp
         builder.connectTimeout(connectionFactoryOptions.getValue(CONNECT_TIMEOUT));
         builder.database(connectionFactoryOptions.getValue(DATABASE));
         builder.password(connectionFactoryOptions.getValue(PASSWORD));
-        builder.schema(connectionFactoryOptions.getValue(SCHEMA));
+
+        if (connectionFactoryOptions.getValue(CURRENT_SCHEMA) != null) {
+            builder.schema(connectionFactoryOptions.getValue(CURRENT_SCHEMA));
+        } else {
+            builder.schema(connectionFactoryOptions.getValue(SCHEMA));
+        }
+
         builder.username(connectionFactoryOptions.getRequiredValue(USER));
 
         String applicationName = connectionFactoryOptions.getValue(APPLICATION_NAME);

src/test/java/io/r2dbc/postgresql/PostgresqlConnectionConfigurationTest.java

@@ -80,12 +80,16 @@ void configuration() {
             .hasFieldOrPropertyWithValue("connectTimeout", Duration.ofMillis(1000))
             .hasFieldOrPropertyWithValue("database", "test-database")
             .hasFieldOrPropertyWithValue("host", "test-host")
-            .hasFieldOrPropertyWithValue("options", options)
+            .hasFieldOrProperty("options")
             .hasFieldOrPropertyWithValue("password", null)
             .hasFieldOrPropertyWithValue("port", 100)
-            .hasFieldOrPropertyWithValue("schema", "test-schema")
             .hasFieldOrPropertyWithValue("username", "test-username")
             .hasFieldOrProperty("sslConfig");
+
+        assertThat(configuration.getOptions())
+                .containsEntry("lock_timeout", "10s")
+                .containsEntry("statement_timeout", "60000")
+                .containsEntry("search_path", "test-schema");
     }
 
     @Test
@@ -104,9 +108,12 @@ void configurationDefaults() {
             .hasFieldOrPropertyWithValue("host", "test-host")
             .hasFieldOrPropertyWithValue("password", "test-password")
             .hasFieldOrPropertyWithValue("port", 5432)
-            .hasFieldOrPropertyWithValue("schema", "test-schema")
+            .hasFieldOrProperty("options")
             .hasFieldOrPropertyWithValue("username", "test-username")
             .hasFieldOrProperty("sslConfig");
+
+        assertThat(configuration.getOptions())
+                .containsEntry("search_path", "test-schema");
     }
 
     @Test

src/test/java/io/r2dbc/postgresql/PostgresqlConnectionFactoryTest.java

@@ -63,7 +63,7 @@ void createAuthenticationMD5Password() {
         // @formatter:off
         Client client = TestClient.builder()
             .window()
-                .expectRequest(new StartupMessage("test-application-name", "test-database", "test-username", null)).thenRespond(new AuthenticationMD5Password(TEST.buffer(4).writeInt(100)))
+                .expectRequest(new StartupMessage("test-application-name", "test-database", "test-username", Collections.emptyMap())).thenRespond(new AuthenticationMD5Password(TEST.buffer(4).writeInt(100)))
                 .expectRequest(new PasswordMessage("md55e9836cdb369d50e3bc7d127e88b4804")).thenRespond(AuthenticationOk.INSTANCE)
                 .done()
             .build();