handleFetchCursor handles FETCH in the Simple Query protocol.
(query string, stmt *pg_query.FetchStmt)
| 283 | // Validate direction |
| 284 | if !isFetchForwardOnly(stmt.Direction) { |
| 285 | c.sendError("ERROR", "0A000", "cursor can only scan forward") |
| 286 | c.logQuery(start, query, query, "FETCH", 0, 0, "0A000", "cursor can only scan forward", "simple") |
| 287 | _ = c.writeReadyForQuery(c.txStatus) |
| 288 | _ = c.flushWriter() |
| 289 | return nil |
| 290 | } |
| 291 | |
| 292 | cursor, ok := c.cursors[stmt.Portalname] |
| 293 | if !ok { |
| 294 | errMsg := fmt.Sprintf("cursor %q does not exist", stmt.Portalname) |
| 295 | c.sendError("ERROR", "34000", errMsg) |
| 296 | c.logQuery(start, query, query, "FETCH", 0, 0, "34000", errMsg, "simple") |
| 297 | _ = c.writeReadyForQuery(c.txStatus) |
| 298 | _ = c.flushWriter() |
| 299 | return nil |
| 300 | } |
| 301 | |
| 302 | // Open cursor on first FETCH |
| 303 | if cursor.rows == nil { |
| 304 | if err := c.openCursor(cursor); err != nil { |
| 305 | errCode := "42000" |
| 306 | errMsg := err.Error() |
| 307 | if c.isCallerCancellation(err) { |
| 308 | errCode = "57014" |
| 309 | errMsg = "canceling statement due to user request" |
| 310 | } |
| 311 | c.sendError("ERROR", errCode, errMsg) |
| 312 | c.setTxError() |
| 313 | c.logQuery(start, query, query, "FETCH", 0, 0, errCode, errMsg, "simple") |
| 314 | _ = c.writeReadyForQuery(c.txStatus) |
| 315 | _ = c.flushWriter() |
| 316 | return nil |
| 317 | } |
| 318 | } |
| 319 | |
| 320 | // Determine how many rows to fetch (pg_query sets HowMany=MaxInt64 for FETCH ALL) |
| 321 | howMany := stmt.HowMany |
| 322 | if howMany < 0 { |
| 323 | c.sendError("ERROR", "0A000", "cursor can only scan forward") |
| 324 | c.logQuery(start, query, query, "FETCH", 0, 0, "0A000", "cursor can only scan forward", "simple") |
| 325 | _ = c.writeReadyForQuery(c.txStatus) |
| 326 | _ = c.flushWriter() |
| 327 | return nil |
| 328 | } |
| 329 | |
| 330 | // MOVE: advance position without returning rows |
| 331 | if stmt.Ismove { |
| 332 | moveCount := int64(0) |
| 333 | for moveCount < howMany && cursor.rows.Next() { |
| 334 | // Read the row to advance position, but don't send it |
| 335 | values := make([]interface{}, len(cursor.cols)) |
| 336 | valuePtrs := make([]interface{}, len(cursor.cols)) |
| 337 | for i := range values { |
| 338 | valuePtrs[i] = &values[i] |
| 339 | } |
| 340 | _ = cursor.rows.Scan(valuePtrs...) |
| 341 | moveCount++ |
| 342 | } |