1 files changed, 44 insertions, 10 deletions

diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c
index e1768d5475d..52d41658c1f 100644
--- a/src/interfaces/libpq/fe-exec.c
+++ b/src/interfaces/libpq/fe-exec.c
@@ -81,6 +81,7 @@ static int	PQsendTypedCommand(PGconn *conn, char command, char type,
 							   const char *target);
 static int	check_field_number(const PGresult *res, int field_num);
 static void pqPipelineProcessQueue(PGconn *conn);
+static int	pqPipelineSyncInternal(PGconn *conn, bool immediate_flush);
 static int	pqPipelineFlush(PGconn *conn);
 
 
@@ -3224,25 +3225,48 @@ pqPipelineProcessQueue(PGconn *conn)
 /*
  * PQpipelineSync
  *		Send a Sync message as part of a pipeline, and flush to server
+ */
+int
+PQpipelineSync(PGconn *conn)
+{
+	return pqPipelineSyncInternal(conn, true);
+}
+
+/*
+ * PQsendPipelineSync
+ *		Send a Sync message as part of a pipeline, without flushing to server
+ */
+int
+PQsendPipelineSync(PGconn *conn)
+{
+	return pqPipelineSyncInternal(conn, false);
+}
+
+/*
+ * Workhorse function for PQpipelineSync and PQsendPipelineSync.
  *
  * It's legal to start submitting more commands in the pipeline immediately,
  * without waiting for the results of the current pipeline. There's no need to
  * end pipeline mode and start it again.
  *
- * If a command in a pipeline fails, every subsequent command up to and including
- * the result to the Sync message sent by PQpipelineSync gets set to
- * PGRES_PIPELINE_ABORTED state. If the whole pipeline is processed without
- * error, a PGresult with PGRES_PIPELINE_SYNC is produced.
+ * If a command in a pipeline fails, every subsequent command up to and
+ * including the result to the Sync message sent by pqPipelineSyncInternal
+ * gets set to PGRES_PIPELINE_ABORTED state. If the whole pipeline is
+ * processed without error, a PGresult with PGRES_PIPELINE_SYNC is produced.
  *
- * Queries can already have been sent before PQpipelineSync is called, but
- * PQpipelineSync needs to be called before retrieving command results.
+ * Queries can already have been sent before pqPipelineSyncInternal is called,
+ * but pqPipelineSyncInternal needs to be called before retrieving command
+ * results.
  *
  * The connection will remain in pipeline mode and unavailable for new
  * synchronous command execution functions until all results from the pipeline
  * are processed by the client.
+ *
+ * immediate_flush controls if the flush happens immediately after sending the
+ * Sync message or not.
  */
-int
-PQpipelineSync(PGconn *conn)
+static int
+pqPipelineSyncInternal(PGconn *conn, bool immediate_flush)
 {
 	PGcmdQueueEntry *entry;
 
@@ -3288,9 +3312,19 @@ PQpipelineSync(PGconn *conn)
 	/*
 	 * Give the data a push.  In nonblock mode, don't complain if we're unable
 	 * to send it all; PQgetResult() will do any additional flushing needed.
+	 * If immediate_flush is disabled, the data is pushed if we are past the
+	 * size threshold.
 	 */
-	if (PQflush(conn) < 0)
-		goto sendFailed;
+	if (immediate_flush)
+	{
+		if (pqFlush(conn) < 0)
+			goto sendFailed;
+	}
+	else
+	{
+		if (pqPipelineFlush(conn) < 0)
+			goto sendFailed;
+	}
 
 	/* OK, it's launched! */
 	pqAppendCmdQueueEntry(conn, entry);