Dapper ile Buffered ve Unbuffered Okuma: Büyük Result Set'leri Akıtmak

Sürüm notu: Bu yazıdaki kodlar Dapper 2.1 ve .NET 10 içindir.

Önce Tahmin Et

SQLite'ın güzel bir özelliği var: connection üzerine C# ile kendi SQL fonksiyonunu tanımlayabiliyorsun. Aşağıda touch adında, her çağrıldığında bir sayacı artıran bir fonksiyon tanımlayıp where koşuluna koyuyoruz. Böylece veritabanının kaç row'a dokunduğunu sayabiliyoruz. Aynı sorgu önce buffered, sonra buffered: false ile çalışıyor ve iki sonucun da önce Count()'u, sonra First()'ü alınıyor. İki satırdaki touched değerleri ne olur?

await using var conn = new SqliteConnection(cs);
await conn.OpenAsync();

var touched = 0;
conn.CreateFunction("touch", (long id) => { touched++; return 1; });

const string sql = "select id, title, price from books where touch(id) = 1 order by id";

var buffered = conn.Query<Book>(sql);
Console.WriteLine($"buffered:   count={buffered.Count()}, first={buffered.First().Title}, touched={touched}");

touched = 0;
var unbuffered = conn.Query<Book>(sql, buffered: false);
Console.WriteLine($"unbuffered: count={unbuffered.Count()}, first={unbuffered.First().Title}, touched={touched}");

Console.WriteLine($"types: {buffered.GetType().Name} / {unbuffered.GetType().Name}");

Tabloda dört kitap var. İpucu: ikinci satırdaki sayı 4 değil.

Hazırlık

Paketler aynı: Dapper 2.1.89 ve Microsoft.Data.Sqlite 10.0.14. Bu yazıda 6. yazıdaki küçük şemayı kullanıyoruz; büyük veriyi tabloya yazmak yerine recursive CTE ile anında üreteceğiz. Her örnek const string cs = "Data Source=bookstore.db"; ve await Setup.CreateAsync(cs); ile başlıyor:

using Dapper;
using Microsoft.Data.Sqlite;

public static class Setup
{
    public static async Task CreateAsync(string cs)
    {
        if (File.Exists("bookstore.db")) File.Delete("bookstore.db");
        await using var conn = new SqliteConnection(cs);
        await conn.ExecuteAsync("""
            create table authors (
                id      integer primary key,
                name    text not null,
                country text not null
            );
            create table books (
                id        integer primary key,
                author_id integer not null references authors(id),
                title     text not null,
                price     real not null,
                stock     integer not null check (stock >= 0)
            );
            insert into authors (id, name, country) values
                (1, 'Ursula K. Le Guin', 'USA'),
                (2, 'Frank Herbert', 'USA'),
                (3, 'Ted Chiang', 'USA');
            insert into books (id, author_id, title, price, stock) values
                (1, 1, 'A Wizard of Earthsea', 9.99, 12),
                (2, 1, 'The Left Hand of Darkness', 12.50, 4),
                (3, 2, 'Dune', 14.99, 20),
                (4, 3, 'Exhalation', 16.00, 7);
            """);
    }
}

public sealed class Book
{
    public int Id { get; set; }
    public string Title { get; set; } = "";
    public decimal Price { get; set; }
}

Cevap: Unbuffered Sonuç Bir Liste Değil, Bir Tarif

buffered:   count=4, first=A Wizard of Earthsea, touched=4
unbuffered: count=4, first=A Wizard of Earthsea, touched=5
types: List`1 / <QueryImpl>d__149`1

Buffered Query<T> (varsayılan hâli) sorguyu hemen çalıştırır, bütün row'ları okuyup bir List<T>'ye koyar, reader'ı kapatır ve sana listeyi verir. Count() ve First() artık yalnızca bellekteki listeye bakar; veritabanı toplamda dört row'a dokundu.

buffered: false ile dönen şey ise bir liste değil, derleyicinin ürettiği bir iterator: <QueryImpl>d__149`1. Bu iterator bir tarif gibi çalışır; her enumerate ettiğinde sorguyu baştan çalıştırır. Count() sorguyu bir kez çalıştırıp dört row'un hepsini okudu. First() sorguyu ikinci kez çalıştırdı ama ilk row'u alır almaz durdu; SQLite yalnızca bir row'a daha dokundu. Toplam beş. Unbuffered'ın iki yüzü burada tek satırda görünüyor: ihtiyacın kadar okursun, ama her enumerate yeni bir round trip demektir.

Bu arada touch'ın çalışması için connection'ı kendimiz açtık. 6. yazıda söylediğimiz gibi, Dapper kapalı connection'ı her çağrı için açıp kapatır ve CreateFunction gibi oturuma bağlı şeyler bu arada kaybolur.

Bellek: Bir Milyon Row

Farkın asıl hissedildiği yer büyük result set'ler. Bir milyon row'luk bir raporu iki şekilde okuyup bellekte ne kadar yer tuttuğuna bakalım:

await using var conn = new SqliteConnection(cs);
await conn.OpenAsync();

const string sql = """
    with recursive n(i) as (select 1 union all select i + 1 from n where i < 1000000)
    select i as id, 'Book #' || i as title, i % 50 + 0.99 as price from n
    """;

var baseline = GC.GetTotalMemory(forceFullCollection: true);

var all = conn.Query<Book>(sql).AsList();
var bufferedMb = (GC.GetTotalMemory(true) - baseline) / 1024 / 1024;
Console.WriteLine($"buffered:   {all.Count:N0} rows, sum={all.Sum(b => b.Price):N2}, held={bufferedMb} MB");
all = null;

baseline = GC.GetTotalMemory(forceFullCollection: true);
var count = 0;
var sum = 0m;
long peakMb = 0;
foreach (var book in conn.Query<Book>(sql, buffered: false))
{
    count++;
    sum += book.Price;
    if (count % 250_000 == 0)
        peakMb = Math.Max(peakMb, (GC.GetTotalMemory(true) - baseline) / 1024 / 1024);
}
Console.WriteLine($"unbuffered: {count:N0} rows, sum={sum:N2}, held={peakMb} MB");

buffered:   1,000,000 rows, sum=25,490,000.00, held=191 MB
unbuffered: 1,000,000 rows, sum=25,490,000.00, held=0 MB

GC.GetTotalMemory(true) tam bir garbage collection yapıp hâlâ canlı olan object'lerin boyutunu döndürür; yani bu sayılar "o anda elde tutulan" bellek. Buffered okumada bir milyon Book ve bir milyon string aynı anda canlı ve 191 MB tutuyor. Unbuffered'da her row işlendikten sonra çöp oluyor; 250 bin row'da bir aldığımız ölçümlerin hiçbiri bir megabaytı geçmedi. Toplamlar aynı, bellek profili tamamen farklı.

Bu, "her zaman unbuffered kullan" demek değil. Buffered okuma reader'ı ve connection'ı mümkün olan en kısa sürede serbest bırakır; Dapper'ın README'si de varsayılanın bu yüzden buffered olduğunu söyler: veritabanındaki kilitler ve ağ kullanımı en aza iner. Unbuffered'da ise enumerate ettiğin süre boyunca reader açık kalır. Döngünün içinde yavaş bir iş yapıyorsan, o süre boyunca connection meşgul ve sunucudaki kaynaklar elde tutuluyor demektir. Kural: sonucun tamamına aynı anda ihtiyacın varsa ve belleğe rahatça sığıyorsa buffered; row'ları tek tek işleyip atabileceğin büyük bir akışsa unbuffered.

QueryUnbufferedAsync ve IAsyncEnumerable

Sync Query'nin buffered: false hâlinin async karşılığı 2.0.138'den beri Dapper'da: QueryUnbufferedAsync<T>. IAsyncEnumerable<T> döner ve await foreach ile tüketilir. Bu bölümde ve sonraki iki bölümde aynı bir milyon row'luk sql sabitini kullanıyoruz:

await using var conn = new SqliteConnection(cs);

var expensive = 0;
await foreach (var book in conn.QueryUnbufferedAsync<Book>(sql))
{
    if (book.Price > 49) expensive++;
}
Console.WriteLine($"expensive: {expensive:N0}, state: {conn.State}");

expensive: 20,000, state: Closed

Önemli bir imza detayı: QueryUnbufferedAsync, diğer Dapper method'ları gibi IDbConnection üzerinde değil, DbConnection üzerinde tanımlı bir extension method; transaction parametresi de DbTransaction. Connection'ı IDbConnection olarak taşıyorsan bu method görünmez. 6. yazıdaki factory'nin DbConnection döndürmesinin sebeplerinden biri buydu. Connection kapalı geldiği için Dapper akış başlarken açtı, akış bitince kapattı; state: Closed bunu gösteriyor.

İptal ve Erken Çıkış

QueryUnbufferedAsync'in CommandDefinition alan bir public overload'u yok; dolayısıyla token'ı CommandDefinition ile veremezsin. Dapper'ın kaynağındaki yorum yolu açıkça söylüyor: iterator'larda iptal WithCancellation ile ifade edilir. İç implementasyon token'ı [EnumeratorCancellation] ile alıyor ve connection'ı açarken, reader'ı çalıştırırken ve her ReadAsync çağrısında kullanıyor:

using var cts = new CancellationTokenSource();
var seen = 0;
try
{
    await foreach (var book in conn.QueryUnbufferedAsync<Book>(sql).WithCancellation(cts.Token))
    {
        if (++seen == 1000) cts.Cancel();
    }
}
catch (OperationCanceledException ex)
{
    Console.WriteLine($"{ex.GetType().Name} after {seen:N0} rows, state: {conn.State}");
}

await foreach (var book in conn.QueryUnbufferedAsync<Book>(sql))
{
    if (book.Id == 3)
    {
        Console.WriteLine($"found {book.Title}, stopping early");
        break;
    }
}
Console.WriteLine($"state: {conn.State}");

TaskCanceledException after 1,000 rows, state: Closed
found Book #3, stopping early
state: Closed

İptal, bininci row'dan hemen sonraki ReadAsync'te yakalandı ve connection yine temiz biçimde kapandı. İkinci döngüde ise break ile erken çıktık; await foreach iterator'ı dispose etti, Dapper da finally bloğunda reader'ı kapattı ve kendi açtığı connection'ı kapattı. Kaynakta bir ayrıntı daha var: reader henüz sonuna gelmemişse Dapper önce command'ın Cancel() method'unu çağırıyor ki sunucu kalan row'ları göndermeye devam etmesin. 6. yazıda gördüğümüz gibi Microsoft.Data.Sqlite 10.0'da Cancel() boş bir method; SQLite'ta bunun etkisi yok, sunucu veritabanlarında ise erken çıkışı ucuzlatır.

QueryAsync ve CommandFlags.None

QueryAsync'te buffered parametresi yok, ama CommandDefinition ile CommandFlags.None verirsen buffering'i kapatabilirsin. Dönen şeye bakalım:

var flagsNone = await conn.QueryAsync<Book>(new CommandDefinition(sql, flags: CommandFlags.None));
Console.WriteLine($"QueryAsync + CommandFlags.None: {flagsNone.GetType().Name}");

QueryAsync + CommandFlags.None: <ExecuteReaderSync>d__55`1

Adı her şeyi söylüyor. Dapper reader'ı async açıyor ama row'ları senkron bir iterator ile, Read() çağırarak veriyor; kaynaktaki yorum da burada ReadAsync ve iptal kullanılamadığını belirtiyor. Yani bu kombinasyon, async kodun ortasında senkron okuma yapan bir hibrit. Async bir zincirde row'ları akıtmak istiyorsan doğru araç QueryUnbufferedAsync.

Sık Düşülen Tuzaklar

Unbuffered sonucu .ToList() ile listeye çevirmek. Bütün row'ları yine belleğe almış olursun, üstüne buffered'ın verdiği "reader'ı erken kapat" avantajını da kaybedersin.

Unbuffered sonucu birden fazla kez enumerate etmek. Tahmin örneğinde gördüğün gibi her enumerate sorguyu yeniden çalıştırır; arada veri değiştiyse iki enumerate farklı sonuç bile verebilir.

Unbuffered sonucu, connection'ı dispose eden bir method'dan döndürmek:

foreach (var book in GetBooks(cs))
    Console.WriteLine(book.Title);

static IEnumerable<Book> GetBooks(string cs)
{
    using var conn = new SqliteConnection(cs);
    return conn.Query<Book>("select id, title, price from books", buffered: false);
}

A Wizard of Earthsea
The Left Hand of Darkness
Dune
Exhalation

SQLite'ta bu kod çalışıyor, ama şans eseri. Sorgu method dönerken değil, foreach başladığında çalışıyor; o anda connection çoktan dispose edilmiş durumda. Dapper connection'ı kapalı görüp açıyor, SqliteConnection da dispose edildikten sonra connection string'ini koruduğu için açılabiliyor. Aynı kodu SqlClient ile denediğimde sorgu daha sunucuya gitmeden InvalidOperationException: The ConnectionString property has not been initialized. aldım; dispose edilmiş bir SqlConnection connection string'ini unutuyor. Unbuffered bir sonucu method dışına veriyorsan connection'ın ömrünü de onunla birlikte taşı, ya da IAsyncEnumerable<T> döndüren bir async iterator yazıp await using'i onun içine koy.

Enumerate ederken aynı connection üzerinde başka bir command çalıştırmak. SQLite buna izin veriyor; döngünün içinde her kitap için bir update çalıştırdığımda dört row da güncellendi. SQL Server'da ise MARS açık değilse, bir reader açıkken aynı connection'da ikinci bir command çalıştırmak exception fırlatır. Unbuffered döngünün içinde yazma yapacaksan ikinci bir connection kullan ya da önce buffered oku.

EF Core'dan gelenler için: AsAsyncEnumerable() ile EF Core'da da sonuçları akıtabilirsin; buffered ToListAsync() ile akış arasındaki tercih iki dünyada da aynı mantıkla yapılır.

Küçük Bir Deneme

★☆☆ Tahmin örneğinde unbuffered değişkenini bir kez .AsList() ile listeye çevir ve Count() ile First()'ü o liste üzerinden al. touched kaç olur, neden?

★★☆ Bir milyon row'luk sorguyu QueryUnbufferedAsync ile oku ve her 100 bin row'da bir GC.GetTotalMemory(true) değerini yazdır. Ölçtüğün değerleri sync buffered: false ile karşılaştır.

★★★ GetBooks method'unu async IAsyncEnumerable<Book> GetBooksAsync(string cs, [EnumeratorCancellation] CancellationToken ct = default) olarak yeniden yaz: connection method'un içinde await using ile açılsın, row'lar QueryUnbufferedAsync ile yield return edilsin. Çağıran taraf WithCancellation ile iptal ettiğinde connection'ın kapandığını StateChange event'ini dinleyerek doğrula.

Şimdiye kadar SQLite'ın kendi tiplerinden şikâyet etmedik, ama Guid, DateTimeOffset ve TimeSpan gibi tipler SQLite'tan string ya da long olarak geri geliyor. Bir sonraki yazıda type handler'larla bu dönüşümleri Dapper'a öğretecek, DateOnly/TimeOnly desteğinin 2.1.86'daki hikâyesine ve özel column mapping'ine bakacağız. Seri, Dapper'ın yeni sürümleri çıktıkça güncellenmeye devam edecek.