Polishing

Tweak fetchsize defaulting to ignore simple batch statements. Introduce FixedFetchSize type to aid fixed fetch size debugging. Rearrange methods.

Reuse statement cache.

[closes #256][closes #257]

Author: mp911de

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

@@ -36,7 +36,6 @@
 import java.util.regex.Matcher;
 
 import static io.r2dbc.postgresql.client.ExtendedQueryMessageFlow.PARAMETER_SYMBOL;
-import static io.r2dbc.postgresql.message.frontend.Execute.NO_LIMIT;
 import static io.r2dbc.postgresql.util.PredicateUtils.not;
 import static io.r2dbc.postgresql.util.PredicateUtils.or;
 
@@ -53,7 +52,7 @@ final class ExtendedQueryPostgresqlStatement implements PostgresqlStatement {
 
     private final String sql;
 
-    private int fetchSize = NO_LIMIT;
+    private int fetchSize;
 
     private String[] generatedColumns;
 

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

@@ -87,7 +87,7 @@ public String toString() {
     private Mono<String> parse(String sql, int[] types) {
         String name = "S_" + this.counter.getAndIncrement();
 
-        ExceptionFactory factory = ExceptionFactory.withSql(name);
+        ExceptionFactory factory = ExceptionFactory.withSql(sql);
         return ExtendedQueryMessageFlow
             .parse(this.client, name, sql, types)
             .handle(factory::handleErrorResponse)

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

@@ -84,7 +84,8 @@ public final class PostgresqlConnectionConfiguration {
     private final SSLConfig sslConfig;
 
     private PostgresqlConnectionConfiguration(String applicationName, boolean autodetectExtensions,
-                                              @Nullable Duration connectTimeout, @Nullable String database, List<Extension> extensions, ToIntFunction<String> fetchSize, boolean forceBinary, @Nullable String host,
+                                              @Nullable Duration connectTimeout, @Nullable String database, List<Extension> extensions, ToIntFunction<String> fetchSize, boolean forceBinary,
+                                              @Nullable String host,
                                               @Nullable Map<String, String> options, @Nullable CharSequence password, int port, @Nullable String schema, @Nullable String socket, String username,
                                               SSLConfig sslConfig) {
         this.applicationName = Assert.requireNonNull(applicationName, "applicationName must not be null");
@@ -166,7 +167,7 @@ ToIntFunction<String> getFetchSize() {
     }
 
     int getFetchSize(String sql) {
-        return this.fetchSize != null ? this.fetchSize.applyAsInt(sql) : NO_LIMIT;
+        return this.fetchSize.applyAsInt(sql);
     }
 
     @Nullable
@@ -259,8 +260,7 @@ public static final class Builder {
 
         private List<Extension> extensions = new ArrayList<>();
 
-        @Nullable
-        private ToIntFunction<String> fetchSize;
+        private ToIntFunction<String> fetchSize = sql -> NO_LIMIT;
 
         private boolean forceBinary = false;
 
@@ -346,7 +346,8 @@ public PostgresqlConnectionConfiguration build() {
                 throw new IllegalArgumentException("username must not be null");
             }
 
-            return new PostgresqlConnectionConfiguration(this.applicationName, this.autodetectExtensions, this.connectTimeout, this.database, this.extensions, this.fetchSize, this.forceBinary, this.host,
+            return new PostgresqlConnectionConfiguration(this.applicationName, this.autodetectExtensions, this.connectTimeout, this.database, this.extensions, this.fetchSize, this.forceBinary,
+                this.host,
                 this.options, this.password, this.port, this.schema, this.socket, this.username, this.createSslConfig());
         }
 
@@ -403,23 +404,25 @@ public Builder extendWith(Extension extension) {
         }
 
         /**
-         * Set the default number of rows to return when fetching results from a query instead deriving fetch size from
-         * back pressure. If the value specified is zero, then the hint is ignored.
+         * Set the default number of rows to return when fetching results from a query. If the value specified is zero, then the hint is ignored and queries request all rows when running a statement.
          *
          * @param fetchSize the number of rows to fetch
          * @return this {@code Builder}
+         * @throws IllegalArgumentException if {@code fetchSize} is negative
+         * @since 0.8.2
          */
         public Builder fetchSize(int fetchSize) {
             Assert.isTrue(fetchSize >= 0, "fetch size must be greater or equal zero");
-            this.fetchSize = sql -> fetchSize;
-            return this;
+            return fetchSize(new FixedFetchSize(fetchSize));
         }
 
         /**
          * Set a function that maps a SQL query to the number of rows to return when fetching results for that query.
          *
          * @param fetchSizeFunction a function that maps the number of rows to fetch
          * @return this {@code Builder}
+         * @throws IllegalArgumentException if {@code fetchSizeFunction} is {@code null}
+         * @since 0.8.2
          */
         public Builder fetchSize(ToIntFunction<String> fetchSizeFunction) {
             Assert.requireNonNull(fetchSizeFunction, "fetch size function must be non null");
@@ -596,6 +599,7 @@ public String toString() {
                 ", connectTimeout='" + this.connectTimeout + '\'' +
                 ", database='" + this.database + '\'' +
                 ", extensions='" + this.extensions + '\'' +
+                ", fetchSize='" + this.fetchSize + '\'' +
                 ", forceBinary='" + this.forceBinary + '\'' +
                 ", host='" + this.host + '\'' +
                 ", parameters='" + this.options + '\'' +
@@ -694,4 +698,23 @@ private Supplier<SslProvider> createSslProvider() {
                 .build();
         }
     }
+
+    static class FixedFetchSize implements ToIntFunction<String> {
+
+        private final int fetchSize;
+
+        public FixedFetchSize(int fetchSize) {
+            this.fetchSize = fetchSize;
+        }
+
+        @Override
+        public int applyAsInt(String value) {
+            return this.fetchSize;
+        }
+
+        @Override
+        public String toString() {
+            return "" + this.fetchSize;
+        }
+    }
 }

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

@@ -176,7 +176,7 @@ private Mono<PostgresqlConnection> doCreateConnection(boolean forReplication, @N
                 }
                 return isolationLevelMono
                     // actual connection to be used
-                    .map(isolationLevel -> new PostgresqlConnection(client, codecs, DefaultPortalNameSupplier.INSTANCE, new IndefiniteStatementCache(client), isolationLevel, this.configuration))
+                    .map(isolationLevel -> new PostgresqlConnection(client, codecs, DefaultPortalNameSupplier.INSTANCE, statementCache, isolationLevel, this.configuration))
                     .delayUntil(connection -> {
                         return prepareConnection(connection, client.getByteBufAllocator(), codecs);
                     })

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

@@ -51,12 +51,16 @@ final class SimpleQueryPostgresqlStatement implements PostgresqlStatement {
 
     private String[] generatedColumns;
 
-    private int fetchSize = NO_LIMIT;
+    private int fetchSize;
 
     SimpleQueryPostgresqlStatement(ConnectionContext context, String sql) {
         this.context = Assert.requireNonNull(context, "context must not be null");
         this.sql = Assert.requireNonNull(sql, "sql must not be null");
-        fetchSize(this.context.getConfiguration().getFetchSize(sql));
+        fetchSize(isBatch() ? NO_LIMIT : this.context.getConfiguration().getFetchSize(sql));
+    }
+
+    private boolean isBatch() {
+        return this.sql.contains(";");
     }
 
     @Override
@@ -96,8 +100,9 @@ public Flux<io.r2dbc.postgresql.api.PostgresqlResult> execute() {
     @Override
     public SimpleQueryPostgresqlStatement fetchSize(int rows) {
         Assert.isTrue(rows >= 0, "Fetch size must be greater or equal zero");
+
         if (rows != NO_LIMIT) {
-            Assert.isTrue(!this.sql.contains(";"), "Fetch size can only be used with a single SQL statement");
+            Assert.isTrue(!isBatch(), "Fetch size can only be used with a single SQL statement");
             this.fetchSize = rows;
         }
 

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

@@ -16,18 +16,14 @@
 
 package io.r2dbc.postgresql;
 
-import org.junit.jupiter.api.BeforeEach;
+import io.r2dbc.spi.Statement;
 import org.junit.jupiter.api.Test;
 import reactor.test.StepVerifier;
 
-public class FetchSizeIntegrationTests extends AbstractIntegrationTests {
-
-    @Override
-    @BeforeEach
-    void setUp() {
-        super.setUp();
-        System.gc();
-    }
+/**
+ * Integration tests for {@link Statement#fetchSize(int)}.
+ */
+final class FetchSizeIntegrationTests extends AbstractIntegrationTests {
 
     @Override
     protected void customize(PostgresqlConnectionConfiguration.Builder builder) {
@@ -37,12 +33,12 @@ protected void customize(PostgresqlConnectionConfiguration.Builder builder) {
     @Test
     void exchangeWithDefaultFetchSize() {
         this.connection.createStatement("SELECT * FROM generate_series(1,20) WHERE $1 = $1")
-                .bind(0, 1)
-                .execute()
-                .flatMap(r -> r.map((row, meta) -> row.get(0, Integer.class)))
-                .as(StepVerifier::create)
-                .expectNextCount(20)
-                .verifyComplete();
+            .bind(0, 1)
+            .execute()
+            .flatMap(r -> r.map((row, meta) -> row.get(0, Integer.class)))
+            .as(StepVerifier::create)
+            .expectNextCount(20)
+            .verifyComplete();
     }
 
     @Test