Enumerate through IQueryable in batches

Question

I'm using Entity Framework Core to create an IQueryable. I'd like to find a way to get the results one page at a time (say 10 results at a time), but then expose this as an IEnumerable (or something similar like an IObservable). I also want to ensure that this is as memory-efficient as possible, so in other words, if the page size is 10, then there are only 10 entities sitting in memory at a time. Lastly, I want the database calls to be async, which is what makes this more difficult.

Here's some pseudo code, in which "ToPagingEnumerable" is the method I want to create or use from another library:

IQueryable<T> query = dbContext.SomeTable;
IEnumerable<T> results = query.ToPagingEnumerable(pageSize: 10);
ConvertAndSaveResults(results);

And here's a quick failed attempt, which won't work because you can't combine "yield" with "async":

public static IEnumerable<TSource> ToPagingEnumerable<TSource>(
    this IQueryable<TSource> source, 
    int size)
{
    var skip = 0;
    var cached = await source.Take(size).ToListAsync();
    while (cached.Any())
    {
        foreach (var item in cached)
        {
            yield return item;
        }
        skip += cached.Count;
        cached = await source.Skip(skip).Take(size).ToListAsync();
    }
}

I briefly looked at Reactive Streams (https://github.com/reactive-streams/reactive-streams-dotnet), and this seems similar to what I want to accomplish.

I think another option is to use Rx (Reactive) and create an Observable that grabs one page of results from the IQueryable (say 10 rows), feeds them to the Subscriber, then grabs another page (say another 10 rows), and feeds them to the Subscriber.

I just don't know enough about either of these libraries to know how to use them to accomplish my goal, or if there's an easier or different way.

Answer 1

Hi you can use this extension method

public static class QuerableExtensions
{
    public static IQueryable<TEntity> ToPage<TEntity>(this IQueryable<TEntity> query, PagingSettings pagingSettings) where TEntity : class
    {
        if (pagingSettings != null)
        {
            return query.Skip((pagingSettings.PageNumber - 1)*pagingSettings.PageSize).Take(pagingSettings.PageSize);
        }
        return query;
    }
    public static IQueryable<T> OrderByField<T>(this IQueryable<T> query, SortingSettings sortingSettings)
    {
        var exp = PropertyGetterExpression<T>(sortingSettings);

        var method = sortingSettings.SortOrder.Equals(SortOrder.Asc) ? "OrderBy" : "OrderByDescending";

        var types = new[] { query.ElementType, exp.Body.Type };

        var callExpression = Expression.Call(typeof(Queryable), method, types, query.Expression, exp);
        return query.Provider.CreateQuery<T>(callExpression);
    }


}

Where

    public class PagingSettings
{
    public PagingSettings()
        : this(50, 1)
    { }

    protected PagingSettings(int pageSize, int pageNumber)
    {
        PageSize = pageSize;
        PageNumber = pageNumber;
    }

    public int PageNumber { get; set; }
    public int PageSize { get; set; }
}

And to use it like this and you have to order your set before make paging

     public async Task<SimplePagedResult<TEntityDto>> GetAllPagedAsync<TEntityDto>(PagingSettins request) where TEntityDto : class
    {
        var projectTo = Set(); // Here is DBSet<TEnitity>


        var entityDtos = projectTo.OrderByField(new SortingSettings());

        if (request.PagingSettings != null)
            entityDtos = entityDtos.ToPage(request.PagingSettings);

        var resultItems = await entityDtos.ToListAsync();

        var result = MakeSimplePagedResult(request.PagingSettings, resultItems);
        return result;
    }

And class for result is

    public class SimplePagedResult<T>
{
    public IEnumerable<T> Results { get; set; }
    public int CurrentPage { get; set; }
    public int PageSize { get; set; }
}

Answer 2

Why run the query multiple times?

How about just:

var results  = source.ToAsyncEnumerable().Buffer(10);

Answer 3

Here is a way to process your IQueryable in batches without using an extension method:

var batchSize = 5000;

var myqueryable = // GetQueryable();

var count = myqueryable.Count();
var processed = 0;

while (processed < count)
{
    var take = batchSize <= count - processed ? batchSize : count - processed;

    var batchToProcess = myqueryable.Skip(processed).Take(take);

    //Do your processing, insert, what have you...

    processed += take;
}

Or you can simply specify the PageSize and PageNumber like so:

var myqueryable = // GetQueryable();

//var pageCount = (int)Math.Ceiling(myqueryable.Count() * 1D / PageSize);

int skip = PageNumber * PageSize - PageSize;

return myqueryable.Skip(skip).Take(PageSize);