Pipelined batch execution of addExec/addQuery operations against the PostgreSQL extended-query protocol. Includes both the single-Sync execute variant and the per-op Sync executeIsolated (error-isolated) variant.
Internal module: not part of the public API. Import the pg_client hub instead.
Types
IsolatedPipelineResults = object results*: seq[PipelineResult] errors*: seq[ref CatchableError] ## errors[i] is nil if ops[i] succeeded
- Results from executeIsolated: per-op error isolation via per-query SYNC.
Pipeline = ref object autoReset*: bool ## When true, `execute`/`executeIsolated` call `reset()` in a `finally` ## block so the Pipeline can be safely reused without leaking state from ## the previous run. Default: false (backward-compatible). ## ## **Warning:** when false, a second `execute`/`executeIsolated` without ## an intervening `reset()` re-sends the same queued ops. Non-idempotent ## commands (INSERT, UPDATE, …) then run twice. Prefer `true` for ## reusable pipelines, or call `reset()` before building a new batch.
- Batch of queries/execs sent through the PostgreSQL pipeline protocol.
PipelineOp = object
PipelineOpKind = enum pokExec, pokQuery
PipelineResult = object case kind*: PipelineResultKind of prkExec: commandResult*: CommandResult of prkQuery: queryResult*: QueryResult
- Result of a single operation within a pipeline.
PipelineResultKind = enum prkExec, prkQuery
- Discriminator for pipeline result variants.
Procs
proc addExec(p: Pipeline; sql: string; params: openArray[PgParamInline]) {. ...raises: [PgTypeError, PgMessageTooLargeError, PgProtocolError], tags: [], forbids: [].}
- Add an exec operation using the heap-alloc-free PgParamInline path.
proc addQuery(p: Pipeline; sql: string; params: openArray[PgParamInline]; resultFormat: ResultFormat = rfAuto) {. ...raises: [PgTypeError, PgMessageTooLargeError, PgProtocolError], tags: [], forbids: [].}
- Add a query operation using the heap-alloc-free PgParamInline path. Under-charges a cache hit's result formats as the typed overload does.
proc addQuery(p: Pipeline; sql: string; params: seq[PgParam] = @[]; resultFormat: ResultFormat = rfAuto) {. ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
- Add a query operation to the pipeline with typed parameters. The Bind pre-flight under-charges a cache hit's replayed result formats: the cache is only consulted in buildSendPhase, long after this add.
proc execute(p: Pipeline; timeout: Duration = ZeroDuration): Future[ seq[PipelineResult]] {....stackTrace: false, raises: [Exception, ValueError, CatchableError], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
-
Execute all queued pipeline operations in a single round trip. The batch shares one Sync, so it runs as one implicit transaction: when an op fails, the backend skips the ops after it, rolls back the ones before it, and the first PgQueryError is raised. Inside an explicit transaction (an earlier BEGIN), that transaction is aborted instead and needs a ROLLBACK. Use executeIsolated for per-op outcomes. On timeout, the connection is retired (csClosed) unless the wire had settled (asyncdispatch always retires: the timed-out op stays on the socket). When p.autoReset is true, the pipeline is reset on exit (including on raise) so it can be safely reused.
When p.autoReset is false (the default), queued ops are left in place. A second call without reset() re-sends them and can duplicate non-idempotent side effects — prefer autoReset = true for reuse.
proc executeIsolated(p: Pipeline; timeout: Duration = ZeroDuration): Future[ IsolatedPipelineResults] {....stackTrace: false, raises: [Exception, ValueError, CatchableError], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
-
Execute all queued pipeline operations with per-query error isolation. Each operation gets its own SYNC message, so a failed operation does not abort subsequent ones. Returns results and per-op errors. errors[i] only ever holds the server's PgQueryError for op i; a transport failure, timeout or cancellation raises instead, and the results are lost. A failed op's results[i] holds no result: an exec's stays default, but a query's may already carry the column descriptions and any rows received before the error. Inside an explicit transaction, the first failure aborts it, so every later op fails too (25P02). On timeout, the connection is retired (csClosed) unless the wire had settled (asyncdispatch always retires: the timed-out op stays on the socket). When p.autoReset is true, the pipeline is reset on exit (including on raise) so it can be safely reused.
When p.autoReset is false (the default), queued ops are left in place. A second call without reset() re-sends them and can duplicate non-idempotent side effects — prefer autoReset = true for reuse.
proc newPipeline(conn: PgConnection; autoReset: bool = false): Pipeline {. ...raises: [], tags: [], forbids: [].}
-
Create a new pipeline for batching multiple operations into a single round trip. When autoReset is true, the pipeline's queued ops and inline buffers are cleared automatically after each execute/executeIsolated call, making it safe to reuse the same Pipeline instance.
When autoReset is false (the default), queued ops remain after execute/executeIsolated. Calling either again without reset() re-sends those ops and can duplicate non-idempotent side effects. Prefer autoReset = true when the same Pipeline will be reused.
proc reset(p: Pipeline) {....raises: [], tags: [], forbids: [].}
-
Clear all queued ops and inline SoA buffers. Safe to call at any time, including while the pipeline is empty. Does not affect the underlying connection or its statement cache. When p.autoReset is true, execute/executeIsolated call this automatically (including on raise), so manual calls are only needed when autoReset is false.
Without autoReset, call this before building a new batch: a second execute/executeIsolated on uncleared ops re-sends them and can duplicate non-idempotent side effects.