Introduce compatibility mode

We now support a compatibility mode for extended query flows. Enabling compatibility mode fetches all rows in auto-commit mode regardless the configured fetch size. Cursors are only used with explicit transactions and using Sync instead of Flush to work with newer pgpool versions.

Compatibility mode is disabled by default to retain semantics of the previous driver version.

[fixes #341][#373]

Author: mp911de

README.md

@@ -79,6 +79,7 @@ Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
 | `database`        | Database to select. _(Optional)_
 | `applicationName` | The name of the application connecting to the database.  Defaults to `r2dbc-postgresql`. _(Optional)_
 | `autodetectExtensions` | Whether to auto-detect and register `Extension`s from the class path.  Defaults to `true`. _(Optional)_
+| `compatibilityMode` | Enable compatibility mode for cursored fetching. Required when using newer pgpool versions. Defaults to `false`. _(Optional)_
 | `errorResponseLogLevel` | Log level for error responses. Any of `OFF`, `DEBUG`, `INFO`, `WARN` or `ERROR`  Defaults to `DEBUG`. _(Optional)_
 | `fetchSize`       | The default number of rows to return when fetching results. Defaults to `0` for unlimited. _(Optional)_
 | `forceBinary`     | Whether to force binary transfer.  Defaults to `false`. _(Optional)_
@@ -156,21 +157,30 @@ If you'd rather like the latest snapshots of the upcoming major version, use our
 </dependency>
 
 <repository>
-  <id>sonatype-nexus-snapshots</id>
-  <name>Sonatype OSS Snapshot Repository</name>
-  <url>https://oss.sonatype.org/content/repositories/snapshots</url>
+<id>sonatype-nexus-snapshots</id>
+<name>Sonatype OSS Snapshot Repository</name>
+<url>https://oss.sonatype.org/content/repositories/snapshots</url>
 </repository>
 ```
 
+## Cursors
+
+R2DBC Postgres supports both, the [simple](https://www.postgresql.org/docs/current/protocol-flow.html#id-1.10.5.7.4)
+and [extended](https://www.postgresql.org/docs/current/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY) message flow.
+
+Cursored fetching is activated by configuring a `fetchSize`. Postgres cursors are valid for the duration of a transaction. R2DBC can use cursors in auto-commit mode (`Execute` and `Flush`) to not
+require an explicit transaction (`BEGIN…COMMIT/ROLLBACK`). Newer pgpool versions don't support this feature. To work around this limitation, either use explicit transactions when configuring a fetch
+size or enable compatibility mode. Compatibility mode avoids cursors in auto-commit mode (`Execute` with no limit + `Sync`). Cursors in a transaction use `Execute` (with fetch size as limit) + `Sync`
+as message flow.
+
 ## Listen/Notify
 
-Listen and Notify provide a simple form of signal or inter-process communication mechanism for processes accessing the same PostgreSQL database.
-For Listen/Notify, two actors are involved: The sender (notify) and the receiver (listen). The following example uses two connections
-to illustrate how they work together:
+Listen and Notify provide a simple form of signal or inter-process communication mechanism for processes accessing the same PostgreSQL database. For Listen/Notify, two actors are involved: The
+sender (notify) and the receiver (listen). The following example uses two connections to illustrate how they work together:
 
 ```java
-PostgresqlConnection sender = …;
-PostgresqlConnection receiver = …;
+PostgresqlConnection sender= …;
+        PostgresqlConnection receiver= …;
 
 Flux<Notification> listen = receiver.createStatement("LISTEN mymessage")
                                 .execute()

pom.xml

@@ -261,6 +261,7 @@
                     <compilerArgs>
                         <arg>-Werror</arg>
                         <arg>-Xlint:all</arg>
+                        <arg>-Xlint:-deprecation</arg>
                         <arg>-Xlint:-options</arg>
                         <arg>-Xlint:-processing</arg>
                         <arg>-Xlint:-serial</arg>

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

@@ -22,13 +22,16 @@
 import io.r2dbc.postgresql.client.Binding;
 import io.r2dbc.postgresql.client.Client;
 import io.r2dbc.postgresql.client.ExtendedQueryMessageFlow;
+import io.r2dbc.postgresql.client.TransactionStatus;
 import io.r2dbc.postgresql.message.backend.BackendMessage;
 import io.r2dbc.postgresql.message.backend.BindComplete;
+import io.r2dbc.postgresql.message.backend.CloseComplete;
 import io.r2dbc.postgresql.message.backend.CommandComplete;
 import io.r2dbc.postgresql.message.backend.ErrorResponse;
 import io.r2dbc.postgresql.message.backend.NoData;
 import io.r2dbc.postgresql.message.backend.ParseComplete;
 import io.r2dbc.postgresql.message.backend.PortalSuspended;
+import io.r2dbc.postgresql.message.backend.ReadyForQuery;
 import io.r2dbc.postgresql.message.frontend.Bind;
 import io.r2dbc.postgresql.message.frontend.Close;
 import io.r2dbc.postgresql.message.frontend.CompositeFrontendMessage;
@@ -98,11 +101,23 @@ public static Flux<BackendMessage> runQuery(ConnectionResources resources, Excep
         messagesToSend.add(new Describe(portal, PORTAL));
 
         Flux<BackendMessage> exchange;
+        boolean compatibilityMode = resources.getConfiguration().isCompatibilityMode();
+        boolean implicitTransactions = resources.getClient().getTransactionStatus() == TransactionStatus.IDLE;
 
-        if (fetchSize == NO_LIMIT) {
-            exchange = fetchAll(messagesToSend, client, portal);
+        if (compatibilityMode) {
+
+            if (fetchSize == NO_LIMIT || implicitTransactions) {
+                exchange = fetchAll(messagesToSend, client, portal);
+            } else {
+                exchange = fetchCursoredWithSync(messagesToSend, client, portal, fetchSize);
+            }
         } else {
-            exchange = fetchOptimisticCursored(messagesToSend, client, portal, fetchSize);
+
+            if (fetchSize == NO_LIMIT) {
+                exchange = fetchAll(messagesToSend, client, portal);
+            } else {
+                exchange = fetchCursoredWithFlush(messagesToSend, client, portal, fetchSize);
+            }
         }
 
         if (prepareRequired) {
@@ -136,6 +151,68 @@ private static Flux<BackendMessage> fetchAll(List<FrontendMessage.DirectEncoder>
             .as(Operators::discardOnCancel);
     }
 
+    /**
+     * Execute a chunked query and indicate to fetch rows in chunks with the {@link Execute} message.
+     *
+     * @param messagesToSend the messages to send
+     * @param client         client to use
+     * @param portal         the portal
+     * @param fetchSize      fetch size per roundtrip
+     * @return the resulting message stream
+     */
+    private static Flux<BackendMessage> fetchCursoredWithSync(List<FrontendMessage.DirectEncoder> messagesToSend, Client client, String portal, int fetchSize) {
+
+        DirectProcessor<FrontendMessage> requestsProcessor = DirectProcessor.create();
+        FluxSink<FrontendMessage> requestsSink = requestsProcessor.sink();
+        AtomicBoolean isCanceled = new AtomicBoolean(false);
+        AtomicBoolean done = new AtomicBoolean(false);
+
+        messagesToSend.add(new Execute(portal, fetchSize));
+        messagesToSend.add(Sync.INSTANCE);
+
+        return client.exchange(it -> done.get() && it instanceof ReadyForQuery, Flux.<FrontendMessage>just(new CompositeFrontendMessage(messagesToSend)).concatWith(requestsProcessor))
+            .handle((BackendMessage message, SynchronousSink<BackendMessage> sink) -> {
+
+                if (message instanceof CommandComplete) {
+                    requestsSink.next(new Close(portal, PORTAL));
+                    requestsSink.next(Sync.INSTANCE);
+                    requestsSink.complete();
+                    sink.next(message);
+                } else if (message instanceof CloseComplete) {
+                    requestsSink.complete();
+                    done.set(true);
+                    sink.next(message);
+                } else if (message instanceof ErrorResponse) {
+                    done.set(true);
+                    requestsSink.next(Sync.INSTANCE);
+                    requestsSink.complete();
+                    sink.next(message);
+                } else if (message instanceof PortalSuspended) {
+
+                    if (isCanceled.get()) {
+                        requestsSink.next(new Close(portal, PORTAL));
+                        requestsSink.next(Sync.INSTANCE);
+                        requestsSink.complete();
+                    } else {
+                        requestsSink.next(new Execute(portal, fetchSize));
+                        requestsSink.next(Sync.INSTANCE);
+                    }
+                } else if (message instanceof NoData) {
+
+                    if (isCanceled.get()) {
+                        requestsSink.next(new Close(portal, PORTAL));
+                        requestsSink.next(Sync.INSTANCE);
+                        requestsSink.complete();
+                    } else {
+                        done.set(true);
+                    }
+                } else {
+                    sink.next(message);
+                }
+            }).doFinally(ignore -> requestsSink.complete())
+            .as(flux -> Operators.discardOnCancel(flux, () -> isCanceled.set(true)));
+    }
+
     /**
      * Execute a contiguous query and indicate to fetch rows in chunks with the {@link Execute} message. Uses {@link Flush}-based synchronization that creates a cursor. Note that flushing keeps the
      * cursor open even with implicit transactions and this method may not work with newer pgpool implementations.
@@ -146,7 +223,7 @@ private static Flux<BackendMessage> fetchAll(List<FrontendMessage.DirectEncoder>
      * @param fetchSize      fetch size per roundtrip
      * @return the resulting message stream
      */
-    private static Flux<BackendMessage> fetchOptimisticCursored(List<FrontendMessage.DirectEncoder> messagesToSend, Client client, String portal, int fetchSize) {
+    private static Flux<BackendMessage> fetchCursoredWithFlush(List<FrontendMessage.DirectEncoder> messagesToSend, Client client, String portal, int fetchSize) {
 
         DirectProcessor<FrontendMessage> requestsProcessor = DirectProcessor.create();
         FluxSink<FrontendMessage> requestsSink = requestsProcessor.sink();

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

@@ -66,6 +66,8 @@ public final class PostgresqlConnectionConfiguration {
 
     private final boolean autodetectExtensions;
 
+    private final boolean compatibilityMode;
+
     private final Duration connectTimeout;
 
     private final String database;
@@ -105,13 +107,14 @@ public final class PostgresqlConnectionConfiguration {
 
     private final String username;
 
-    private PostgresqlConnectionConfiguration(String applicationName, boolean autodetectExtensions, @Nullable Duration connectTimeout, @Nullable String database, LogLevel errorResponseLogLevel,
+    private PostgresqlConnectionConfiguration(String applicationName, boolean autodetectExtensions, @Nullable boolean compatibilityMode,Duration connectTimeout, @Nullable String database, LogLevel errorResponseLogLevel,
                                               List<Extension> extensions, ToIntFunction<String> fetchSize, boolean forceBinary, @Nullable String host, @Nullable LoopResources loopResources,
                                               LogLevel noticeLogLevel, @Nullable Map<String, String> options, @Nullable CharSequence password, int port, boolean preferAttachedBuffers,
                                               int preparedStatementCacheQueries, @Nullable String schema, @Nullable String socket, SSLConfig sslConfig, boolean tcpKeepAlive, boolean tcpNoDelay,
                                               String username) {
         this.applicationName = Assert.requireNonNull(applicationName, "applicationName must not be null");
         this.autodetectExtensions = autodetectExtensions;
+        this.compatibilityMode = compatibilityMode;
         this.connectTimeout = connectTimeout;
         this.errorResponseLogLevel = errorResponseLogLevel;
         this.extensions = Assert.requireNonNull(extensions, "extensions must not be null");
@@ -152,6 +155,7 @@ public String toString() {
         return "PostgresqlConnectionConfiguration{" +
             "applicationName='" + this.applicationName + '\'' +
             ", autodetectExtensions='" + this.autodetectExtensions + '\'' +
+            ", compatibilityMode=" + this.compatibilityMode +
             ", connectTimeout=" + this.connectTimeout +
             ", errorResponseLogLevel=" + this.errorResponseLogLevel +
             ", database='" + this.database + '\'' +
@@ -176,6 +180,10 @@ String getApplicationName() {
         return this.applicationName;
     }
 
+    boolean isCompatibilityMode() {
+        return this.compatibilityMode;
+    }
+
     @Nullable
     Duration getConnectTimeout() {
         return this.connectTimeout;
@@ -313,6 +321,8 @@ public static final class Builder {
 
         private boolean autodetectExtensions = true;
 
+        private boolean compatibilityMode = false;
+
         @Nullable
         private Duration connectTimeout;
 
@@ -422,11 +432,25 @@ public PostgresqlConnectionConfiguration build() {
                 throw new IllegalArgumentException("username must not be null");
             }
 
-            return new PostgresqlConnectionConfiguration(this.applicationName, this.autodetectExtensions, this.connectTimeout, this.database, this.errorResponseLogLevel, this.extensions,
-                this.fetchSize, this.forceBinary, this.host, this.loopResources, this.noticeLogLevel, this.options, this.password, this.port, this.preferAttachedBuffers,
+            return new PostgresqlConnectionConfiguration(this.applicationName, this.autodetectExtensions, this.compatibilityMode, this.connectTimeout, this.database, this.errorResponseLogLevel, this.extensions,
+                this.fetchSize
+                , this.forceBinary, this.host, this.loopResources, this.noticeLogLevel, this.options, this.password, this.port, this.preferAttachedBuffers,
                 this.preparedStatementCacheQueries, this.schema, this.socket, this.createSslConfig(), this.tcpKeepAlive, this.tcpNoDelay, this.username);
         }
 
+        /**
+         * Enables protocol compatibility mode for cursored query execution. Cursored query execution applies when configuring a non-zero fetch size. Compatibility mode uses {@code Execute+Sync}
+         * messages instead of {@code Execute+Flush}. The default mode uses optimized fetching which does not work with newer pgpool versions.
+         *
+         * @param compatibilityMode whether to enable compatibility mode
+         * @return this {@link Builder}
+         * @since 0.8.7
+         */
+        public Builder compatibilityMode(boolean compatibilityMode) {
+            this.compatibilityMode = compatibilityMode;
+            return this;
+        }
+
         /**
          * Configure the connection timeout. Default unconfigured.
          *
@@ -786,6 +810,7 @@ public String toString() {
             return "Builder{" +
                 "applicationName='" + this.applicationName + '\'' +
                 ", autodetectExtensions='" + this.autodetectExtensions + '\'' +
+                ", compatibilityMode='" + this.compatibilityMode + '\'' +
                 ", connectTimeout='" + this.connectTimeout + '\'' +
                 ", database='" + this.database + '\'' +
                 ", extensions='" + this.extensions + '\'' +

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

@@ -57,6 +57,13 @@ public final class PostgresqlConnectionFactoryProvider implements ConnectionFact
      */
     public static final Option<Boolean> AUTODETECT_EXTENSIONS = Option.valueOf("autodetectExtensions");
 
+    /**
+     * Compatibility query mode for cursored query execution.
+     *
+     * @since 0.8.7
+     */
+    public static final Option<Boolean> COMPATIBILITY_MODE = Option.valueOf("compatibilityMode");
+
     /**
      * Error Response Log Level.
      */
@@ -226,6 +233,7 @@ private static PostgresqlConnectionConfiguration.Builder fromConnectionFactoryOp
 
         mapper.from(APPLICATION_NAME).to(builder::applicationName);
         mapper.from(AUTODETECT_EXTENSIONS).map(OptionMapper::toBoolean).to(builder::autodetectExtensions);
+        mapper.from(COMPATIBILITY_MODE).map(OptionMapper::toBoolean).to(builder::compatibilityMode);
         mapper.from(CONNECT_TIMEOUT).map(OptionMapper::toDuration).to(builder::connectTimeout);
         mapper.from(CURRENT_SCHEMA).to(builder::schema).otherwise(() -> mapper.from(SCHEMA).to(builder::schema));
         mapper.from(DATABASE).to(builder::database);

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

@@ -21,26 +21,31 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcOperations;
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 /**
- * Integration tests for {@link ExtendedQueryPostgresqlStatement}
+ * Integration tests for {@link ExtendedQueryPostgresqlStatement}.
  */
-final class ExtendedQueryPostgresqlStatementIntegrationTests extends AbstractIntegrationTests {
+class ExtendedQueryPostgresqlStatementIntegrationTests extends AbstractIntegrationTests {
 
     @BeforeEach
     void setUp() {
         super.setUp();
-        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
-        SERVER.getJdbcOperations().execute("CREATE TABLE test (id SERIAL PRIMARY KEY, val VARCHAR(255))");
-        SERVER.getJdbcOperations().execute("INSERT INTO test (val) VALUES ('a'), ('a'), ('b'), ('c'), ('c')");
+        getJdbcOperations().execute("DROP TABLE IF EXISTS test");
+        getJdbcOperations().execute("CREATE TABLE test (id SERIAL PRIMARY KEY, val VARCHAR(255))");
+        getJdbcOperations().execute("INSERT INTO test (val) VALUES ('a'), ('a'), ('b'), ('c'), ('c')");
     }
 
     @AfterEach
     void tearDown() {
         super.tearDown();
-        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
+        getJdbcOperations().execute("DROP TABLE IF EXISTS test");
+    }
+
+    private JdbcOperations getJdbcOperations() {
+        return SERVER.getJdbcOperations();
     }
 
     @Override