2022-03-07

Parallele gruppierte Aggregation in DuckDB

Hannes Mühleisen, Mark Raasveldt

Gruppierte Aggregationen sind ein Kernbefehl der Datenanalyse. Besonders wichtig sind sie für großskalige Datenanalyse („OLAP“), weil sie nützlich sind, um statistische Zusammenfassungen riesiger Tabellen zu berechnen. DuckDB enthält eine stark optimierte parallele Aggregationsfähigkeit für schnelle und skalierbare Summarisierung.

Direkt zu den Benchmarks?

Einleitung

GROUP BY ändert die Kardinalität der Ergebnismenge – statt dieselbe Zahl von Zeilen wie die Eingabe zurückzugeben (wie ein normales SELECT), gibt GROUP BY so viele Zeilen zurück, wie es Gruppen in den Daten gibt. Betrachten Sie diese (seltsam vertraute) Beispielabfrage:

SELECT
    l_returnflag,
    l_linestatus,
    sum(l_extendedprice),
    avg(l_quantity)
FROM
    lineitem
GROUP BY
    l_returnflag,
    l_linestatus;

Auf GROUP BY folgen zwei Spaltennamen, l_returnflag und l_linestatus. Das sind die Spalten, über die die Gruppen berechnet werden, und die resultierende Tabelle enthält alle Kombinationen derselben Spalte, die in den Daten vorkommen. Wir nennen die Spalten in der GROUP BY-Klausel die „Gruppierungsspalten“ und alle vorkommenden Wertekombinationen darin „Gruppen“. Die SELECT-Klausel enthält vier (nicht fünf) Ausdrücke: Verweise auf die Gruppierungsspalten und zwei Aggregationen: die sum über l_extendedprice und das avg über l_quantity. Wir nennen diese die „Aggregationen“. Wird sie ausgeführt, sieht das Ergebnis dieser Abfrage etwa so aus:

Allgemein erlaubt SQL nur Spalten, die in der GROUP BY-Klausel genannt sind, direkt Teil der SELECT-Ausdrücke zu sein, alle anderen Spalten müssen einer der Aggregatfunktionen wie sum, avg usw. unterliegen. Es gibt viele weitere Aggregatfunktionen, je nachdem, welches SQL-System Sie nutzen.

Wie sollte eine Query-Processing-Engine eine solche Aggregation berechnen? Es gibt viele Designentscheidungen, und wir diskutieren diese unten und insbesondere die von DuckDB getroffenen. Das Hauptproblem beim Berechnen von Gruppierungsergebnissen ist, dass die Gruppen in der Eingabetabelle in beliebiger Reihenfolge vorkommen können. Wäre die Eingabe bereits nach den Gruppierungsspalten sortiert, wäre die Aggregation trivial, weil wir einfach die aktuellen Werte der Gruppierungsspalten mit den vorherigen vergleichen könnten. Tritt eine Änderung auf, beginnt die nächste Gruppe und ein neues Aggregationsergebnis muss berechnet werden. Da der sortierte Fall einfach ist, ist ein geradliniger Weg, gruppierte Aggregationen zu berechnen, die Eingabetabelle zuerst nach den Gruppierungsspalten zu sortieren und dann den trivialen Ansatz zu nutzen. Das Sortieren der Eingabe ist jedoch leider immer noch eine rechnerisch teure Operation trotz unserer besten Bemühungen. Allgemein hat Sortieren eine Rechenkomplexität von O(nlogn), wobei n die Zahl der sortierten Zeilen ist.

Hash-Tabellen für Aggregation

Ein besserer Weg ist eine Hash-Tabelle. Hash-Tabellen sind eine grundlegende Datenstruktur in der Informatik, die uns erlaubt, Einträge mit einer Rechenkomplexität von O(1) zu finden. Eine volle Diskussion, wie Hash-Tabellen funktionieren, sprengt den Rahmen dieses Beitrags. Unten versuchen wir, uns auf eine sehr grundlegende Beschreibung und Überlegungen zur Aggregatberechnung zu konzentrieren.

Um n Zeilen zu einer Hash-Tabelle hinzuzufügen, betrachten wir eine Komplexität von O(n), viel, viel besser als O(nlogn) für Sortieren, besonders wenn n in die Milliarden geht. Die Abbildung oben illustriert, wie sich die Komplexität mit wachsender Tabellengröße entwickelt. Ein weiterer großer Vorteil: Wir müssen nicht zuerst eine sortierte Kopie der Eingabe anlegen, die genauso groß wäre wie die Eingabe. Stattdessen hat die Hash-Tabelle höchstens so viele Einträge, wie es Gruppen gibt, die (und das ist üblich) dramatisch weniger sein können als Eingabezeilen. Der Gesamtprozess ist also: Die Eingabetabelle scannen und für jede Zeile die Hash-Tabelle entsprechend aktualisieren. Sobald die Eingabe erschöpft ist, scannen wir die Hash-Tabelle, um Zeilen an Upstream-Operatoren oder direkt das Abfrageergebnis zu liefern.

Kollisionsbehandlung

Also, Hash-Tabelle! Wir bauen eine Hash-Tabelle auf der Eingabe mit den Gruppen als Schlüssel und den Aggregationen als Einträge. Dann berechnen wir für jede Eingabezeile einen Hash der Gruppenwerte, finden den Eintrag in der Hash-Tabelle und legen die Aggregatzustände an oder aktualisieren sie mit den Werten aus der Zeile? So einfach ist es leider nicht: Zwei Zeilen mit unterschiedlichen Werten für die Gruppierungsspalten können zu einem Hash führen, der auf denselben Hash-Tabellen-Eintrag zeigt, was zu falschen Ergebnissen führen würde.

Es gibt zwei Hauptansätze, dieses Problem zu umgehen: „Chaining“ oder „Linear Probing“. Beim Chaining halten wir die Aggregatwerte nicht direkt in der Hash-Tabelle, sondern eine Liste von Gruppenwerten und Aggregationen. Zeigen Gruppierungswerte auf einen Hash-Tabellen-Eintrag mit leerer Liste, werden die neue Gruppe und die Aggregationen einfach hinzugefügt. Zeigen Gruppierungswerte auf eine bestehende Liste, prüfen wir für jeden Listeneintrag, ob die Gruppierungswerte matchen. Wenn ja, aktualisieren wir die Aggregationen für diese Gruppe. Wenn nicht, legen wir einen neuen Listeneintrag an. Beim Linear Probing gibt es keine solchen Listen, aber beim Finden eines bestehenden Eintrags vergleichen wir die Gruppierungswerte, und wenn sie matchen, aktualisieren wir den Eintrag. Wenn nicht, gehen wir einen Eintrag in der Hash-Tabelle weiter und versuchen es erneut. Dieser Prozess endet, wenn entweder ein passender Gruppeneintrag gefunden wurde oder ein leerer Hash-Tabellen-Eintrag. Theoretisch gleichwertig, begünstigt die Computer-Hardware-Architektur Linear Probing wegen Cache-Lokalität. Weil Linear Probing die Hash-Tabellen-Einträge linear durchläuft, liegt der nächste Eintrag sehr wahrscheinlich im CPU-Cache, und der Zugriff ist schneller. Chaining führt im Allgemeinen zu wahlfreiem Zugriff und deutlich schlechterer Leistung auf modernen Hardware-Architekturen. Wir haben daher Linear Probing für unsere Aggregate-Hash-Tabelle übernommen.

Sowohl Chaining als auch Linear Probing verschlechtern die theoretische Lookup-Leistung von O(1) auf O(n) bezüglich der Hash-Tabellen-Größe, wenn es zu viele Kollisionen gibt, also zu viele Gruppen auf denselben Hash-Tabellen-Eintrag hashen. Eine gängige Lösung ist, die Hash-Tabelle zu vergrößern, sobald das „Fill Ratio“ eine Schwelle überschreitet, z. B. 75 % als Default für Javas HashMap. Das ist besonders wichtig, weil wir die Zahl der Gruppen im Ergebnis nicht kennen, bevor die Aggregation beginnt. Wir nehmen auch nicht an, die Zahl der Zeilen in der Eingabetabelle zu kennen. Wir beginnen daher mit einer recht kleinen Hash-Tabelle und vergrößern sie, sobald das Fill Ratio eine Schwelle überschreitet. Die grundlegende Hash-Tabellen-Struktur ist in der Abbildung unten gezeigt, die Tabelle hat vier Slots 0–4. Es sind bereits drei Gruppen in der Tabelle, mit Gruppenschlüsseln 12, 5 und 2. Jede Gruppe hat Aggregatwerte (z. B. aus einer SUM) von 43 usw.

Eine große Herausforderung beim Vergrößern einer teilweise gefüllten Hash-Tabelle: Nach dem Resize sind alle Gruppen am falschen Platz, und wir müssten alles verschieben, was sehr teuer wäre.

Um Resize effizient zu unterstützen, haben wir eine zweiteilige Aggregate-Hash-Tabelle umgesetzt, bestehend aus einem getrennt allokierten Pointer-Array, das in Payload-Blöcke zeigt, die Gruppierungswerte und Aggregatzustände für jede Gruppe enthalten. Die Pointer sind keine tatsächlichen Pointer, sondern symbolisch, sie verweisen auf eine Block-ID und einen Zeilenoffset innerhalb dieses Blocks. Das ist in der Abbildung oben gezeigt, die Hash-Tabellen-Einträge sind über zwei Payload-Blöcke verteilt. Beim Resize werfen wir das Pointer-Array weg und allokieren ein größeres. Dann lesen wir alle Payload-Blöcke erneut, hashen die Gruppenwerte und fügen Pointer darauf ins neue Pointer-Array ein. Die Gruppendaten bleiben also unverändert, was die Kosten des Vergrößerns der Hash-Tabelle stark senkt. Das sieht man in der Abbildung unten, wo wir das Pointer-Array verdoppeln, die Payload-Blöcke aber unverändert bleiben.

Das naive zweiteilige Hash-Tabellen-Design würde beim Resize ein erneutes Hashen aller Gruppenwerte erfordern, was besonders bei String-Werten teuer sein kann. Um das zu beschleunigen, schreiben wir auch den Roh-Hash der Gruppenwerte für jede Gruppe in die Payload-Blöcke. Beim Resize müssen wir die Gruppen dann nicht neu hashen, sondern können sie aus den Payload-Blöcken lesen, den neuen Offset ins Pointer-Array berechnen und dort einfügen.

Die zweiteilige Hash-Tabelle hat einen großen Nachteil beim Nachschlagen von Einträgen: Es gibt keine Ordnung zwischen Pointer-Array und Gruppeneinträgen in den Payload-Blöcken. Dem Pointer zu folgen erzeugt daher wahlfreien Zugriff in der Speicherhierarchie. Das führt zu unnötigen Stalls in der Berechnung. Um das zu mildern, erweitern wir das Speicherlayout des Pointer-Arrays um einige (1 oder 2) Bytes aus dem Gruppenhash zusätzlich zum Pointer auf den Payload-Wert. So kann Linear Probing zuerst die Hash-Bits im Pointer-Array mit dem aktuellen Gruppenhash vergleichen und entscheiden, ob es sich lohnt, dem Payload-Pointer zu folgen oder nicht. Das kann potenziell für jede Gruppe in der Pointer-Kette weitergehen. Nur wenn die Hash-Bits matchen, müssen wir dem Pointer tatsächlich folgen und die tatsächlichen Gruppen vergleichen. Diese Optimierung reduziert stark, wie oft dem Pointer zu den Payload-Blöcken gefolgt werden muss, und senkt damit die Zahl der wahlfreien Speicherzugriffe, die direkt mit der Gesamtleistung zusammenhängen. Sie hat den netten Nebeneffekt, auch vollständige Gruppenvergleiche stark zu reduzieren, die ebenfalls teuer sein können, z. B. beim Aggregieren über Gruppen, die Strings enthalten.

Eine weitere (kleinere) Optimierung betrifft die Breite der Pointer-Array-Einträge. Für kleine Hash-Tabellen mit wenigen Einträgen brauchen wir nicht viele Bits, um die Payload-Block-Offset-Pointer zu kodieren. DuckDB unterstützt sowohl 4-Byte- als auch 8-Byte-Pointer-Array-Einträge.

Für die meisten Aggregatabfragen wird der Großteil der Query-Processing-Zeit mit dem Nachschlagen von Hash-Tabellen-Einträgen verbracht, weshalb es sich lohnt, Zeit in ihre Optimierung zu stecken. Wer neugierig ist: Der Code für all das liegt im DuckDB-Repo, aggregate_hashtable.cpp. Es gibt eine weitere Optimierung, wenn wir aus Spaltenstatistiken wissen, dass es nur wenige verschiedene Gruppen gibt, die Perfect Hash Aggregate, aber das ist ein Thema für einen anderen Beitrag. Wir sind hier aber noch nicht fertig.

Parallele Aggregation

Wir haben jetzt ein Aggregate-Hash-Tabellen-Design, das für gruppierte Aggregationen recht gut funktionieren sollte, haben aber noch nicht berücksichtigt, dass DuckDB alle Abfragen automatisch parallelisiert, um mehrere Hardware-Threads („CPUs“) zu nutzen. Wie funktioniert Parallelität zusammen mit Hash-Tabellen? Allgemein lautet die Antwort leider: „Schlecht“. Hash-Tabellen sind empfindliche Strukturen, die parallele Änderungen nicht gut vertragen. Stellen Sie sich vor, ein Thread wollte die Hash-Tabelle vergrößern, während ein anderer neue Gruppendaten hinzufügen will. Oder wie sollen wir mehrere Threads handhaben, die gleichzeitig neue Gruppen für denselben Eintrag einfügen? Man könnte Locks nutzen, damit jeweils nur ein Thread die Tabelle nutzt, das würde die Parallelisierung der Abfrage aber weitgehend zunichtemachen. Es gab reichlich Forschung zu concurrency-freundlichen Hash-Tabellen, die kurze Zusammenfassung ist aber: Es ist immer noch ein offenes Thema.

Es ist möglich, jeden Thread Daten von Downstream-Operatoren lesen und individuelle, lokale Hash-Tabellen bauen zu lassen und diese später von einem einzelnen Thread zusammenzuführen. Das funktioniert recht gut, wenn es wenige Gruppen gibt, wie im Beispiel am Anfang dieses Beitrags. Bei wenigen Gruppen kann ein einzelner Thread viele thread-lokale Hash-Tabellen mergen, ohne einen Engpass zu erzeugen. Es ist jedoch durchaus möglich, dass es so viele Gruppen wie Eingabezeilen gibt, das passiert oft, wenn jemand nach einer Spalte gruppiert, die Kandidat für einen Primärschlüssel wäre, z. B. observation_number, timestamp usw. Was also nötig ist, ist ein paralleles Mergen der parallelen Hash-Tabellen. Wir übernehmen eine Methode von Leis et al.: Jeder Thread baut nicht eine, sondern mehrere partitionierte Hash-Tabellen anhand einer Radix-Partitionierung auf dem Gruppenhash.

Die zentrale Beobachtung hier: Wenn zwei Gruppen einen unterschiedlichen Hash-Wert haben, können sie unmöglich dieselbe sein. Aufgrund dieser Eigenschaft ist es möglich, die Hash-Werte zu nutzen, um vollständig unabhängige Partitionen der Gruppen zu erzeugen, ohne jegliche Kommunikation zwischen Threads, solange alle Threads dasselbe Partitionierungsschema nutzen (siehe Phase 1 im Diagramm oben).

Nachdem alle lokalen Hash-Tabellen gebaut sind, weisen wir einzelnen Worker-Threads einzelne Partitionen zu und mergen die Hash-Tabellen innerhalb dieser Partition zusammen (Phase 2). Weil die Partitionen über das Radix-Partitionierungsschema auf dem Hash erzeugt wurden, können alle Worker-Threads unabhängig die Hash-Tabellen innerhalb ihrer jeweiligen Partitionen mergen. Das Ergebnis ist korrekt, weil jede Gruppe in genau eine Partition geht.

Ein interessantes Detail: Wir müssen nie eine finale (möglicherweise riesige) Hash-Tabelle bauen, die alle Gruppen hält, weil die Radix-Gruppenpartitionierung sicherstellt, dass jede Gruppe in einer Partition lokalisiert ist.

Es gibt zwei zusätzliche Optimierungen für die Strategie der parallelen partitionierten Hash-Tabelle:

1. Wir beginnen erst zu partitionieren, wenn die Aggregate-Hash-Tabelle eines einzelnen Threads ein festes Limit an Einträgen überschreitet, derzeit auf 10.000 Zeilen gesetzt. Das liegt daran, dass eine partitionierte Hash-Tabelle nicht umsonst ist. Für jede hinzugefügte Zeile müssen wir herausfinden, in welche Partition sie gehört, und am Ende müssen wir alles wieder zusammenführen. Deshalb beginnen wir nicht zu partitionieren, bis der Parallelisierungsnutzen die Kosten überwiegt. Da die Partitionierungsentscheidung individuell für jeden Thread ist, kann es durchaus sein, dass nur einige Threads zu partitionieren beginnen. Ist das der Fall, müssen wir die Hash-Tabellen der Threads, die das nicht getan haben, vor dem Mergen partitionieren. Das ist jedoch eine vollständig thread-lokale Operation und stört die Parallelität nicht.
2. Wir hören auf, Werte zu einer Hash-Tabelle hinzuzufügen, sobald ihr Pointer-Array eine bestimmte Schwelle überschreitet. Jeder Thread baut dann mehrere Sätze potenziell partitionierter Hash-Tabellen. Das liegt daran, dass wir nicht wollen, dass das Pointer-Array beliebig groß wird. Das erzeugt potenziell Duplikateinträge für dieselbe Gruppe in mehreren Hash-Tabellen, ist aber unproblematisch, weil wir sie später ohnehin alle mergen. Diese Optimierung funktioniert besonders gut auf Datensätzen mit vielen verschiedenen Gruppen, deren Gruppenwerte in der Eingabe irgendwie geclustert sind. Zum Beispiel beim Gruppieren nach Tag in einem Datensatz, der nach Datum geordnet ist.

Es gibt Arten von Aggregationen, die den parallelen und partitionierten Hash-Tabellen-Ansatz nicht nutzen können. Eine Summe zu parallelisieren ist trivial, weil die Summe des Gesamtergebnisses einfach die Summe der Einzelergebnisse ist, das ist für Berechnungen wie median ziemlich unmöglich, die DuckDB ebenfalls unterstützt. Auch aus diesem Grund unterstützt DuckDB approx_quantile, das parallelisierbar ist.

Experimente

Nachdem das alles zusammenkommt, ist es Zeit für einige Leistungsexperimente. Wir vergleichen DuckDBs Aggregationsoperator wie oben beschrieben mit demselben Operator in verschiedenen Python-Data-Wrangling-Bibliotheken. Die anderen Kandidaten sind Pandas, Polars und Arrow. Die sind gewählt, weil sie alle einen Aggregationsoperator auf Pandas-DataFrames ausführen können, ohne zuerst in ein anderes Speicherformat zu konvertieren, genau wie DuckDB.

Für unsere Benchmarks erzeugen wir einen synthetischen Datensatz mit einer vordefinierten Zahl von Gruppen über zwei Integer-Spalten und einigen zufälligen Integer-Daten zum Aggregieren. Der gesamte Datensatz wird vor den Experimenten gemischt, um zu verhindern, dass die geclusterte Natur der synthetisch erzeugten Daten ausgenutzt wird. Für jede Gruppe berechnen wir zwei Aggregationen, Summe der Datenspalte und einen einfachen Count. Die SQL-Version dieser Aggregation wäre SELECT g1, g2, sum(d), count(*) FROM dft GROUP BY g1, g2 LIMIT 1;. In den Experimenten unten variieren wir die Datensatzgröße und die Zahl der Gruppen darin. Das sollte das Skalierungsverhalten der Aggregation schön zeigen.

Weil wir nicht an der Materialisierungszeit der Ergebnismenge interessiert sind, die bei Millionen von Gruppen erheblich wäre, folgt auf die Aggregation ein Operator, der nur die erste Zeile holt. Das ändert die Komplexität der Aggregation überhaupt nicht, weil sie alle Daten sammeln muss, bevor sie überhaupt die erste Ergebniszeile erzeugt, da Daten in der allerletzten Eingabezeile die Ergebnisse für die erste Ergebniszeile ändern könnten. Natürlich wäre das in der Praxis recht unrealistisch, isoliert aber schön das Verhalten nur des Aggregationsoperators, weil eine head(1)-Operation auf drei Spalten recht günstig und konstant in der Ausführungszeit sein sollte.

Wir messen die verstrichene Wanduhrzeit, die nötig ist, um jede Aggregation abzuschließen. Um kleinere Schwankungen auszugleichen, wiederholen wir jede Messung dreimal und berichten die Medianzeit. Alle Experimente liefen auf einem MacBook Pro 2021 mit einem Zehn-Kern-M1-Max-Prozessor und 64 GB RAM. Unser Datenerzeugungs-Benchmark-Skript ist online verfügbar, und wir laden interessierte Leser ein, das Experiment auf ihren Maschinen erneut auszuführen.

Nun zu einigen Ergebnissen. Wir beginnen damit, die Zahl der Zeilen in der Tabelle zwischen einer Million und 100 Millionen zu variieren. Wir wiederholen das Experiment sowohl für eine feste (kleine) Gruppenzahl von 1000 als auch wenn die Zahl der Gruppen der Zahl der Zeilen entspricht. Die Ergebnisse sind als Log-Log-Plot dargestellt, wir sehen, wie DuckDB die anderen Systeme durchgängig übertrifft, wobei das einthreadige Pandas am langsamsten ist und Polars und Arrow im Allgemeinen ähnlich sind.

Für das nächste Experiment fixieren wir die Zahl der Zeilen auf 100 Millionen (die größte Größe, mit der wir experimentiert haben) und zeigen das volle Verhalten beim Erhöhen der Gruppengröße. Wieder sehen wir, wie DuckDB durchgängig gutes Skalierungsverhalten beim Erhöhen der Gruppengröße zeigt, weil es alle Phasen der Aggregation wie oben skizziert wirksam parallelisieren kann. Wenn Sie interessiert sind, wie wir diese Plots erzeugt haben, das Plot-Skript ist ebenfalls verfügbar.

Fazit

Datenanalyse-Pipelines, die überwiegend aggregieren, verbringen den Großteil ihrer Ausführungszeit in der Aggregate-Hash-Tabelle, weshalb es sich lohnt, eine unverschämte Menge menschlicher Zeit in ihre Optimierung zu stecken. Wir haben einige Ideen für zukünftige Arbeit dazu, zum Beispiel möchten wir unsere Arbeit beim Vergleichen von Sortierschlüsseln auf das Vergleichen von Gruppen in der Aggregate-Hash-Tabelle erweitern. Wir möchten auch die Fähigkeit ergänzen, dynamisch die Zahl der Partitionen zu wählen, die ein Thread nutzt, anhand dynamischer Beobachtung der erzeugten Hash-Tabelle, z. B. wenn Partitionen unausgewogen sind, könnten wir mehr Bits dafür nutzen. Ein weiteres großes Gebiet zukünftiger Arbeit ist, unsere Aggregate-Hash-Tabelle mit Out-of-Core-Operationen arbeiten zu lassen, wenn eine einzelne Hash-Tabelle nicht mehr in den Speicher passt, das ist besonders beim Mergen problematisch. Und natürlich gibt es immer Möglichkeiten, einen Aggregationsoperator feinabzustimmen, und wir verbessern DuckDBs Aggregationsoperator laufend.

Wenn Sie an so schneidiger Data Engineering wie dieser mitarbeiten wollen, die von Tausenden Menschen genutzt wird, erwägen Sie, zu DuckDB beizutragen oder kommen Sie zu uns zu DuckLabs in Amsterdam!