integration with AsyncSeq

[fwaris]

Requesting support for AsyncSeq for composable processing of large datasets.

I could not make asyncdb {...} work effectively to produce sequences.

[jacentino]

Hi,

Do you think about something like asyncdbseq?
I'll think over it.

For now, you have ResultStream to work with large results.
You can even transform them into asyncSeq, but I'm not sure if it can be easily used inside asyncdb.

[fwaris]

I am using result stream now. However, I think converting to asyncSeq would make the stream processing composable - you can do filtering, buffering, stateful sequential processing, etc.

I tried to make asyncSeq from the result stream but kept running into something getting disposed the code errored out.

specifically I want to download large dataset and stream that into a compressed dataset (from Microsoft.ML.Data.IDataView) without needing intermediate storage.

[jacentino]

I added AsyncResultStream to work with async sequences.
You can also use AsyncSeq in for statement of asyncdb computation expression.

Most calculations involving asyncSeq finish with some async value. To handle it properly, you should lift it to AsyncDb using AsyncDb.fromAsync function. It shouldn't leak outside asyncdb, i.e. you shouldn't return such async value from computation expression.

E.g. for query function defined as follows:

let getPostsByTags: string list -> AsyncDb<Post AsyncResultStream> = 
    sql "select id, blogId, p.name, title, content, author, createdAt, modifiedAt, modifiedBy, status from 
         post p join tag t on t.postId = p.id
         where t.name in (@tagName)
         group by id, blogId, p.name, title, content, author, createdAt, modifiedAt, modifiedBy, status"

This code is correct:

 let ``Query results can be async sequences``() =
     let posts = 
         asyncdb {
             use! ps = TestQueries.getPostsByTags ["framework"; "options"]
             let! p = ps |> AsyncSeq.toListAsync |> AsyncDb.fromAsync
             return p
         } |> run |> Async.RunSynchronously
     printfn "%A" posts

But this one fails with "connection closed" exception:

 let ``Query results can be async sequences``() =
     let postsAsync = 
         asyncdb {
             use! ps = TestQueries.getPostsByTags ["framework"; "options"]
             let! p = ps |> AsyncSeq.toListAsync 
             return p
         } |> run |> Async.RunSynchronously
     let posts = postsAsync |> Async.RunSynchronously
     printfn "%A" posts

Actually, async value shouldn't even leak outside use! scope in these examples, but you can just use let! bindings, since the source data reader is closed after reading all records.

[fwaris]

Thanks!! I will give a try soon.