Re: Bypassing cursors in postgres_fdw to enable parallel plans

On Thu, 27 Aug 2026 at 17:52, Robert Haas <robertmhaas(at)gmail(dot)com> wrote:

> On Wed, Aug 26, 2026 at 5:45 AM Rafia Sabih <rafia(dot)pghackers(at)gmail(dot)com>
> wrote:
> > CFBot informed me that the patches need a rebase. Please find the
> attached files for the rebased patches.
>
> Thanks.
>
> + if (is_active_scan(fsstate) &&
> + !pgfdw_cancel_query(fsstate->conn, fsstate->conn_state))
> + ereport(ERROR,
> + errcode(ERRCODE_CONNECTION_FAILURE),
> + errmsg("could not cancel query"));
>
> You have two places where you do this, one in
> postgresReScanForeignScan and one in postgresEndForeignScan. It's
> unsafe in both places. Think about what happens afterwards: the remote
> transaction is now in an aborted state, but the local transaction is
> not. If the local transaction tries to do something afterwards that
> uses the remote connection, it will fail. I think this means that the
> re-scan path must be untested: if you ended the remote scan early and
> then actually rescanned, the second scan would fail due to this
> problem. When it works, it's because the remote scan didn't actually
> end early: it managed to complete before the cancel request arrived.
>
> I think what needs to happen here is you have to read all the results
> from the foreign side, no matter how many of them there are, and just
> discard them all. That's pretty unappealing from a performance point
> of view and will cause this feature to lose as compared with the
> current model in a bunch of situations, but those situations are maybe
> not cases where this feature would make any sense in the first place.
> Also, it's not clear what alternative we really have. I suppose we
> could try wrapping an extra level of subtransaction around the remote
> side, but that probably adds quite a bit of complexity and I'm not
> convinced that it would work out well overall.
>
> You are right. To resolve this, I was trying to
reuse fetch_streaming_result, but realised that easier would be to have a
wrapper function called discard_stream_result(), which just reads (via
calls to fetch_stream_result) and discards the tuples.

> + /*
> + * fetch_stream_result() hands back each PGRES_TUPLES_CHUNK result in
> turn
> + * and, once the wire protocol is fully drained, clears active_scan
> and
> + * returns NULL. Keep calling it until then, saving every chunk we
> get
> + * along the way.
> + */
> + for (;;)
> + {
>
> Something we need to think about is that failures are possible inside
> this loop. CHECK_FOR_INTERRUPTS() can fail.
> make_tuple_from_result_row() can fail, e.g. because the rows we're
> reading don't match the expected data type. Maybe
> tuplestore_begin_heap() or tuplestore_puttuple() can fail for lack of
> memory. Failures are certainly possible. What will then happen is that
> the rows returned by the last call to fetch_stream_result() for which
> we have not yet called tuplestore_puttuple() are silently lost. You
> might think that this isn't really a problem, because surely if we've
> had an error then the contents of the tuplestore are irrelevant
> anyway, but that's actually not true. This function can be called from
> a subtransaction of the one that started the currently-running remote
> query, and so after a failure here only the subtransaction will abort,
> and the outer transaction level can then still try to use the
> tuplestore contents, which are now silently incorrect.
>
> I think it is OK if we don't handle this situation 100% perfectly.
> There is no perfect answer, and it's a weird, rare case. However,
> giving silent wrong answers is not OK. I think we need to structure
> this so that we *know* we've lost data. I think what we need to do is
> set a flag just before calling fetch_stream_result() and clear it
> after we've added all tuples to the tuplestore. If we later see that
> flag set, we know that the tuplestore has potentially lost data and
> that we therefore need to error out.
>
To handle this scenario, I added a flag called tuplestore_lost_rows(feel
free to suggest a better name) and we set it at the beginning and once
everything goes on well it is set back to false. Later it is checked in
fetch_from_tuplestore before starting the fetch, and error is reported if
it is found.

>
> + /*
> + * fetch_stream_result() unconditionally set eof_reached once it
> drained
> + * the wire. That's wrong if we saved any rows above: the scan still
> has
> + * pending data to read from the tuplestore, so it must not be
> treated as
> + * EOF yet. fetch_from_tuplestore() will set eof_reached again once
> the
> + * tuplestore itself is drained.
> + */
> + if (active_fsstate->tuplestore != NULL)
> + active_fsstate->eof_reached = false;
>
> Resetting the flag here means that if you error out before you get to
> this point, the flag is in the wrong state. You're going to need to go
> through this whole code path really carefully and make sure that if
> you abort out at any point, the state at that point is what you want
> it to be. Even if it doesn't end up mattering in practice, it's a good
> idea to avoid setting wrong values and then fixing them up later,
> because future code changes can make things that don't matter now
> start mattering.
>
> Good point, in the attached version I made the changes so that
fetch_stream_result doesn't change the eof_reached flag, rather the callers
take care of it. So now, in fetch_more_data when we encounter NULL res,
which means it really is EOF, then only this flag is set. And
in drain_other_active_scan(), we set it only when we find that tuplestore
is NULL, that means truly there is nothing left there.

> On another topic, pgfdw_xact_callback and pgfdw_subxact_callback don't
> drain active scans before issuing SQL commands. Most of those cases
> appear to be OK anyway. For example, if we're aborting, we call
> pgfdw_abort_cleanup() which will cancel any in-progress query and
> discard the results, and if we're committing, there shouldn't be any
> scans still in progress, though maybe an Assert() to verify that would
> be a good idea. But at *subtransaction* commit, there can still be
> scans in progress, and in that case, issuing a RELEASE SAVEPOINT
> command without draining the active scan first will not work out.
>
> In a few places, you still have drain_other_active_scan() calls before
> pgfdw_exec_query(), which isn't necessary now that pgfdw_exec_query()
> calls that function internally.
>
> Fixed

> In fetch_from_tuplestore, the call to tuplestore_gettupleslot()
> doesn't need to pass copy = true, because ExecFetchSlotHeapTuple()
> also copies.
>
> Also fixed.

There is a new test case added to cover the case when the new flag
tuplestore_lost_rows is set.
There are some other cosmetic additions like a new helper
function reset_batch_state() to reset the state so that fresh fetch can
proceed. Since there were a couple of places in postgresReScanForeignScan
where the same block of code was duplicated. However, a couple of times is
not too many times to have a helper function. So I have mixed feelings
about it, let me know if it is better to keep the duplicated code instead
of this helper function.
Removed set_streaming_fetch as it was too thin and felt unnecessary.
I also added things in the sgml file to include the case of early
termination of scan like LIMIT, rescan, etc.

Please find the reworked patchset in the attachment.
Looking forward to your feedback on this one.

--
Regards,
Rafia Sabih
CYBERTEC PostgreSQL International GmbH