impl Stream for QueryResult

[jkoudys]

In light of stable async/await, a lot of these collect/try_collect/map/etc. methods seem redundant if we could instead stream the async results.

e.g. this pattern is replicated with minor changes in several places:

    pub async fn for_each<F>(self, mut fun: F) -> Result<Self>
    where
        F: FnMut(Row),
    {
        if self.is_empty() {
            Ok(self)
        } else {
            let mut qr = self;
            loop {
                let (qr_, row) = qr.get_row().await?;
                qr = qr_;
                if let Some(row) = row {
                    fun(row);
                } else {
                    break Ok(qr);
                }
            }
        }
    }

Why not simply swap out the fun(row); and instead make that the poll on a Stream?
The big thing this would open up for me, is I could impl From<my::Row> (which would call the FromRow), and then let the From and standard collect() calls build my result sets.

As it's written, I'm stuck forced to build full Vecs of my entire QueryResult, when a stream would let me map, filter, zip, etc., and let type inference handle my result from collecting (could even make a HashMap or whatever). Would also make it a lot easier to return an Err on the very first TryFrom<my::Row> that fails, instead of building the full result with all errors like try_collect does now.

[jkoudys]

Ah I see you've implemented streams; just got thrown by the _async vs base lib.

[jkoudys]

The stream I found wasn't what I thought it was. Updated description.

[jkoudys]

Alternatively, perhaps make get_row public? I could roll my own stream easy enough on a newtype if I could get into that.

[jkoudys]

Got a working branch up: https://github.com/jkoudys/mysql_async/blob/master/src/queryable/query_result/mod.rs#L153

lmk if I'm on the right track and I'll clean up this match-hell and get it PR ready.

[blackbeam]

Hi. Thanks for your work.

Looks like there is no way to get back wrapped ConnectionLike out of returned impl Stream. Also i'm not sure that it'll properly handle multi-result set, i.e. get_row will return None on the first result set boundary.

[jkoudys]

Can I see some examples of a multi-result set? I was following the patterns already established in the for_each/reduce/map/etc., that appear to return as soon as one of the get_rows returns None:

                if let Some(row) = row {
                    acc = fun(acc, row);
                } else {
                    break Ok((qr, acc));
                }

Is it necessary that all these methods give access to a Self at the end? It'll have consumed all the rows by the time it reaches the end of its stream. Perhaps the method could simply be named something so you know it's consuming the QueryResult.

[blackbeam]

Can I see some examples of a multi-result set?

The following query should emit multi-result set, where second result set will be empty. Also note, that in general each result set will have it's own meta (i.e. last insert id and number of rows affected). It's also possible for stored procedures to emit multi-result sets.

SELECT 'FOO'; DO 1; SELECT 42;

Is it necessary that all these methods give access to a Self at the end?

It's kind of OK when connection is taken from pool, but if not, then you'll lose the connection and will have to reconnect.

Also i have plans to change the API to address blackbeam/rust-mysql-simple#199 (i want to adopt something similar to the tokio-postgres API and it's currently in progress for rust-mysql-simple), but I want to leave QueryResult as the source of most detailed information about the query.

[jkoudys]

If this doesn't align with the direction you're taking it, could we at least have get_row made pub? My method would be simple enough to create myself (even inside a newtype) in that case. Seems like a safe method to expose since it's the same QueryResult-first tuple I see everywhere else.

I could take a stab at writing something modelled after the Unfold struct, except give it some way to abort the Fut, and keep ownership of the QR that can be returned once the implemented Stream completes (ie Ready(None)). It probably should be in its own struct anyways, because then we could do nice things like implement size_hint.

[jkoudys]

I'm thinking through an approach where I could define a newtype on top of Unfold, which would also track an Option<QueryResult>. Then I'd impl Stream against it and write a poll_next around the poll_next of the interior Unwrap. The interior unwrap callback can then return an enum with the usual rx-style types (next, complete, and err), the outer poll_next can see if we've got a Next (so do nothing), or a Complete, in which case we can end the stream with a Poll::Ready(None), then let the RowSteam own the qr again.
I haven't even tried to compile yet, but the basic idea is:

pub struct RowStream<T, P> {
    unfold: Unfold,
    qr: Option<QueryResult<T, P>>,
}

enum RowPollResponse {
    Next(QueryResult<_, _>, Row),
    Err(QueryResult<_, _>, Err),
    Complete(QueryResult<_, _>),
}

impl RowStream<T, P> {
    fn new(qr: QueryResult<T, P>) -> QueryResult<T, P> {
        Self {
            unfold: stream::unfold(Some(self), |qr| {
                async move {
                    match q.get_row().await {
                        Ok((qr_, opt)) => match opt {
                            Some(row) => Some(RowPollResponse::Next(qr_, row)),
                            None => Some(RowPollResponse::Complete(qr_)),
                        },
                        Err(e) => Some(RowPollResponse::Err(q, Err(e))),
                    }
                }
            }),
            qr: None,
        }
    }
}

impl Stream for RowStream<T, P> {
    type Item = Result<Row>;
    // yes I know this is probably way off...
    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if let Poll::Ready(Some(rpr)) = self.unfold.poll_next(cx) {
            match rpr {
                RowPollResponse::Next(qr, row) => {
                    self.qr = qr;
                    Poll::Ready(Some(Ok(row)))
                }
                RowPollResponse::Err(qr, err) => {
                    self.qr = qr;
                    Poll::Ready(Some(Result::Err(err)))
                }
                RowPollResponse::Complete(qr) => {
                    self.qr = qr;
                    Poll::Ready(None)
                }
            }
        } else {
            Poll::Ready(None)
        }
    }
}

[blackbeam]

Fixed in v0.28.1