Support query pipelineing in a single session

[lucianojoublanc-da]

This follows from a discussion on discord.

When submitting multiple queries concurrently, in the same session, you'll get a portal_n does not exist. Here's a minimal example from repo:

  def run(args: List[String]): IO[ExitCode] =
    session.use { s =>
      import cats.syntax.all._
      (makeQuery(s), makeQuery(s)).parMapN{ case _ => ExitCode.Success }
    }

  def makeQuery(session: Session[IO]) =
    for {
      p <- session.prepare(sql"select current_date".query(date))
      d <- p.unique(Void)
    } yield ()

2023-06-26 15:59:37.731 UTC [154] DEBUG:  parse statement_1: select current_date
2023-06-26 15:59:37.731 UTC [154] DEBUG:  StartTransaction(1) name: unnamed; blockState: DEFAULT; state: INPROGRESS, xid/subid/cid: 0/1/0
2023-06-26 15:59:37.734 UTC [154] DEBUG:  parse statement_2: select current_date
2023-06-26 15:59:37.747 UTC [154] DEBUG:  bind portal_3 to statement_1
2023-06-26 15:59:37.748 UTC [154] DEBUG:  bind portal_4 to statement_2
2023-06-26 15:59:37.756 UTC [154] DEBUG:  CommitTransaction(1) name: unnamed; blockState: STARTED; state: INPROGRESS, xid/subid/cid: 0/1/0
2023-06-26 15:59:37.760 UTC [154] ERROR:  portal "portal_3" does not exist

This can be gotten around by increasing the connection pool size, or by using a separate session for each makeQuery call.

PostgreSQL seems to say that this should work however; it's documented here. But perhaps I've misunderstood this.

[lucianojoublanc-da]

Wanted to add two more points from the doc.

First is that:

If successfully created, a named portal object lasts till the end of the current transaction ...

So this is why I mentioned the Start/CommitTransaction messages in the log above.

And second,

If the client has not issued an explicit BEGIN, then each Sync ordinarily causes an implicit COMMIT if the preceding step(s) succeeded, or an implicit ROLLBACK if they failed.

Looking at the trace of the minimized example, you can see this also:

 → Bind(portal_3,statement_1,List())
 → Flush
 ← BindComplete
 → Bind(portal_4,statement_2,List())
 → Flush
 ← BindComplete
 → Execute(portal_4,2)
 → Flush
 ← RowData(List(Some(2023-06-27)))
 ← CommandComplete(Select(1))
 → Sync                               <---
 ← ReadyForQuery(Idle)
 → Execute(portal_3,2)
 → Flush
 ← ErrorResponse(F -> postgres.c, M -> portal "portal_3" does not exist, V -> ERROR, L -> 1917, C -> 34000, R -> exec_execute_message, S -> ERROR

Perhaps, it's a matter of being clever about the sync, but this might also impact error recovery.