postgresql-jdbc.git / summary / log / commit / refs
commit 1471bd93fce5401c8bc80f686f8beace9572ec79
Author: Craig Ringer <craig@2ndquadrant.com>
Date: Fri Oct 10 05:34:24 2014 +0000
Support batching for prepared statements that return rows.
With this change, PgJDBC calculates the estimated return size for the result
row, assuming that each batch item returns one row for getGeneratedKeys() use.
Instead of forcing batching off and sending individual queries, it then queues
enough binds and executes to fit safely within the receive buffer and server's
send buffer before sending a sync and reading the result buffer.
Batching is still forced off when the result set size cannot be estimated (e.g.
due to unbounded 'text' columns, use of arrays in the result set, etc).
This is a minimal workaround for github issue #195, though it doesn't prevent
deadlocks that can already occur due to #194.
org/postgresql/core/QueryExecutor.java | 3 +
org/postgresql/core/v3/QueryExecutorImpl.java | 74 +++++++++++++++++++-----
org/postgresql/core/v3/SimpleQuery.java | 47 +++++++++++++++
org/postgresql/jdbc2/AbstractJdbc2Statement.java | 10 ++--
4 files changed, 114 insertions(+), 20 deletions(-)
diff --git a/org/postgresql/core/QueryExecutor.java b/org/postgresql/core/QueryExecutor.java
index e80a23c0..10bd5025 100644
--- a/org/postgresql/core/QueryExecutor.java
+++ b/org/postgresql/core/QueryExecutor.java
@@ -92,7 +92,10 @@ public interface QueryExecutor {
/**
* Flag to disable batch execution when we expect results (generated keys)
* from a statement.
+ *
+ * @deprecated in PgJDBC 9.4 as we now auto-size batches.
*/
+ @Deprecated
static int QUERY_DISALLOW_BATCHING = 128;
/**
diff --git a/org/postgresql/core/v3/QueryExecutorImpl.java b/org/postgresql/core/v3/QueryExecutorImpl.java
index 35e33faa..0bbb2179 100644
--- a/org/postgresql/core/v3/QueryExecutorImpl.java
+++ b/org/postgresql/core/v3/QueryExecutorImpl.java
@@ -252,6 +252,7 @@ public class QueryExecutorImpl implements QueryExecutor {
sendQuery((V3Query)query, (V3ParameterList)parameters, maxRows, fetchSize, flags, trackingHandler);
sendSync();
processResults(handler, flags);
+ estimatedReceiveBufferBytes = 0;
}
catch (PGBindException se)
{
@@ -271,6 +272,7 @@ public class QueryExecutorImpl implements QueryExecutor {
//
sendSync();
processResults(handler, flags);
+ estimatedReceiveBufferBytes = 0;
handler.handleError(new PSQLException(GT.tr("Unable to bind parameter values for statement."), PSQLState.INVALID_PARAMETER_VALUE, se.getIOException()));
}
}
@@ -416,6 +418,7 @@ public class QueryExecutorImpl implements QueryExecutor {
{
sendSync();
processResults(handler, flags);
+ estimatedReceiveBufferBytes = 0;
}
}
catch (IOException e)
@@ -562,6 +565,7 @@ public class QueryExecutorImpl implements QueryExecutor {
sendOneQuery(beginTransactionQuery, SimpleQuery.NO_PARAMETERS, 0, 0, QueryExecutor.QUERY_NO_METADATA);
sendSync();
processResults(handler, 0);
+ estimatedReceiveBufferBytes = 0;
}
catch (IOException ioe)
{
@@ -1124,25 +1128,58 @@ public class QueryExecutorImpl implements QueryExecutor {
return op;
}
-
+
/*
- * To prevent client/server protocol deadlocks, we try to manage the estimated
- * recv buffer size and force a sync +flush and process results if we think it
- * might be getting too full.
- *
+ * To prevent client/server protocol deadlocks, we try to manage the
+ * estimated recv buffer size and force a sync +flush and process results if
+ * we think it might be getting too full.
+ *
* See the comments above MAX_BUFFERED_RECV_BYTES's declaration for details.
*/
- private void flushIfDeadlockRisk(boolean disallowBatching, ErrorTrackingResultHandler trackingHandler, int flags) throws IOException
- {
- estimatedReceiveBufferBytes += NODATA_QUERY_RESPONSE_SIZE_BYTES;
- if (disallowBatching || estimatedReceiveBufferBytes >= MAX_BUFFERED_RECV_BYTES)
- {
- if (logger.logDebug())
- logger.debug("Forcing Sync, receive buffer full or batching disallowed");
+ private void flushIfDeadlockRisk(Query query, boolean disallowBatching,
+ ErrorTrackingResultHandler trackingHandler, final int flags)
+ throws IOException {
+ // Assume all statements need at least this much reply buffer space,
+ // plus params
+ estimatedReceiveBufferBytes += NODATA_QUERY_RESPONSE_SIZE_BYTES;
+
+ SimpleQuery sq = (SimpleQuery) query;
+ if (sq.isStatementDescribed()) {
+ /*
+ * Estimate the response size of the fields and add it to the
+ * expected response size.
+ *
+ * It's impossible for us to estimate the rowcount. We'll assume one
+ * row, as that's the common case for batches and we're leaving
+ * plenty of breathing room in this approach. It's still not
+ * deadlock-proof though; see pgjdbc github issues #194 and #195.
+ */
+ int maxResultRowSize = sq.getMaxResultRowSize();
+ if (maxResultRowSize >= 0) {
+ estimatedReceiveBufferBytes += maxResultRowSize;
+ } else {
+ logger.debug("Couldn't estimate result size or result size unbounded, "
+ + "disabling batching for this query.");
+ disallowBatching = true;
+ }
+ } else {
+ /*
+ * We only describe a statement if we're expecting results from it,
+ * so it's legal to batch unprepared statements. We'll abort later
+ * if we get any uresults from them where none are expected. For now
+ * all we can do is hope the user told us the truth and assume that
+ * NODATA_QUERY_RESPONSE_SIZE_BYTES is enough to cover it.
+ */
+ }
+
+ if (disallowBatching
+ || estimatedReceiveBufferBytes >= MAX_BUFFERED_RECV_BYTES) {
+ logger.debug("Forcing Sync, receive buffer full or batching disallowed");
sendSync();
processResults(trackingHandler, flags);
estimatedReceiveBufferBytes = 0;
}
+
}
/*
@@ -1152,11 +1189,16 @@ public class QueryExecutorImpl implements QueryExecutor {
// Now the query itself.
SimpleQuery[] subqueries = query.getSubqueries();
SimpleParameterList[] subparams = parameters.getSubparams();
+
+ // We know this is deprecated, but still respect it in case anyone's using it.
+ // PgJDBC its self no longer does.
+ @SuppressWarnings("deprecation")
boolean disallowBatching = (flags & QueryExecutor.QUERY_DISALLOW_BATCHING) != 0;
if (subqueries == null)
{
- flushIfDeadlockRisk(disallowBatching, trackingHandler, flags);
+ flushIfDeadlockRisk(query, disallowBatching, trackingHandler, flags);
+
// If we saw errors, don't send anything more.
if (!trackingHandler.hasErrors())
sendOneQuery((SimpleQuery)query, (SimpleParameterList)parameters, maxRows, fetchSize, flags);
@@ -1165,7 +1207,8 @@ public class QueryExecutorImpl implements QueryExecutor {
{
for (int i = 0; i < subqueries.length; ++i)
{
- flushIfDeadlockRisk(disallowBatching, trackingHandler, flags);
+ final SimpleQuery subquery = subqueries[i];
+ flushIfDeadlockRisk(subquery, disallowBatching, trackingHandler, flags);
// If we saw errors, don't send anything more.
if (trackingHandler.hasErrors())
@@ -1182,7 +1225,7 @@ public class QueryExecutorImpl implements QueryExecutor {
{
subparam = subparams[i];
}
- sendOneQuery(subqueries[i], subparam, maxRows, fetchSize, flags);
+ sendOneQuery(subquery, subparam, maxRows, fetchSize, flags);
}
}
}
@@ -2146,6 +2189,7 @@ public class QueryExecutorImpl implements QueryExecutor {
sendSync();
processResults(handler, 0);
+ estimatedReceiveBufferBytes = 0;
}
catch (IOException e)
{
diff --git a/org/postgresql/core/v3/SimpleQuery.java b/org/postgresql/core/v3/SimpleQuery.java
index 39e75fce..e48c7d9b 100644
--- a/org/postgresql/core/v3/SimpleQuery.java
+++ b/org/postgresql/core/v3/SimpleQuery.java
@@ -9,6 +9,7 @@
package org.postgresql.core.v3;
import org.postgresql.core.*;
+
import java.lang.ref.PhantomReference;
/**
@@ -63,6 +64,46 @@ class SimpleQuery implements V3Query {
return null;
}
+ /*
+ * Return maximum size in bytes that each result row from this query may
+ * return. Mainly used for batches that return results.
+ *
+ * Results are cached until/unless the query is re-described.
+ *
+ * @return Max size of result data in bytes according to returned fields, 0
+ * if no results, -1 if result is unbounded.
+ *
+ * @throws IllegalStateException if the query is not described
+ */
+ public int getMaxResultRowSize() {
+ if (cachedMaxResultRowSize != null) {
+ return cachedMaxResultRowSize.intValue();
+ }
+ if (!this.statementDescribed) {
+ throw new IllegalStateException(
+ "Cannot estimate result row size on a statement that is not described");
+ }
+ int maxResultRowSize = 0;
+ if (fields != null) {
+ for (int i = 0; i < fields.length; i++) {
+ Field f = fields[i];
+ final int fieldLength = f.getLength();
+ if (fieldLength < 1 || fieldLength >= 65535) {
+ /*
+ * Field length unknown or large; we can't make any safe
+ * estimates about the result size, so we have to fall back to
+ * sending queries individually.
+ */
+ maxResultRowSize = -1;
+ break;
+ }
+ maxResultRowSize += fieldLength;
+ }
+ }
+ cachedMaxResultRowSize = maxResultRowSize;
+ return maxResultRowSize;
+ }
+
//
// Implementation guts
//
@@ -125,6 +166,7 @@ class SimpleQuery implements V3Query {
*/
void setFields(Field[] fields) {
this.fields = fields;
+ this.cachedMaxResultRowSize = null;
}
/**
@@ -143,6 +185,7 @@ class SimpleQuery implements V3Query {
}
void setPortalDescribed(boolean portalDescribed) {
this.portalDescribed = portalDescribed;
+ this.cachedMaxResultRowSize = null;
}
// Have we sent a Describe Statement message for this query yet?
@@ -152,6 +195,7 @@ class SimpleQuery implements V3Query {
}
void setStatementDescribed(boolean statementDescribed) {
this.statementDescribed = statementDescribed;
+ this.cachedMaxResultRowSize = null;
}
void setCleanupRef(PhantomReference cleanupRef) {
@@ -175,6 +219,7 @@ class SimpleQuery implements V3Query {
fields = null;
portalDescribed = false;
statementDescribed = false;
+ cachedMaxResultRowSize = null;
}
private final String[] fragments;
@@ -191,6 +236,8 @@ class SimpleQuery implements V3Query {
private PhantomReference cleanupRef;
private int[] preparedTypes;
+ private Integer cachedMaxResultRowSize;
+
final static SimpleParameterList NO_PARAMETERS = new SimpleParameterList(0, null);
}
diff --git a/org/postgresql/jdbc2/AbstractJdbc2Statement.java b/org/postgresql/jdbc2/AbstractJdbc2Statement.java
index c4b575c3..f8038ca5 100644
--- a/org/postgresql/jdbc2/AbstractJdbc2Statement.java
+++ b/org/postgresql/jdbc2/AbstractJdbc2Statement.java
@@ -2857,11 +2857,11 @@ public abstract class AbstractJdbc2Statement implements BaseStatement
boolean preDescribe = false;
if (wantsGeneratedKeysAlways) {
- // If a batch requests generated keys, we disable batching internally
- // and send each statement individually. See v3.QueryExecutorImpl
- // and the comments on MAX_BUFFERED_RECV_BYTES .
- //
- flags = QueryExecutor.QUERY_BOTH_ROWS_AND_STATUS | QueryExecutor.QUERY_DISALLOW_BATCHING;
+ /*
+ * This batch will return generated keys, tell the executor to
+ * expect result rows. We force a Describe later, too.
+ */
+ flags = QueryExecutor.QUERY_BOTH_ROWS_AND_STATUS;
} else {
// If a batch hasn't specified that it wants generated keys, using the appropriate
// Connection.createStatement(...) interfaces, disallow any result set.
[parent: 24b45e426a90]