Method handle_fetch

nodedb/src/control/server/pgwire/handler/cursor_cmds.rs:15–55

Handle FETCH [ALL | FORWARD n | BACKWARD n | n] FROM cursor_name.

(
        &self,
        addr: &std::net::SocketAddr,
        sql: &str,
        upper: &str,
    )

impl NodeDbPgHandler {
    /// Handle FETCH [ALL | FORWARD n | BACKWARD n | n] FROM cursor_name.
    pub(super) fn handle_fetch(
        &self,
        addr: &std::net::SocketAddr,
        sql: &str,
        upper: &str,
    ) -> PgWireResult<Vec<Response>> {
        let parts: Vec<&str> = sql.split_whitespace().collect();

        // Parse fetch direction and count.
        let (direction, count, cursor_name) = parse_fetch(upper, &parts)?;

        let rows = match direction {
            FetchDirection::Forward => {
                let (rows, _) = self
                    .sessions
                    .fetch_cursor(addr, &cursor_name, count)
                    .map_err(|e| cursor_error(&e.to_string()))?;
                rows
            }
            FetchDirection::All => self
                .sessions
                .fetch_cursor_all(addr, &cursor_name)
                .map_err(|e| cursor_error(&e.to_string()))?,
            FetchDirection::Backward => self
                .sessions
                .fetch_cursor_backward(addr, &cursor_name, count)
                .map_err(|e| cursor_error(&e.to_string()))?,
        };

        let schema = Arc::new(vec![text_field("result")]);
        let mut encoded_rows = Vec::with_capacity(rows.len());
        for row_json in &rows {
            let mut encoder = DataRowEncoder::new(schema.clone());
            let _ = encoder.encode_field(row_json);
            encoded_rows.push(Ok(encoder.take_row()));
        }
        Ok(vec![Response::Query(QueryResponse::new(
            schema,
            futures::stream::iter(encoded_rows),
        ))])
    }

    /// Handle MOVE [FORWARD | BACKWARD] n IN cursor_name.
    pub(super) fn handle_move(