Subject: Re: Asynchronous Append on postgres_fdw nodes.
To: Etsuro Fujita <etsuro.fujita@gmail.com>,
 Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Cc: Justin Pryzby <pryzby@telsasoft.com>, "movead.li" <movead.li@highgo.ca>,
 PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
References: 
 <CAPmGK153oorYtTpW_-aZrjH-iecHbykX7qbxX_5630ZK8nqVHg@mail.gmail.com>
 <20201002.153925.2111528346612886434.horikyota.ntt@gmail.com>
 <CAPmGK16YXCADSwsFLSxqTBBLbt3E_=iigKTtjS=dqu+8K8DWCw@mail.gmail.com>
 <20201005.132959.1424481454550541470.horikyota.ntt@gmail.com>
 <CAPmGK17Ura3NUXYP98W4jz0VT9BAyVFhNRqZyFV-_mqFDB0j7Q@mail.gmail.com>
From: "Andrey V. Lepikhov" <a.lepikhov@postgrespro.ru>
Organization: Postgres Professional
Message-ID: <80e914bd-9677-114f-d96f-aa972dc6ad0d@postgrespro.ru>
Date: Thu, 8 Oct 2020 14:39:47 +0500
User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:68.0) Gecko/20100101
 Thunderbird/68.10.0
MIME-Version: 1.0
In-Reply-To: 
 <CAPmGK17Ura3NUXYP98W4jz0VT9BAyVFhNRqZyFV-_mqFDB0j7Q@mail.gmail.com>
Content-Type: multipart/mixed;
 boundary="------------0E67542495DA2ED0A4B926D9"
Content-Language: en-US
Precedence: bulk

This is a multi-part message in MIME format.
--------------0E67542495DA2ED0A4B926D9
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 7bit

On 10/5/20 11:35 AM, Etsuro Fujita wrote:
Hi,
I found a small problem. If we have a mix of async and sync subplans 
when we catch an assertion on a busy connection. Just for example:

PLAN
====
Nested Loop  (cost=100.00..174316.95 rows=975 width=8) (actual 
time=5.191..9.262 rows=9 loops=1)
    Join Filter: (frgn.a = l.a)
    Rows Removed by Join Filter: 8991
    ->  Append  (cost=0.00..257.20 rows=11890 width=4) (actual 
time=0.419..2.773 rows=1000 loops=1)
          Async subplans: 4
          ->  Async Foreign Scan on f_1 l_2  (cost=100.00..197.75 
rows=2925 width=4) (actual time=0.381..0.585 rows=211 loops=1)
          ->  Async Foreign Scan on f_2 l_3  (cost=100.00..197.75 
rows=2925 width=4) (actual time=0.005..0.206 rows=195 loops=1)
          ->  Async Foreign Scan on f_3 l_4  (cost=100.00..197.75 
rows=2925 width=4) (actual time=0.003..0.282 rows=187 loops=1)
          ->  Async Foreign Scan on f_4 l_5  (cost=100.00..197.75 
rows=2925 width=4) (actual time=0.003..0.316 rows=217 loops=1)
          ->  Seq Scan on l_0 l_1  (cost=0.00..2.90 rows=190 width=4) 
(actual time=0.017..0.057 rows=190 loops=1)
    ->  Materialize  (cost=100.00..170.94 rows=975 width=4) (actual 
time=0.001..0.002 rows=9 loops=1000)
          ->  Foreign Scan on frgn  (cost=100.00..166.06 rows=975 
width=4) (actual time=0.766..0.768 rows=9 loops=1)

Reproduction script 'test1.sql' see in attachment. Here I force the 
problem reproduction with setting enable_hashjoin and enable_mergejoin 
to off.

'asyncmix.patch' contains my solution to this problem.

-- 
regards,
Andrey Lepikhov
Postgres Professional

--------------0E67542495DA2ED0A4B926D9
Content-Type: application/sql;
 name="test1.sql"
Content-Transfer-Encoding: base64
Content-Disposition: attachment;
 filename="test1.sql"

RFJPUCBFWFRFTlNJT04gSUYgRVhJU1RTIHBvc3RncmVzX2ZkdyBDQVNDQURFOwpEUk9QIFRB
QkxFIElGIEVYSVNUUyBsLCBsXzEsIGxfMiwgbF8zLCBsXzQgQ0FTQ0FERTsKCkNSRUFURSBF
WFRFTlNJT04gcG9zdGdyZXNfZmR3OwpDUkVBVEUgU0VSVkVSIGxvb3BiYWNrIEZPUkVJR04g
REFUQSBXUkFQUEVSIHBvc3RncmVzX2ZkdzsKQ1JFQVRFIFVTRVIgTUFQUElORyBGT1IgUFVC
TElDIFNFUlZFUiBsb29wYmFjazsKCkNSRUFURSBUQUJMRSBsKGEgaW50KSBQQVJUSVRJT04g
QlkgSEFTSCAoYSk7CkNSRUFURSBUQUJMRSBsXzEgKGEgaW50KTsKQ1JFQVRFIFRBQkxFIGxf
MiAoYSBpbnQpOwpDUkVBVEUgVEFCTEUgbF8zIChhIGludCk7CkNSRUFURSBUQUJMRSBsXzQg
KGEgaW50KTsKQ1JFQVRFIFRBQkxFIGxvY2FsIChhIGludCk7CkNSRUFURSBUQUJMRSBsXzAg
UEFSVElUSU9OIE9GIGwgRk9SIFZBTFVFUyBXSVRIIChtb2R1bHVzIDUsIHJlbWFpbmRlciAw
KTsKQ1JFQVRFIEZPUkVJR04gVEFCTEUgZl8xIFBBUlRJVElPTiBPRiBsIEZPUiBWQUxVRVMg
V0lUSCAobW9kdWx1cyA1LCByZW1haW5kZXIgMSkgU0VSVkVSIGxvb3BiYWNrIE9QVElPTlMg
KHRhYmxlX25hbWUgJ2xfMScpOwpDUkVBVEUgRk9SRUlHTiBUQUJMRSBmXzIgUEFSVElUSU9O
IE9GIGwgRk9SIFZBTFVFUyBXSVRIIChtb2R1bHVzIDUsIHJlbWFpbmRlciAyKSBTRVJWRVIg
bG9vcGJhY2sgT1BUSU9OUyAodGFibGVfbmFtZSAnbF8yJyk7CkNSRUFURSBGT1JFSUdOIFRB
QkxFIGZfMyBQQVJUSVRJT04gT0YgbCBGT1IgVkFMVUVTIFdJVEggKG1vZHVsdXMgNSwgcmVt
YWluZGVyIDMpIFNFUlZFUiBsb29wYmFjayBPUFRJT05TICh0YWJsZV9uYW1lICdsXzMnKTsK
Q1JFQVRFIEZPUkVJR04gVEFCTEUgZl80IFBBUlRJVElPTiBPRiBsIEZPUiBWQUxVRVMgV0lU
SCAobW9kdWx1cyA1LCByZW1haW5kZXIgNCkgU0VSVkVSIGxvb3BiYWNrIE9QVElPTlMgKHRh
YmxlX25hbWUgJ2xfNCcpOwoKQ1JFQVRFIEZPUkVJR04gVEFCTEUgZnJnbiAoYSBpbnQpIFNF
UlZFUiBsb29wYmFjayBPUFRJT05TICh0YWJsZV9uYW1lICdsb2NhbCcpOwoKSU5TRVJUIElO
VE8gbCAoYSkgU0VMRUNUICogRlJPTSBnZW5lcmF0ZV9zZXJpZXMoMSwxMDAwKTsKSU5TRVJU
IElOVE8gZnJnbiAoYSkgU0VMRUNUICogRlJPTSBnZW5lcmF0ZV9zZXJpZXMoMSwxMCk7ClZB
Q1VVTSBBTkFMWVpFOwoKU0VUIGVuYWJsZV9oYXNoam9pbiA9ICdvZmYnOwpTRVQgZW5hYmxl
X21lcmdlam9pbiA9ICdvZmYnOwpleHBsYWluIGFuYWx5emUgU0VMRUNUICogRlJPTSBmcmdu
LGwgV0hFUkUgZnJnbi5hIDwgMTAgQU5EIGZyZ24uYSA9IGwuYTsK
--------------0E67542495DA2ED0A4B926D9
Content-Type: text/x-patch; charset=UTF-8;
 name="asyncmix.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
 filename="asyncmix.patch"

diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 14824368cc..613d406982 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -455,7 +455,7 @@ static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel,
 									  void *arg);
 static void create_cursor(ForeignScanState *node);
 static void request_more_data(ForeignScanState *node);
-static void fetch_received_data(ForeignScanState *node);
+static void fetch_received_data(ForeignScanState *node, bool vacateconn);
 static void vacate_connection(PgFdwState *fdwconn, bool clear_queue);
 static void close_cursor(PGconn *conn, unsigned int cursor_number);
 static PgFdwModifyState *create_foreign_modify(EState *estate,
@@ -1706,15 +1706,19 @@ postgresIterateForeignScan(ForeignScanState *node)
 		{
 			/*
 			 * finish the running query before sending the next command for
-			 * this node
+			 * this node.
+			 * When the plan contains both asynchronous subplans and non-async
+			 * subplans backend could request more data in async mode and want to
+			 * get data in sync mode by the same connection. Here it must wait
+			 * for async data before request another.
 			 */
-			if (!fsstate->s.commonstate->busy)
-				vacate_connection((PgFdwState *)fsstate, false);
+			if (fsstate->s.commonstate->busy)
+				vacate_connection(&fsstate->s, false);
 
 			request_more_data(node);
 
 			/* Fetch the result immediately. */
-			fetch_received_data(node);
+			fetch_received_data(node, false);
 		}
 		else if (!fsstate->s.commonstate->busy)
 		{
@@ -1749,7 +1753,7 @@ postgresIterateForeignScan(ForeignScanState *node)
 			/* fetch the leader's data and enqueue it for the next request */
 			if (available)
 			{
-				fetch_received_data(leader);
+				fetch_received_data(leader, false);
 				add_async_waiter(leader);
 			}
 		}
@@ -3729,7 +3733,7 @@ request_more_data(ForeignScanState *node)
  * Fetches received data and automatically send requests of the next waiter.
  */
 static void
-fetch_received_data(ForeignScanState *node)
+fetch_received_data(ForeignScanState *node, bool vacateconn)
 {
 	PgFdwScanState *fsstate = GetPgFdwScanState(node);
 	PGresult   *volatile res = NULL;
@@ -3817,7 +3821,8 @@ fetch_received_data(ForeignScanState *node)
 	waiter = move_to_next_waiter(node);
 
 	/* send the next request if any */
-	if (waiter)
+	if (waiter && (!vacateconn ||
+		GetPgFdwScanState(node)->s.conn != GetPgFdwScanState(waiter)->s.conn))
 		request_more_data(waiter);
 
 	MemoryContextSwitchTo(oldcontext);
@@ -3843,7 +3848,7 @@ vacate_connection(PgFdwState *fdwstate, bool clear_queue)
 	 * query
 	 */
 	leader = commonstate->leader;
-	fetch_received_data(leader);
+	fetch_received_data(leader, true);
 
 	/* let the first waiter be the next leader of this connection */
 	move_to_next_waiter(leader);

--------------0E67542495DA2ED0A4B926D9--