executeSingleStatement transpiles and executes a single SQL statement, sending results to the client. Does NOT send ReadyForQuery (the caller is responsible for that). Returns (true, nil) if an error was sent to the client (so the caller can stop processing a batch), or (false, err) for fatal connec
(query string)
| 370 | c.setTxError() |
| 371 | _ = c.writeReadyForQuery(c.txStatus) |
| 372 | _ = c.flushWriter() |
| 373 | return 0, errCode, errMsg, nil |
| 374 | } |
| 375 | |
| 376 | c.updateTxStatus(cmdType) |
| 377 | tag := buildCommandTagFromRowCount(cmdType, int64(rowCount)) |
| 378 | _ = c.writeCommandComplete(tag) |
| 379 | _ = c.writeReadyForQuery(c.txStatus) |
| 380 | _ = c.flushWriter() |
| 381 | return int64(rowCount), "", "", nil |
| 382 | } |
| 383 | |
| 384 | // selectStream is the outcome of streaming one result set to the client. |
| 385 | // Exactly one of scanErr / rowsErr / writeErr is non-nil (or none of them, on |
| 386 | // a clean stream): the three failure modes are kept apart because the caller |
| 387 | // answers them differently — a Scan failure is always 42000 with rowCount |
| 388 | // suppressed, a terminal RowSet error maps a caller cancellation to 57014, and |
| 389 | // a client-write failure is a connection error with no ErrorResponse at all. |
| 390 | type selectStream struct { |
| 391 | rowsSent int |
| 392 | scanErr error |
| 393 | rowsErr error |
| 394 | // writeErr is the RAW pgwire write error (returned to the caller |
| 395 | // unwrapped, as before); writeStage names where it happened so the caller |
| 396 | // can reproduce the exact log wording. |
| 397 | writeErr error |
| 398 | writeStage string |
| 399 | // limitReached is set when maxRows stopped the stream with the rowset not |
| 400 | // (yet) known to be exhausted — the extended protocol suspends the portal |
| 401 | // (PortalSuspended) and a later Execute resumes from the same rowset. All |
| 402 | // three errors are nil when it is set, and rows.Err() has deliberately |
| 403 | // NOT been consulted: the rowset is still live. |
| 404 | limitReached bool |
| 405 | } |
| 406 | |
| 407 | // streamSelectRows sends (optionally) the RowDescription and then every DataRow |
| 408 | // of rows. Extracted from executeSelectQuery so the exploratory tier can retry |
| 409 | // a zero-row OOM stream on the escalated worker WITHOUT resending |
| 410 | // RowDescription — pass sendRowDesc=false on such a retry. |
| 411 | // |
| 412 | // formats are the Bind result-format codes (nil = all text, the simple-query |
| 413 | // case). maxRows > 0 caps the DataRows sent, for the extended protocol's |
| 414 | // Execute row limit (portal suspension): the limit is checked BEFORE advancing |
| 415 | // the rowset, so a suspended stream never consumes — and loses — the first row |
| 416 | // of the next page. Matching PostgreSQL, the cap suspends even when the rowset |
| 417 | // happens to be exactly exhausted: knowing would mean consuming the next row, |
| 418 | // so the client's follow-up Execute gets 0 rows and the completion instead. |
| 419 | func (c *clientConn) streamSelectRows(rows RowSet, cols []string, colTypes []ColumnTyper, typeOIDs []int32, sendRowDesc bool, formats []int16, maxRows int32) selectStream { |
| 420 | if sendRowDesc { |
| 421 | if err := c.sendRowDescription(cols, colTypes); err != nil { |
| 422 | return selectStream{writeErr: err, writeStage: "sending row description"} |
| 423 | } |
| 424 | } |
| 425 | |
| 426 | var out selectStream |
| 427 | for { |
| 428 | if maxRows > 0 && int32(out.rowsSent) >= maxRows { |
| 429 | out.limitReached = true |