DataTables statt DataReader
Wie bitte? NDO nutzt DataTables statt direkt auf DataReader zuzugreifen? Ist das nicht furchtbar langsam und skaliert schlecht? In diesem Beitrag zeigen wir, warum NDO auch mit DataTables sehr gut zurecht kommt.
Wenn Sie NDO ohne weitere Vorkehrungen nutzen, cacht NDO die Resultate Ihrer Queries komplett in einer DataTable. Wenn Sie nun einen Report über einen größeren Resultset erstellen wollen, brauchen Sie dafür sehr viel Speicher. Wenn Sie direkt mit ADO.NET programmieren, verwenden Sie in dieser Situation einfach einen DataReader, mit dem Sie einen Datensatz nach dem anderen einlesen können. Sie bearbeiten den Datensatz und schreiben die Ergebnisse in den Output-Stream. Dann können Sie den Datensatz verwerfen.
Auf diese Weise braucht die Operation nicht viel mehr Speicher, als das Einlesen eines Datensatzes erfordert. Wobei nicht verschwiegen werden darf, dass auch der DataReader einen Buffer benutzt, in dem eine gewisse Anzahl an Datensätzen gecacht wird.
Aber ist es wirklich notwendig, das Einlesen von Daten so fein zu granulieren? Natürlich ist das nicht der Fall. Die beste Lösung ist es, immer eine gewisse Anzahl an Datensätzen in einer Schleife einzulesen und zu verarbeiten, bis alle Datensätze verarbeitet sind.
SQL unterstützt diese Art an Abfragen mit der OFFSET/FETCH clause. Dadurch ist es möglich, eine gewisse Anzahl an Datensätzen im Resultset zu überspringen und auch nur eine gewisse Anzahl an Datensätzen als Ergebnis zu übertragen. NDO unterstützt dieses Feature mit den Skip und Take-Parametern in Abfragen.
Dies führt zu einer neuen Herausforderung, die gelöst werden will. Wenn wir unsere Abfrage in Teilabfragen unterteilen, kann sich die Ergebnismenge ändern, währen wir über die Teilabfragen iterieren. Das Problem lässt sich lösen, wenn man TransactionMode.Pessimistic und IsolationLevel.Snapshot nutzt. Wenn man IsolationLevel.Snapshot nutzen will, muss man dies in der Datenbank explizit erlauben:
ALTER DATABASE IsolationLevels
SET READ_COMMITTED_SNAPSHOT ON
GO
ALTER DATABASE IsolationLevels
SET ALLOW_SNAPSHOT_ISOLATION ON
GO
Der folgende Code zeigt, wie dieser IsolationLevel funktioniert.
class Program
{
static object lockObject = new object();
static DateTime dtStart; static void Main( string[] args )
{
PersistenceManager pmMain = new PersistenceManager();
// Prepare the data
pmMain.BuildDatabase();
var newEmp = new Employee() { Name = "Peter", Salary = 1000 };
pmMain.MakePersistent( newEmp );
newEmp = new Employee() { Name = "Paul", Salary = 2000 };
pmMain.MakePersistent( newEmp );
newEmp = new Employee() { Name = "Mary", Salary = 3000 };
pmMain.MakePersistent( newEmp );
pmMain.Save();
// Run concurrent operations while reading
// a snapshot
pmMain.TransactionMode = TransactionMode.Pessimistic;
pmMain.IsolationLevel = System.Data.IsolationLevel.Snapshot;
dtStart = DateTime.Now;
Output( pmMain.Objects<Employee>().ResultTable, "Task 1a" );
Task task = Task.Run( () =>
{
PersistenceManager pm = new PersistenceManager();
pm.TransactionMode = TransactionMode.Optimistic;
var emp = new Employee() { Name = "Mirko", Salary = 10000 };
pm.MakePersistent( emp );
pm.Save();
Output( pm.Objects<Employee>().ResultTable, "Task 2a" );
emp = pm.Objects<Employee>().Where( e => e.Name == "Mirko" ).Single();
emp.Salary = 11000;
pm.Save();
Output( pm.Objects<Employee>().ResultTable, "Task 2b" );
} );
Thread.Sleep( 1000 );
// Read from the same snapshot
Output( pmMain.Objects<Employee>().ResultTable, "Task 1b" );
Task.WaitAll( task );
pmMain.Abort(); // End Transaction
pmMain.IsolationLevel = System.Data.IsolationLevel.ReadCommitted;
pmMain.TransactionMode = TransactionMode.Optimistic;
Output( pmMain.Objects<Employee>().ResultTable, "After all" );
Console.WriteLine( "Ready" );
}
static void Output( List<Employee> emps, string taskName )
{
lock (lockObject)
{
Console.WriteLine( DateTime.Now - dtStart );
Console.WriteLine( taskName );
foreach (var item in emps)
{
Console.WriteLine( $"{item.Name} {item.Salary}" );
}
}
}
}
Wir arbeiten hier mit einer persistenten Klasse mit zwei Felder name und salary. In den ersten paar Zeilen von Main() legen wir drei Datensätze an. Dann setzen wir den TransactionMode und IsolationLevel auf Pessimistic / Snapshot. Das heißt: Alles, was ab jetzt bis zum nächsten Aufruf von Save() oder Close() passiert, wird in einer Transaktion ablaufen. Und der IsolationLevel.Snapshot stellt sicher, dass jede Abfrage, die in dieser Zeit passiert, aus demselben Snapshot der Datenbank genommen wird, der bei der ersten Abfrage erstellt wurde.
Nachdem alle Datensätze einmal angezeigt wurden, starten wir eine zweite Task, die einige Transaktionen mit der Kombination Optimistic / ReadCommitted ausführt. Wir legen einen neuen Datensatz an und ändern diesen. Während dieser Anweisungen zeigen wir die Resultate an. Sie können sehen, dass der neue und veränderte Datensatz korrekt dargestellt wird. Während dies geschieht, schläft die Haupt-Task für eine Sekunde. Nach dieser Zeit, werden alle Daten im Verlauf der Haupt-Task noch einmal dargestellt (Task 1b). Es zeigt sich, dass die Ausgabe exakt die gleichen Daten anzeigt, wie in unserer ersten Ausgabe (Task 1a).
Nach Beenden der Transaktion mit Abort() zeigen wir die Daten noch einmal an. Nun werden alle eingefügten und geänderen Daten angezeigt. Das sind die Ausgaben des Programms:
00:00:00.0279831
Task 1a
Peter 1000
Paul 2000
Mary 3000
00:00:00.0379771
Task 2a
Peter 1000
Paul 2000
Mary 3000
Mirko 10000
00:00:00.0549667
Task 2b
Peter 1000
Paul 2000
Mary 3000
Mirko 11000
00:00:01.0328228
Task 1b
Peter 1000
Paul 2000
Mary 3000
00:00:01.0388190
After all
Peter 1000
Paul 2000
Mary 3000
Mirko 11000
Ready
Beachten Sie die Zeitangaben. Task 1b läuft nach den Task2x-Ausgaben. Aber sie zeigt exakt den gleichen Zustand der Datenbank, wie er vorher vorlag.
Nun können Sie die gezeigte Technik zu einer Lösung zusammenbauen. Wählen Sie eine Anzahl an Datensätzen, die Sie in einer Teilabfrage bearbeiten wollen. Setzen Sie TransactionMode und IsolationLevel wie oben gezeigt. Nun iterieren Sie über die Datensätze, die Sie für Ihren Report benötigen und begrenzen Sie die Teilabfragen mit Take und Skip. Verarbeiten Sie die Daten der Teilabfrage und schreiben Sie die Ergebnisse in den Output-Stream. Nachdem alle Datensätze verarbeitet sind (Bedingung: Count == 0), beenden Sie die Transaktion mit Save() oder Abort(), oder rufen Sie Close() auf. Sie können den PersistenceManager auch in einem using-Block nutzen, dann wird Close() automatisch aufgerufen.
Achtung: Vermeiden Sie Änderungen an Objekten während einer Transaktion auf einen Snapshot.
Die Lösung sieht so aus:
static void MakeReport()
{
PersistenceManager pm = new PersistenceManager();
pm.TransactionMode = TransactionMode.Pessimistic;
pm.IsolationLevel = System.Data.IsolationLevel.Snapshot;
int pageSize = 2;
List<Employee> list;
NDOQuery<Employee> q = new NDOQuery<Employee>( pm );
q.Orderings.Add( new Query.AscendingOrder( "oid" ) );
int skip = 0;
do
{
q.Skip = skip;
q.Take = pageSize;
Console.WriteLine( $"Offset: {skip}" );
list = q.Execute();
foreach (var item in list)
{
Console.WriteLine( $" {item.Name} {item.Salary}" );
}
skip += pageSize;
pm.UnloadCache(); // Objects are cached with weak links, but we don't need the objects further.
} while (list.Count > 0);
pm.Abort(); // End Transaction
pm.IsolationLevel = System.Data.IsolationLevel.ReadCommitted;
pm.TransactionMode = TransactionMode.Optimistic;
}
Für OFFSET/FETCH benötigen Sie eine Sortierreihenfolge mit ORDER BY. Für diesen Zweck verwenden wir die ID-Spalte der Tabelle, die mit dem Ausdruck oid angesprochen werden kann. Da die ID-Spalte einen Clustered Index hat, sind die Teilabfragen mit OFFSET/FETCH sehr effizient. Wenn Sie nach anderen Kriterien sortieren wollen, erstellen Sie am Besten einen entsprechenden Index, um die Abfragen zu beschleunigen. Das Beispielprogramm erzeugt folgende Ausgaben:
Offset: 0
Peter 1000
Paul 2000
Offset: 2
Mary 3000
Mirko 11000
Offset: 4
Müssen wir extra erwähnen, dass Sie in Anwendungen aus dem richtigen Leben eine höhere pageSize verwenden sollten? Und beschweren Sie sich bitte nicht über die Ungleichheit der Gehälter... ;-)