2024-03-29

Kein Speicher? Kein Problem. Externe Aggregation in DuckDB

Laurens Kuiper

Die meisten gruppierten Aggregationsabfragen liefern nur wenige Ausgabezeilen. Zum Beispiel liefert „Wie viele Flüge sind in den letzten zehn Jahren von jeder europäischen Hauptstadt gestartet?“ eine Zeile pro europäischer Hauptstadt, selbst wenn die Tabelle mit allen Fluginformationen Millionen Zeilen hat. Das ist nicht immer so: „Wie viele Bestellungen hat jeder Kunde in den letzten zehn Jahren aufgegeben?“ liefert eine Zeile pro Kunde, das können Millionen sein, was den Speicherverbrauch der Abfrage deutlich erhöht. Selbst wenn die Aggregation nicht in den Speicher passt, kann DuckDB die Abfrage aber trotzdem beenden.

Kein Interesse an der Implementierung? Springen Sie direkt zu den Experimenten!

Einleitung

Vor etwa zwei Jahren haben wir unseren ersten Blogpost zu DuckDBs Hash-Aggregation veröffentlicht: „Parallel Grouped Aggregation in DuckDB“. Warum schreiben wir jetzt einen weiteren Blogpost?

Anders als die meisten Datenbanksysteme, die Server sind, wird DuckDB in allen möglichen Umgebungen genutzt, die vielleicht nicht viel Speicher haben. Manche Datenbankabfragen, etwa Aggregationen mit vielen eindeutigen Gruppen, brauchen aber viel Speicher. Der Laptop, auf dem ich das schreibe, hat 16 GB RAM. Was, wenn eine Abfrage 20 GB braucht? Dann passiert das:

Out of Memory Error: could not allocate block of size X (Y/Z used)

Die Abfrage wird abgebrochen. Leider können wir nicht mehr RAM herunterladen. Zum Glück hat dieser Laptop auch eine schnelle SSD mit 1 TB Speicher. In vielen Fällen brauchen wir nicht alle 20 GB Daten gleichzeitig im Speicher, und wir können einen Teil vorübergehend im Speicher ablegen. Laden wir ihn bei Bedarf zurück, können wir die Abfrage trotzdem beenden. Wir müssen den Speicher sparsam nutzen, denn obwohl moderne SSDs schnell sind, sind sie immer noch deutlich langsamer als RAM.

Kurz gesagt geht es in diesem Post darum. Seit dem Release 0.9.0 kann DuckDBs Hash-Aggregation mehr eindeutige Gruppen verarbeiten, als in den Speicher passen, indem Daten auf den Speicher ausgelagert werden. In diesem Post erklären wir, wie das funktioniert. Wenn Sie wissen wollen, was Hash-Aggregation ist, wie Hash-Kollisionen aufgelöst werden oder wie DuckDBs Hash-Tabelle aufgebaut ist, schauen Sie sich unseren ersten Blogpost zur Hash-Aggregation an.

Speicherverwaltung

Die meisten Datenbanksysteme speichern persistente Daten auf „Pages“. Auf Anfrage können diese Pages aus der Datenbankdatei im Speicher gelesen, in den RAM gelegt und bei Bedarf wieder zurückgeschrieben werden. Die gängige Weisheit ist, alle Pages gleich groß zu machen: Das erlaubt das Austauschen von Pages und vermeidet Fragmentierung in RAM und Speicher. Beim Start der Datenbank wird ein Teil des Speichers für diese Pages allokiert und reserviert, der „Buffer Pool“. Die Datenbankkomponente, die den Buffer Pool verwaltet, heißt treffend „Buffer Manager“.

Der restliche Speicher ist für kurzlebige, also temporäre Speicherallokationen reserviert, etwa Hash-Tabellen für Aggregation. Diese Allokationen geschehen anders, was gut ist, denn bei vielen eindeutigen Gruppen müssen Hash-Tabellen sehr groß sein, und wir hätten die Pages fester Größe dafür sowieso nicht nutzen können. Haben wir mehr temporäre Daten, als in den Speicher passen, müssen Operatoren wie Aggregation entscheiden, wann sie selektiv Daten in eine temporäre Datei im Speicher schreiben.

… Zumindest ist das der traditionelle Weg. Für DuckDB machte das wenig Sinn. Warum sollten wir persistente und temporäre Daten so unterschiedlich verwalten? Der Unterschied ist, dass persistente Daten persistiert werden sollen und temporäre nicht. Warum kann ein Buffer Manager nicht beides verwalten?

DuckDBs Buffer Manager ist nicht traditionell. Die meisten persistenten und temporären Daten liegen auf Pages fester Größe und werden vom Buffer Manager verwaltet. Der Buffer Manager versucht, Ihren Speicher bestmöglich zu nutzen. Das bedeutet, wir reservieren keinen Teil des Speichers für einen Buffer Pool. So kann DuckDB den gesamten Speicher für persistente Daten nutzen, nicht nur einen Teil, wenn das für Ihre Last am besten ist. Machen Sie große Aggregationen, die viel Speicher brauchen, kann DuckDB persistente Daten aus dem Speicher verdrängen, um Platz für eine große Hash-Tabelle zu schaffen.

Weil DuckDBs Buffer Manager allen Speicher verwaltet, persistente und temporäre Daten, ist er viel besser darin, zu entscheiden, wann temporäre Daten auf den Speicher geschrieben werden, als Operatoren wie Aggregation es je sein könnten. Die Verantwortung fürs Auslagern dem Buffer Manager zu überlassen, erspart uns auch den Aufwand, in jedem Operator, der Daten verarbeiten muss, die nicht in den Speicher passen, das Lesen und Schreiben in eine temporäre Datei zu implementieren.

Warum verwalten Buffer Manager in anderen Datenbanksystemen keine temporären Daten? Es gibt zwei Probleme: Speicherfragmentierung und ungültige Referenzen.

Speicherfragmentierung

Hash-Tabellen und andere Datenstrukturen in Query-Operatoren haben nicht genau eine feste Größe wie die Pages für persistente Daten. Wir wollen auch nicht viele Pages variabler Größe im Speicher neben den Pages fester Größe haben, weil das Speicherfragmentierung verursachen würde.

Ideal wäre eine feste Größe für alle unsere Speicherallokationen, aber das ist keine gute Idee: Manchmal ist der effizienteste Weg, eine Abfrage zu verarbeiten, zum Beispiel ein großes Array zu allokieren. Also haben wir uns darauf geeinigt, eine feste Größe für fast alle Allokationen zu nutzen. Diese kurzlebigen Allokationen werden nach Gebrauch sofort deallokiert, anders als die Pages fester Größe für persistente Daten, die behalten werden. Diese Allokationen fragmentieren einander nicht, weil jemalloc, das DuckDB nach Möglichkeit zur Speicherallokation nutzt, Allokationen über Größenklassen kategorisiert und getrennte Arenas dafür hält.

Ungültige Referenzen

Temporäre Daten können meist nicht so, wie sie sind, auf den Speicher geschrieben werden, weil sie oft Pointer enthalten. DuckDB implementiert zum Beispiel den von Umbra vorgeschlagenen String-Typ, der eine feste Breite hat. Strings länger als 12 Zeichen werden nicht im String-Typ selbst gespeichert, sondern woanders, und stattdessen wird ein Pointer auf dieses „Woanders“ gespeichert.

Das schafft ein Problem, wenn wir Daten auf den Speicher auslagern wollen. Angenommen, dieses „Woanders“, wo Strings länger als 12 Zeichen liegen, ist eine jener Pages, die der Buffer Manager jederzeit auf den Speicher auslagern kann, um Speicher freizugeben. Wird die Page ausgelagert und dann wieder geladen, landet sie höchstwahrscheinlich an einer anderen Adresse im Speicher. Die Pointer, die auf die langen Strings zeigten, sind jetzt ungültig, weil sie noch auf die vorherige Adresse zeigen!

Der übliche Weg, Daten mit Pointern auf den Speicher zu schreiben, ist, sie zuerst zu serialisieren. Beim Zurücklesen in den Speicher müssen sie wieder deserialisiert werden. (De-)Serialisierung kann eine teure Operation sein, deshalb gibt es Datenformate wie Arrow Flight, die die Kosten minimieren wollen. Arrow können wir hier aber nicht nutzen, weil Arrow ein spaltenmajorisches Layout ist, ein zeilenmajorisches Layout für Hash-Tabellen aber effizienter ist.

Wir könnten eine zeilenmajorische Version von Arrow Flight erzeugen, aber wir können (De-)Serialisierung ganz vermeiden: Wir haben ein spezialisiertes zeilenmajorisches Page-Layout geschaffen, das die alten ungültig gewordenen Pointer nutzt, um nach dem Zurücklesen der Daten in den Speicher neue gültige Pointer neu zu berechnen.

Das Page-Layout legt Zeilen fester Größe und Daten variabler Größe wie Strings auf getrennte Pages. Die Größe der Zeilen ist für eine Abfrage fest: Nach einer SQL-Abfrage erzeugt und führt DuckDB einen Query-Plan aus. Noch vor der Ausführung dieses Plans wissen wir also bereits, welche Spalten wir brauchen, ihre Typen und wie breit diese Typen sind.

Wie im Bild unten gezeigt, braucht es eine kleine Menge „MetaData“, um die Pointer neu zu berechnen. Die Zeilen fester Größe liegen in „Row Pages“, Daten variabler Größe in „Var Pages“.

Erinnern Sie sich, dass in den Zeilen fester Größe Pointer auf Daten variabler Größe stehen. Die MetaData beschreibt, welche Zeilen fester Größe auf welche Var Page zeigen und die zuletzt bekannte Adresse der Var Page. Zum Beispiel beschreibt MetaData 1 fünf Zeilen, die in Row Page 1 bei Offset 0 liegen, mit Daten variabler Größe in Var Page 1, die die Adresse 0x42 hatte.

Angenommen, der Buffer Manager entscheidet, Var Page 1 auszulagern. Wenn wir Var Page 1 wieder anfordern, wird sie an Adresse 0x500 geladen. Die Pointer in diesen fünf Zeilen sind jetzt ungültig. Eine der Zeilen enthält zum Beispiel den Pointer 0x48, das heißt, sie liegt bei Offset 0x48 - 0x42 = 6 in Var Page 1. Wir können den Pointer neu berechnen, indem wir den Offset zur neuen Adresse der Page addieren: 0x500 + 6 = 0x506. Die Pointer-Neuberechnung geschieht für Zeilen, deren Strings auf derselben Row- und Var-Page liegen, daher erzeugen wir jedes Mal neue MetaData, wenn eine Row Page oder Var Page voll ist.

Der Vorteil der Pointer-Neuberechnung gegenüber (De-)Serialisierung ist, dass sie lazy erfolgen kann. Wir können prüfen, ob die Var Page ausgelagert wurde, indem wir den Pointer in der MetaData mit dem aktuellen Pointer auf die Page vergleichen. Sind sie gleich, müssen wir die Pointer nicht neu berechnen.

Externe Aggregation

Jetzt, da wir wissen, wie wir mit temporären Daten umgehen, ist es endlich Zeit, über Hash-Aggregation zu sprechen. Die erste große Herausforderung ist, die Aggregation parallel auszuführen.

DuckDB nutzt Morsel-Driven Parallelism, um die Abfrageausführung zu parallelisieren, was im Wesentlichen bedeutet, dass Query-Operatoren wie Aggregation parallelismussensibel sein müssen. Das unterscheidet sich von plan-driven Parallelism, der Operatoren nichts von Parallelität wissen lässt.

Kurz zusammengefasst aus unserem ersten Blogpost zur Aggregation: In DuckDB hat jeder aktive Thread seine eigene thread-lokale Hash-Tabelle, in die er Eingabedaten senkt. Das hält Threads beschäftigt, bis alle Eingabedaten gelesen sind. Mehrere Threads haben wahrscheinlich genau dieselbe Gruppe in ihrer Hash-Tabelle. Daher müssen die thread-lokalen Hash-Tabellen kombiniert werden, um die gruppierte Aggregation abzuschließen. Das kann parallel geschehen, indem die Hash-Tabellen partitioniert und jeder Thread der Kombination der Daten einer Partition zugewiesen wird. Größtenteils nutzen wir immer noch denselben Ansatz. Das sehen Sie im Bild unten, das unsere neue Implementierung darstellt.

Die erste Phase nennen wir Thread-Local Pre-Aggregation. Die Eingabedaten sind Morsels, Chunks von etwa 100.000 Zeilen. Diese werden aktiven Threads zugewiesen, die sie einzeln in ihre thread-lokale Hash-Tabelle senken, bis alle Eingabedaten gelesen sind. Wir nutzen Linear Probing, um Kollisionen aufzulösen, und Salt, um den Overhead im Umgang mit diesen Kollisionen zu senken. Das ist in unserem ersten Blogpost zur Aggregation erklärt, daher wiederhole ich es hier nicht.

Jetzt, da wir erklärt haben, was sich nicht geändert hat, können wir über das sprechen, was sich geändert hat. Der erste Unterschied zur letzten Zeit ist die Art, wie wir partitionieren. Früher, bei zum Beispiel 32 Threads, erzeugte jeder Thread 32 Hash-Tabellen, eine pro Partition. Das summiert sich auf stolze 1024 Hash-Tabellen, was bei noch mehr aktiven Threads nicht gut skalierte. Jetzt hat jeder Thread eine Hash-Tabelle, aber die Daten in jeder Hash-Tabelle sind partitioniert. Die Daten liegen außerdem auf dem spezialisierten Page-Layout, das wir zuvor vorgestellt haben, sodass sie leicht auf den Speicher ausgelagert werden können.

Der zweite Unterschied ist, dass die Hash-Tabellen während der Thread-Local Pre-Aggregation nicht resized werden. Wir halten die Hash-Tabellen klein, was die Zahl der Cache Misses in dieser Phase senkt. Das bedeutet, dass die Hash-Tabelle irgendwann voll ist. Wenn sie voll ist, setzen wir sie zurück und fangen von vorn an. Das können wir, weil wir die Aggregation später in der zweiten Phase beenden. Beim Zurücksetzen der Hash-Tabelle „unpin“en wir die Pages, die die eigentlichen Daten speichern, was unserem Buffer Manager sagt, dass er sie auf den Speicher schreiben kann, wenn er Speicher freigeben muss.

Zusammen führen diese beiden Änderungen zu einem geringen Speicherbedarf in der ersten Phase. Jeder Thread muss nur eine kleine Hash-Tabelle im Speicher halten. Wir können durch mehrfaches Füllen der Hash-Tabelle viele Daten sammeln, aber der Buffer Manager kann fast alles auslagern, wenn nötig.

Für die zweite Phase, Partition-Wise Aggregation, werden die thread-lokalen partitionierten Daten ausgetauscht, und jeder Thread kombiniert die Daten einer einzelnen Partition in eine Hash-Tabelle. Diese Phase ist größtenteils wie zuvor, außer dass wir jetzt manchmal deutlich mehr Partitionen als Threads erzeugen. Warum? Die Hash-Tabelle für eine Partition könnte in den Speicher passen, aber 8 Threads könnten gleichzeitig eine Partition kombinieren, und 8 Partitionen passen vielleicht nicht in den Speicher. Die einfache Lösung ist Over-Partitioning. Machen wir mehr Partitionen als Threads, zum Beispiel 32 Partitionen, werden die Partitionen kleiner, und die 8 Threads kombinieren gleichzeitig nur 8 von 32 Partitionen, was deutlich weniger Speicher braucht.

Experimente

Aggregationen, die nur wenige eindeutige Gruppen ergeben, passen leicht in den Speicher. Um unsere Implementierung der externen Hash-Aggregation zu bewerten, brauchen wir Aggregationen mit vielen eindeutigen Gruppen. Dazu nutzen wir den H2O.ai database-like ops Benchmark, den wir wiederbelebt haben und jetzt pflegen. Konkret nutzen wir die Datei G1_1e9_2e0_0_0.csv.zst, die unkomprimiert 50 GB groß ist. Der Quellcode des H2O.ai-Benchmarks liegt auf GitHub. Sie können die Datei selbst von https://blobs.duckdb.org/data/G1_1e9_2e0_0_0.csv.zst herunterladen (18,8 GB komprimiert).

Wir nutzen die folgenden Abfragen aus dem Benchmark, um die Daten zu laden:

SET preserve_insertion_order = false;
CREATE TABLE y (
    id1 VARCHAR, id2 VARCHAR, id3 VARCHAR,
    id4 INTEGER, id5 INTEGER, id6 INTEGER,
    v1 INTEGER, v2 INTEGER, v3 FLOAT);
COPY y FROM 'G1_1e9_2e0_0_0.csv.zst' (FORMAT csv, AUTO_DETECT true);
CREATE TYPE id1ENUM AS ENUM (SELECT id1 FROM y);
CREATE TYPE id2ENUM AS ENUM (SELECT id2 FROM y);
CREATE TABLE x (
    id1 id1ENUM, id2 id2ENUM, id3 VARCHAR,
    id4 INTEGER, id5 INTEGER, id6 INTEGER,
    v1 INTEGER, v2 INTEGER, v3 FLOAT);
INSERT INTO x (SELECT * FROM y);
DROP TABLE IF EXISTS y;

Der H2O.ai-Aggregationsbenchmark besteht aus 10 Abfragen, die in der Zahl eindeutiger Gruppen variieren:

-- Query 1: ~100 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id1, sum(v1) AS v1
FROM x
GROUP BY id1;

-- Query 2: ~10,000 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id1, id2, sum(v1) AS v1
FROM x
GROUP BY id1, id2;

-- Query 3: ~10,000,000 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id3, sum(v1) AS v1, avg(v3) AS v3
FROM x
GROUP BY id3;

-- Query 4: ~100 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id4, avg(v1) AS v1, avg(v2) AS v2, avg(v3) AS v3
FROM x
GROUP BY id4;

-- Query 5: ~1,000,000 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id6, sum(v1) AS v1, sum(v2) AS v2, sum(v3) AS v3
FROM x
GROUP BY id6;

-- Query 6: ~10,000 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT
    id4,
    id5,
    quantile_cont(v3, 0.5) AS median_v3,
    stddev(v3) AS sd_v3
FROM x
GROUP BY id4, id5;

-- Query 7: ~10,000,000 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id3, max(v1) - min(v2) AS range_v1_v2
FROM x
GROUP BY id3;

-- Query 8: ~10,000,000 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id6, v3 AS largest2_v3
FROM (
    SELECT id6, v3, row_number() OVER (
          PARTITION BY id6
          ORDER BY v3 DESC) AS order_v3
    FROM x
    WHERE v3 IS NOT NULL) sub_query
WHERE order_v3 <= 2;

-- Query 9: ~10,000 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id2, id4, pow(corr(v1, v2), 2) AS r2
FROM x
GROUP BY id2, id4;

-- Query 10: ~1,000,000,000 unique groups
CREATE OR REPLACE TABLE ans AS
SELECT id1, id2, id3, id4, id5, id6, sum(v3) AS v3, count(*) AS count
FROM x
GROUP BY id1, id2, id3, id4, id5, id6;

Die Ergebnisse auf der Benchmark-Seite stammen von der AWS-EC2-Instanz c6id.metal. Auf dieser Instanz passen alle Abfragen leicht in den Speicher, und viele Threads schaden der Performance auch nicht. DuckDB braucht nur 8,58 Sekunden, selbst für die größte Abfrage, Query 10, die 1 Milliarde eindeutige Gruppen zurückgibt. Viele Menschen werden aber keine so kräftige Maschine zum Rechnen nutzen. Auf meinem Laptop, einem MacBook Pro 2020, passen manche kleineren Abfragen in den Speicher, etwa Query 1, Query 10 aber definitiv nicht.

Die folgende Tabelle fasst die genutzte Hardware zusammen.

Obwohl die CPU-Kerne der AWS-EC2-Instanz nicht direkt mit denen meines Laptops vergleichbar sind, hat die Instanz klar deutlich mehr Rechenleistung und Speicher. Trotz der großen Hardwareunterschiede kann DuckDB alle 10 Abfragen ohne Problem beenden:

Die Laufzeit der Abfragen ist in Sekunden angegeben und wurde als Median von 3 Läufen auf meinem Laptop mit DuckDB 0.10.1 ermittelt. Die Ergebnisse der Instanz c6id.metal stammen von der Benchmark-Website. Obwohl nicht alle eindeutigen Gruppen in den Speicher meines Laptops passen, kann DuckDB alle eindeutigen Gruppen berechnen und zurückgeben. Die größte Abfrage, Query 10, braucht fast 4,5 Minuten. Das ist über 30× länger als mit der kräftigen Instanz c6id.metal. Der große Unterschied erklärt sich natürlich durch die großen Hardwareunterschiede. Interessant: Das ist immer noch schneller als Spark auf der Instanz c6id.metal, das 603,05 Sekunden braucht!

Fazit

DuckDB verbessert ständig seine Fähigkeiten zur Larger-than-Memory-Query-Verarbeitung. In diesem Blogpost haben wir einige Tricks gezeigt, die DuckDB zum Auslagern und Laden von Daten aus dem Speicher nutzt. Diese Tricks stecken in DuckDBs externer Hash-Aggregation, die seit 0.9.0 veröffentlicht ist. Wir haben die Hash-Aggregation auf dem H2O.ai-Benchmark ausprobiert, und DuckDB konnte alle 50-GB-Abfragen auf einem Laptop mit nur 16 GB Speicher beenden.

Lust auf mehr? Lesen Sie unser Paper zur externen Aggregation.