Method handleFetchCursorExtended

server/conn_cursor.go:443–522

handleFetchCursorExtended handles FETCH in the Extended Query protocol.

(p *portal)

	if !ok {
		c.sendError("ERROR", "34000", fmt.Sprintf("cursor %q does not exist", p.stmt.cursorName))
		return
	}

	// Open cursor on first FETCH
	if cursor.rows == nil {
		if err := c.openCursor(cursor); err != nil {
			if c.isCallerCancellation(err) {
				c.sendError("ERROR", "57014", "canceling statement due to user request")
			} else {
				c.sendError("ERROR", "42000", err.Error())
			}
			c.setTxError()
			return
		}
	}

	howMany := p.stmt.fetchCount

	// MOVE: advance position without returning rows (mirrors the
	// simple-protocol path in handleFetchCursor).
	if p.stmt.cursorIsMove {
		moveCount := int64(0)
		for moveCount < howMany && cursor.rows.Next() {
			// Read the row to advance position, but don't send it
			values := make([]interface{}, len(cursor.cols))
			valuePtrs := make([]interface{}, len(cursor.cols))
			for i := range values {
				valuePtrs[i] = &values[i]
			}
			_ = cursor.rows.Scan(valuePtrs...)
			moveCount++
		}
		_ = wire.WriteCommandComplete(c.writer, fmt.Sprintf("MOVE %d", moveCount))
		return
	}

	// Send RowDescription if Describe wasn't already called
	if !p.described && len(cursor.cols) > 0 {
		if err := c.sendRowDescriptionWithFormats(cursor.cols, cursor.colTypes, p.resultFormats); err != nil {
			return
		}
	}

	// Stream rows
	rowCount := int64(0)
	for rowCount < howMany && cursor.rows.Next() {
		values := make([]interface{}, len(cursor.cols))
		valuePtrs := make([]interface{}, len(cursor.cols))
		for i := range values {
			valuePtrs[i] = &values[i]
		}

		if err := cursor.rows.Scan(valuePtrs...); err != nil {
			c.sendError("ERROR", "42000", err.Error())
			c.setTxError()
			return
		}