2021-08-27

Die schnellste Tabellensortierung im Westen – DuckDBs Sort neu entworfen

Laurens Kuiper

Datenbanksysteme nutzen Sortieren für viele Zwecke, der offensichtlichste, wenn ein Nutzer eine ORDER BY-Klausel zu seiner Abfrage hinzufügt. Sortieren wird auch innerhalb von Operatoren genutzt, etwa Window-Funktionen. DuckDB hat kürzlich seine Sortierimplementierung verbessert, die jetzt Daten parallel sortieren und mehr Daten sortieren kann, als in den Speicher passen. In diesem Beitrag schauen wir uns an, wie DuckDB sortiert und wie das im Vergleich zu anderen Datenverwaltungssystemen dasteht.

Nicht an der Implementierung interessiert? Springen Sie direkt zu den Experimenten!

Relationale Daten sortieren

Sortieren ist eines der am besten untersuchten Probleme der Informatik und ein wichtiger Aspekt der Datenverwaltung. Es gibt ganze Communities, die sich damit beschäftigen, wer am schnellsten sortiert. Forschung zu Sortieralgorithmen konzentriert sich oft auf das Sortieren großer Arrays oder Schlüssel/Wert-Paare. Das ist wichtig, deckt aber nicht ab, wie Sortieren in einem Datenbanksystem umzusetzen ist. Es gehört viel mehr zum Sortieren von Tabellen als nur ein großes Integer-Array zu sortieren!

Betrachten Sie die folgende Beispielabfrage auf einem Ausschnitt einer TPC-DS-Tabelle:

SELECT c_customer_sk, c_birth_country, c_birth_year
FROM customer
ORDER BY c_birth_country DESC,
         c_birth_year    ASC NULLS LAST;

Die ergibt:

Mit anderen Worten: c_birth_country ist absteigend geordnet, und wo c_birth_country gleich ist, sortieren wir nach c_birth_year aufsteigend. Durch die Angabe NULLS LAST werden Nullwerte als der niedrigste Wert in der Spalte c_birth_year behandelt. Ganze Zeilen werden also umgeordnet, nicht nur die Spalten in der ORDER BY-Klausel. Die Spalten, die nicht in der ORDER BY-Klausel stehen, nennen wir „Payload-Spalten“. Daher muss die Payload-Spalte c_customer_sk ebenfalls umgeordnet werden.

Es ist leicht, etwas umzusetzen, das die Beispielabfrage mit jeder Sortierimplementierung auswerten kann, zum Beispiel __C++__s std::sort. std::sort ist algorithmisch hervorragend, bleibt aber ein einthreadiger Ansatz, der nicht effizient nach mehreren Spalten sortieren kann, weil Funktionsaufruf-Overhead die Sortierzeit schnell dominieren würde. Unten diskutieren wir, warum das so ist.

Um gute Leistung beim Sortieren von Tabellen zu erreichen, ist eine eigene Sortierimplementierung nötig. Wir sind – natürlich – nicht die Ersten, die relationales Sortieren umsetzen, also haben wir uns in der Literatur umgesehen.

2006 schrieb der berühmte Goetz Graefe einen Überblick über die Umsetzung von Sortieren in Datenbanksystemen. In diesem Überblick sammelte er viele Sortiertechniken, die der Community bekannt sind. Das ist eine großartige Leitlinie, wenn Sie gerade dabei sind, Sortieren für Tabellen umzusetzen.

Die Kosten des Sortierens werden vom Vergleichen von Werten und vom Herumschieben von Daten dominiert. Alles, was diese beiden Operationen günstiger macht, hat großen Einfluss auf die Gesamtlaufzeit.

Es gibt zwei offensichtliche Wege, einen Comparator umzusetzen, wenn wir mehrere ORDER BY-Klauseln haben:

1. Durch die Klauseln loopen: Spalten vergleichen, bis wir eine finden, die nicht gleich ist, oder bis wir alle Spalten verglichen haben. Das ist schon recht komplex, weil das eine Schleife mit einem if/else darin für jede einzelne Datenzeile erfordert. Haben wir spaltenweisen Speicher, muss dieser Comparator zwischen Spalten springen, was wahlfreien Speicherzugriff verursacht.
2. Die Daten vollständig nach der ersten Klausel sortieren, dann nach der zweiten Klausel, aber nur wo die erste Klausel gleich war, und so weiter. Dieser Ansatz ist besonders ineffizient, wenn es viele Duplikate gibt, weil er mehrere Durchläufe über die Daten erfordert.

Binärer String-Vergleich

Die Technik des binären String-Vergleichs verbessert die Sortierleistung, indem sie den Comparator vereinfacht. Sie kodiert alle Spalten in der ORDER BY-Klausel in eine einzelne binäre Sequenz, die beim Vergleich mit memcmp die korrekte Gesamtsortierordnung ergibt. Das Kodieren der Daten ist nicht umsonst, aber weil wir den Comparator beim Sortieren so oft nutzen, zahlt es sich aus. Schauen wir uns noch einmal 3 Zeilen des Beispiels an:

Auf Little-Endian-Hardware sehen die Bytes, die diese Werte darstellen, so im Speicher aus, angenommen 32-Bit-Integer-Darstellung für das Jahr:

c_birth_country
-- NETHERLANDS
01001110 01000101 01010100 01001000 01000101 01010010 01001100 01000001 01001110 01000100 01010011 00000000
-- GERMANY
01000111 01000101 01010010 01001101 01000001 01001110 01011001 00000000

c_birth_year
-- 1991
11000111 00000111 00000000 00000000
-- 1992
11001000 00000111 00000000 00000000
-- 1924
10000100 00000111 00000000 00000000

Der Trick ist, diese in einen Binärstring umzuwandeln, der die Sortierordnung kodiert:

-- NETHERLANDS | 1991
10110001 10111010 10101011 10110111 10111010 10101101 10110011 10111110 10110001 10111011 10101100 11111111
10000000 00000000 00000111 11000111
-- NETHERLANDS | 1992
10110001 10111010 10101011 10110111 10111010 10101101 10110011 10111110 10110001 10111011 10101100 11111111
10000000 00000000 00000111 11001000
-- GERMANY     | 1924
10111000 10111010 10101101 10110010 10111110 10110001 10100110 11111111 11111111 11111111 11111111 11111111
10000000 00000000 00000111 10000100

Der Binärstring ist festgroß, weil das das Herumschieben beim Sortieren viel einfacher macht.

Der String „GERMANY“ ist kürzer als „NETHERLANDS“, daher wird er mit 00000000s aufgefüllt. Alle Bits in der Spalte c_birth_country werden anschließend invertiert, weil diese Spalte absteigend sortiert wird. Ist ein String zu lang, kodieren wir sein Prefix und schauen uns den ganzen String nur an, wenn die Prefixes gleich sind.

Die Bytes in c_birth_year werden getauscht, weil wir die Big-Endian-Darstellung brauchen, um die Sortierordnung zu kodieren. Das erste Bit wird ebenfalls geflippt, um die Ordnung zwischen positiven und negativen Integern für vorzeichenbehaftete Integer zu bewahren. Gibt es NULL-Werte, müssen diese mit einem zusätzlichen Byte kodiert werden (im Beispiel nicht gezeigt).

Mit diesem Binärstring können wir jetzt beide Spalten gleichzeitig vergleichen, indem wir nur die Binärstring-Darstellung vergleichen. Das geht mit einem einzelnen memcmp in C++! Der Compiler emittiert effiziente Assembler für den einzelnen Funktionsaufruf und erzeugt sogar automatisch SIMD-Anweisungen.

Diese Technik löst eines der oben genannten Probleme, nämlich den Funktionsaufruf-Overhead bei komplexen Comparators.

Radix Sort

Jetzt, da wir einen günstigen Comparator haben, müssen wir unseren Sortieralgorithmus wählen. Jeder Informatikstudent lernt über vergleichsbasierte Sortieralgorithmen wie Quicksort und Merge Sort, die O (n log n) Zeitkomplexität haben, wobei n die Zahl der zu sortierenden Datensätze ist.

Es gibt jedoch auch verteilungsbasierte Sortieralgorithmen, die typischerweise eine Zeitkomplexität von O (n k) haben, wobei k die Breite des Sortierschlüssels ist. Diese Klasse von Sortieralgorithmen skaliert viel besser mit größerem n, weil k konstant ist, log n aber nicht.

Ein solcher Algorithmus ist Radix Sort. Dieser Algorithmus sortiert die Daten, indem er die Datenverteilung mit Counting Sort mehrfach berechnet, bis alle Ziffern gezählt wurden.

Es mag kontraintuitiv klingen, die Sortierschlüsselspalten so zu kodieren, dass wir einen günstigen Comparator haben, und dann einen Sortieralgorithmus zu wählen, der Datensätze nicht vergleicht. Die Kodierung ist für Radix Sort jedoch nötig: Binärstrings, die mit memcmp eine korrekte Ordnung erzeugen, erzeugen eine korrekte Ordnung, wenn wir ein Byte-für-Byte-Radix-Sort machen.

Zweiphasiges paralleles Sortieren

DuckDB nutzt Morsel-Driven Parallelism, ein Framework für parallele Query-Ausführung. Für den Sortieroperator bedeutet das, dass mehrere Threads grob eine gleiche Datenmenge parallel aus der Tabelle sammeln.

Wir nutzen diese Parallelität zum Sortieren, indem zuerst jeder Thread die Daten, die er sammelt, mit unserem Radix Sort sortiert. Nach dieser ersten Sortierphase hat jeder Thread einen oder mehrere sortierte Datenblöcke, die zum finalen sortierten Ergebnis kombiniert werden müssen. Merge Sort ist der Algorithmus der Wahl für diese Aufgabe. Es gibt zwei Hauptwege, Merge Sort umzusetzen: K-Way Merge und Cascade Merge.

K-Way Merge merged K Listen in einem Durchlauf zu einer sortierten Liste und wird traditionell für externes Sortieren (Sortieren von mehr Daten, als in den Speicher passen) genutzt, weil es I/O minimiert. Cascade Merge merged jeweils zwei Listen sortierter Daten, bis nur eine sortierte Liste übrig bleibt, und wird für In-Memory-Sortieren genutzt, weil es effizienter ist als K-Way Merge. Wir wollen eine Implementierung, die hohe In-Memory-Leistung hat und elegant nachlässt, wenn wir das Limit des verfügbaren Speichers überschreiten. Daher wählen wir Cascade Merge.

Bei einem Cascade Merge Sort mergen wir jeweils zwei Blöcke sortierter Daten, bis nur ein sortierter Block übrig bleibt. Natürlich wollen wir alle verfügbaren Threads nutzen, um das Merge zu berechnen. Haben wir deutlich mehr sortierte Blöcke als Threads, können wir jedem Thread zwei Blöcke zum Mergen zuweisen. Wenn die Blöcke jedoch gemerged werden, haben wir nicht genug Blöcke, um alle Threads beschäftigt zu halten. Das ist besonders langsam, wenn die letzten beiden Blöcke gemerged werden: Ein Thread muss alle Daten verarbeiten.

Um diese Phase vollständig zu parallelisieren, haben wir Merge Path von Oded Green et al. umgesetzt. Merge Path berechnet vorab wo sich die sortierten Listen beim Mergen schneiden, gezeigt im Bild unten (aus dem Paper).

Die Schnittpunkte entlang des Merge Path können effizient mit Binärer Suche berechnet werden. Wenn wir wissen, wo die Schnittpunkte sind, können wir Partitionen der sortierten Daten unabhängig parallel mergen. Das erlaubt uns, alle verfügbaren Threads wirksam für die gesamte Merge-Phase zu nutzen. Für einen weiteren Trick zur Verbesserung von Merge Sort siehe den Anhang.

Spalten oder Zeilen?

Neben Vergleichen sind die anderen großen Kosten des Sortierens das Herumschieben von Daten. DuckDB hat eine vektorisierte Ausführungsengine. Daten werden in einem spaltenweisen Layout gespeichert, das in Batches (Chunks genannt) auf einmal verarbeitet wird. Dieses Layout ist großartig für analytische Query-Verarbeitung, weil die Chunks in den CPU-Cache passen und dem Compiler viele Gelegenheiten geben, SIMD-Anweisungen zu erzeugen. Wenn die Tabelle sortiert wird, werden jedoch ganze Zeilen umgeschaufelt, nicht Spalten.

Wir könnten beim Sortieren beim spaltenweisen Layout bleiben: Die Schlüsselspalten sortieren, dann die Payload-Spalten eine nach der anderen umordnen. Das Umordnen verursacht jedoch ein wahlfreies Zugriffsmuster im Speicher für jede Spalte. Gibt es viele Payload-Spalten, wird das langsam sein. Die Spalten in Zeilen umzuwandeln macht das Umordnen von Zeilen viel einfacher. Diese Umwandlung ist natürlich nicht umsonst: Spalten müssen in Zeilen kopiert werden und nach dem Sortieren wieder von Zeilen in Spalten.

Weil wir externes Sortieren unterstützen wollen, müssen wir Daten in buffer-verwalteten Blöcken speichern, die auf die Platte ausgelagert werden können. Weil wir die Eingabedaten ohnehin in diese Blöcke kopieren müssen, ist die Umwandlung der Zeilen in Spalten effektiv umsonst.

Es gibt ein paar Operatoren, die von Natur aus zeilenbasiert sind, etwa Joins und Aggregationen. DuckDB hat ein einheitliches internes Zeilenlayout für diese Operatoren, und wir haben uns entschieden, es auch für den Sortieroperator zu nutzen. Dieses Layout wurde bisher nur im Speicher genutzt. Im nächsten Abschnitt erklären wir, wie wir es auch auf der Platte zum Laufen gebracht haben. Wir sollten anmerken, dass wir Sortierdaten nur auf die Platte schreiben, wenn der Hauptspeicher sie nicht halten kann.

Externes Sortieren

Der Buffer Manager kann Blöcke vom Speicher auf die Platte ausladen. Das ist nichts, was wir in unserer Sortierimplementierung aktiv tun, sondern etwas, das der Buffer Manager entscheidet, wenn der Speicher sonst volllaufen würde. Er nutzt eine Least-Recently-Used-Queue, um zu entscheiden, welche Blöcke geschrieben werden. Mehr dazu, wie man diese Queue richtig nutzt, im Anhang.

Wenn wir einen Block brauchen, „pinnen“ wir ihn, was ihn von der Platte liest, wenn er noch nicht geladen ist. Plattenzugriff ist viel langsamer als Speicherzugriff, daher ist es entscheidend, die Zahl der Reads und Writes zu minimieren.

Daten auf die Platte auszuladen ist einfach für festgroße Spalten wie Integer, aber schwieriger für variabel große Spalten wie Strings. Unser Zeilenlayout nutzt festgroße Zeilen, die Strings beliebiger Größe nicht aufnehmen können. Daher werden Strings durch einen Pointer dargestellt, der in einen separaten Speicherblock zeigt, in dem die tatsächlichen String-Daten leben, ein sogenannter „String Heap“.

Wir haben unseren Heap so geändert, dass er Strings ebenfalls zeilenweise in buffer-verwalteten Blöcken speichert:

Jede Zeile hat ein zusätzliches 8-Byte-Feld pointer, das auf den Beginn dieser Zeile im Heap zeigt. Das ist in der In-Memory-Darstellung nutzlos, aber wir werden gleich sehen, warum es für die On-Disk-Darstellung nützlich ist.

Passen die Daten in den Speicher, bleiben die Heap-Blöcke gepinnt, und nur die festgroßen Zeilen werden beim Sortieren umgeordnet. Passen die Daten nicht in den Speicher, müssen die Blöcke auf die Platte ausgelagert werden, und der Heap wird beim Sortieren ebenfalls umgeordnet. Wenn ein Heap-Block auf die Platte ausgelagert wird, werden die Pointer, die hineinzeigen, ungültig. Wenn wir den Block zurück in den Speicher laden, haben sich die Pointer geändert.

Hier kommt unser zeilenweises Layout ins Spiel. Das 8-Byte-Feld pointer wird mit einem 8-Byte-Feld offset überschrieben, das angibt, wo im Heap-Block Strings dieser Zeile zu finden sind. Diese Technik heißt „Pointer Swizzling“. Wenn wir die Pointer swizzlen, sehen Zeilenlayout und Heap-Block so aus:

Die Pointer auf die nachfolgenden String-Werte werden ebenfalls mit einem 8-Byte-Relativoffset überschrieben, der angibt, wie weit dieser String vom Beginn der Zeile im Heap versetzt ist (daher hat jedes stringA einen Offset von 0: Es ist der erste String in der Zeile). Relative Offsets innerhalb von Zeilen statt absoluter Offsets zu nutzen, ist beim Sortieren sehr nützlich, weil diese Relativoffsets konstant bleiben und beim Kopieren einer Zeile nicht aktualisiert werden müssen.

Wenn die Blöcke gescannt werden müssen, um das sortierte Ergebnis zu lesen, „unswizzlen“ wir die Pointer, sodass sie wieder auf den String zeigen.

Mit dieser Dual-Purpose-zeilenweisen Darstellung können wir sowohl die festgroßen Zeilen als auch die variabel großen Zeilen im Heap leicht herumkopieren. Neben dem Laden/Entladen von Blöcken durch den Buffer Manager ist der einzige Unterschied zwischen In-Memory- und externem Sortieren, dass wir Pointer auf die Heap-Blöcke swizzlen/unswizzlen und Daten aus den Heap-Blöcken während des Merge Sort kopieren.

All das reduziert Overhead, wenn Blöcke in und aus dem Speicher bewegt werden müssen, was zu elegantem Leistungsabfall führt, wenn wir uns dem Limit des verfügbaren Speichers nähern.

Vergleich mit anderen Systemen

Nachdem wir die meisten Techniken in unserer Sortierimplementierung behandelt haben, wollen wir wissen, wie wir im Vergleich zu anderen Systemen dastehen. DuckDB wird oft für interaktive Datenanalyse genutzt und daher oft mit Werkzeugen wie dplyr verglichen.

In diesem Setting laufen die Leute üblicherweise auf Laptops oder PCs, daher führen wir diese Experimente auf einem MacBook Pro 2020 aus. Dieses Laptop hat eine Apple-M1-CPU, die ARM-basiert ist. Der M1-Prozessor hat 8 Kerne: 4 High-Performance- (Firestorm-) Kerne und 4 energieeffiziente (Icestorm-) Kerne. Die Firestorm-Kerne haben sehr, sehr schnelle Einthread-Leistung, das sollte das Spielfeld zwischen ein- und mehrthreadigen Sortierimplementierungen etwas ebnen. Das MacBook hat 16 GB Speicher und eine der schnellsten SSDs, die man in einem Laptop findet.

Wir vergleichen mit den folgenden Systemen:

1. ClickHouse, Version 21.7.5
2. HyPer, Version 2021.2.1.12564
3. Pandas, Version 1.3.2
4. SQLite, Version 3.36.0

ClickHouse und HyPer sind in unserem Vergleich, weil sie analytische SQL-Engines mit Betonung auf Leistung sind. Pandas und SQLite sind in unserem Vergleich, weil sie genutzt werden können, um relationale Operationen innerhalb von Python auszuführen, wie DuckDB. Pandas arbeitet vollständig im Speicher, SQLite ist ein traditionelleres plattenbasiertes System. Diese Liste von Systemen sollte uns eine gute Mischung aus ein-/mehrthreadigem und In-Memory-/externem Sortieren geben.

ClickHouse wurde für M1 nach diesem Leitfaden gebaut. Wir haben das Speicherlimit auf 12 GB und max_bytes_before_external_sort auf 10 GB gesetzt, nach diesem Vorschlag.

HyPer ist Tableaus Datenengine, geschaffen von der Datenbankgruppe der Universität München. Es läuft (noch) nicht nativ auf ARM-basierten Prozessoren wie dem M1. Wir nutzen Rosetta 2, macOS’ x86-Emulator, um es auszuführen. Emulation verursacht etwas Overhead, daher haben wir ein Experiment auf einer x86-Maschine im Anhang aufgenommen.

Sortieren in Datenbanksystemen zu benchmarken ist nicht geradlinig. Idealerweise würden wir nur die Zeit messen, die das Sortieren der Daten braucht, nicht die Zeit zum Lesen der Eingabedaten und zum Anzeigen der Ausgabe. Nicht jedes System hat einen Profiler, um die Zeit des Sortieroperators genau zu messen, das ist also keine Option.

Um uns einem fairen Vergleich zu nähern, messen wir die End-to-End-Zeit von Abfragen, die die Daten sortieren und das Ergebnis in eine temporäre Tabelle schreiben, d. h.:

CREATE TEMPORARY TABLE output AS
SELECT ...
FROM ...
ORDER BY ...;

Es gibt keine perfekte Lösung für dieses Problem, das sollte uns aber einen guten Vergleich geben, weil die End-to-End-Zeit dieser Abfrage vom Sortieren dominiert sein sollte. Für Pandas nutzen wir sort_values mit inplace=False, um diese Abfrage nachzubilden.

In ClickHouse können temporäre Tabellen nur im Speicher existieren, was für unsere Out-of-Core-Experimente problematisch ist. Daher nutzen wir eine normale TABLE, müssen dann aber auch eine Table Engine wählen. Die meisten Table Engines wenden Kompression an oder erzeugen einen Index, was wir nicht messen wollen. Daher haben wir die einfachste On-Disk-Engine gewählt, File, mit Format Native.

Die Table Engine, die wir für die Eingabetabellen für ClickHouse gewählt haben, ist MergeTree mit ORDER BY tuple(). Wir haben das gewählt, weil wir seltsames Verhalten mit File(Native)-Eingabetabellen gesehen haben, bei dem es keinen Laufzeitunterschied zwischen den Abfragen SELECT * FROM ... ORDER BY und SELECT col1 FROM ... ORDER BY gab. Vermutlich, weil alle Spalten in der Tabelle sortiert wurden, unabhängig davon, wie viele ausgewählt waren.

Um stabile End-to-End-Abfragezeiten zu messen, führen wir jede Abfrage 5-mal aus und berichten die Median-Laufzeit. Es gibt einige Unterschiede beim Lesen/Schreiben von Tabellen zwischen den Systemen. Pandas kann zum Beispiel nicht von/auf die Platte lesen/schreiben, daher liegen sowohl der Eingabe- als auch der Ausgabe-DataFrame im Speicher. DuckDB schreibt die Ausgabetabelle nicht auf die Platte, außer es ist nicht genug Platz, sie im Speicher zu halten, und kann daher ebenfalls einen Vorteil haben. Das Sortieren dominiert jedoch die Gesamtlaufzeit, sodass diese Unterschiede nicht so wirkungsvoll sind.

Zufällige Integer

Wir beginnen mit einem einfachen Beispiel. Wir haben die ersten 100 Millionen Integer erzeugt und gemischt, und wir wollen wissen, wie gut die Systeme sie sortieren können. Dieses Experiment ist eher ein Mikrobenchmark als etwas anderes und hat wenig reale Bedeutung.

Für unser erstes Experiment schauen wir uns an, wie die Systeme mit der Zahl der Zeilen skalieren. Aus der Ausgangstabelle mit Integern haben wir 9 weitere Tabellen gemacht, mit jeweils 10M, 20M, …, 90M Integern.

Als traditionelles plattenbasiertes Datenbanksystem entscheidet sich SQLite immer für eine externe Sortierstrategie. Es schreibt zwischensortierte Blöcke auf die Platte, auch wenn sie in den Hauptspeicher passen, daher ist es viel langsamer. Die Leistung der anderen Systeme liegt in derselben Größenordnung, DuckDB und ClickHouse gehen Kopf an Kopf mit ~3 und ~4 Sekunden für 100M Integer. Weil SQLite so viel langsamer ist, nehmen wir es in unsere nächste Experimentreihe (TPC-DS) nicht auf.

DuckDB und ClickHouse nutzen beide alle verfügbaren Threads sehr gut, mit einem einthreadigen Sort parallel, gefolgt von einem parallelen Merge Sort. Wir sind uns nicht sicher, welche Strategie HyPer nutzt. Für unser nächstes Experiment zoomen wir auf Multithreading und sehen, wie gut ClickHouse und DuckDB mit der Zahl der Threads skalieren (wir konnten die Thread-Zahl für HyPer nicht setzen).

Dieser Plot zeigt, dass Radix Sort sehr schnell ist. DuckDB sortiert 100M Integer in knapp unter 5 Sekunden mit einem einzelnen Thread, deutlich schneller als ClickHouse. Das Hinzufügen von Threads verbessert die Leistung für DuckDB nicht so stark, weil Radix Sort so viel schneller ist als Merge Sort. Beide Systeme landen bei etwa derselben Leistung bei 4 Threads.

Jenseits von 4 Threads sehen wir die Leistung nicht viel weiter steigen, aufgrund der CPU-Architektur. Für alle anderen Experimente haben wir DuckDB und ClickHouse auf 4 Threads gesetzt.

Für unser letztes Experiment mit zufälligen Integern schauen wir, wie die Vorsortiertheit der Eingabe die Leistung beeinflussen kann. Das ist besonders wichtig in Systemen, die Quicksort nutzen, weil Quicksort auf invers sortierten Daten viel schlechter läuft als auf zufälligen Daten.

Nicht überraschend: Alle Systeme laufen auf sortierten Daten besser, manchmal mit großem Abstand. ClickHouse, Pandas und SQLite haben hier wahrscheinlich eine Optimierung: z. B. Vorsortiertheit im Katalog verfolgen oder Vorsortiertheit beim Scannen der Eingabe prüfen. DuckDB und HyPer haben nur einen sehr kleinen Leistungsunterschied, wenn die Eingabedaten sortiert sind, und haben keine solche Optimierung. Für DuckDB lässt sich die leicht verbesserte Leistung durch ein besseres Speicherzugriffsmuster beim Sortieren erklären: Wenn die Daten bereits sortiert sind, ist das Zugriffsmuster überwiegend sequenziell.

Ein weiteres interessantes Ergebnis: DuckDB sortiert Daten schneller, als manche der anderen Systeme bereits sortierte Daten lesen können.

TPC-DS

Für den nächsten Vergleich haben wir einen relationalen Sortierbenchmark auf zwei Tabellen aus dem Standard-TPC Decision Support Benchmark (TPC-DS) improvisiert. TPC-DS ist für Sortierimplementierungen anspruchsvoll, weil es breite Tabellen hat (mit vielen Spalten, anders als die Tabellen in TPC-H) und eine Mischung aus fest- und variabel großen Typen. Die Zahl der Zeilen steigt mit dem Scale Factor. Die hier genutzten Tabellen sind catalog_sales und customer.

catalog_sales hat 34 Spalten, alle festgroße Typen (Integer und Double), und wächst mit steigendem Scale Factor auf viele Zeilen. customer hat 18 Spalten (10 Integer und 8 Strings) und eine anständige Zahl von Zeilen mit steigendem Scale Factor. Die Zeilenzahlen beider Tabellen bei jedem Scale Factor sind in der Tabelle unten gezeigt.

Wir nutzen customer bei SF100 und SF300, das bei jedem Scale Factor in den Speicher passt. Wir nutzen die Tabelle catalog_sales bei SF10 und SF100, die bei SF100 nicht mehr in den Speicher passt.

Die Daten wurden mit DuckDBs TPC-DS-Erweiterung erzeugt, dann in zufälliger Reihenfolge nach CSV exportiert, um alle Ordnungsmuster rückgängig zu machen, die in den erzeugten Daten gewesen sein könnten.

Catalog Sales (numerische Typen)

Unser erstes Experiment auf der Tabelle catalog_sales wählt 1 Spalte, dann 2 Spalten, …, bis zu allen 34, immer geordnet nach cs_quantity und cs_item_sk. Dieses Experiment sagt uns, wie gut die verschiedenen Systeme Payload-Spalten umordnen können.

Wir sehen ähnliche Trends bei SF10 und SF100, aber bei SF100, bei etwa 12 Payload-Spalten oder so, passen die Daten nicht mehr in den Speicher, und ClickHouse und HyPer zeigen einen großen Leistungsabfall. ClickHouse wechselt zu einer externen Sortierstrategie, die viel langsamer ist als seine In-Memory-Strategie. Daher führt das Hinzufügen weniger Payload-Spalten zu einer Laufzeit, die um Größenordnungen höher ist. Bei 20 Payload-Spalten läuft ClickHouse in folgenden Fehler:

DB::Exception: Memory limit (for query) exceeded: would use 11.18 GiB (attempt to allocate chunk of 4204712 bytes), maximum: 11.18 GiB: (while reading column cs_list_price): (while reading from part ./store/523/5230c288-7ed5-45fa-9230-c2887ed595fa/all_73_108_2/ from mark 4778 with max_rows_to_read = 8192): While executing MergeTreeThread.

HyPer fällt ebenfalls in der Leistung ab, bevor es mit folgender Meldung fehlschlägt:

ERROR:  Cannot allocate 333982248 bytes of memory: The `global memory limit` limit of 12884901888 bytes was exceeded.

Soweit uns bekannt, nutzt HyPer mmap, das eine Abbildung zwischen Speicher und einer Datei erzeugt. Das erlaubt dem Betriebssystem, Daten zwischen Speicher und Platte zu bewegen. Nützlich, aber kein Ersatz für ein richtiges externes Sort, weil es wahlfreien Zugriff auf die Platte erzeugt, der sehr langsam ist.

Pandas läuft bei SF100 überraschend gut, obwohl die Daten nicht in den Speicher passen. Pandas kann das nur, weil macOS die Swap-Größe dynamisch erhöht. Die meisten Betriebssysteme tun das nicht und würden die Daten gar nicht laden können. Swap verlangsamt die Verarbeitung üblicherweise erheblich, aber die SSD ist so schnell, dass es keinen sichtbaren Leistungsabfall gibt!

Während Pandas die Daten lädt, wächst die Swap-Größe auf beeindruckende ~40 GB: Sowohl die Datei als auch der DataFrame sind vollständig im Speicher/Swap, statt in den Speicher gestreamt zu werden. Das geht auf ~20 GB Speicher/Swap runter, wenn das Lesen der Datei fertig ist. Pandas kommt recht weit ins Experiment, bis es mit folgendem Fehler abstürzt:

UserWarning: resource_tracker: There appear to be 1 leaked semaphore objects to clean up at shutdown

DuckDB läuft sowohl In-Memory als auch extern gut, und es gibt keinen klar sichtbaren Punkt, an dem die Daten nicht mehr in den Speicher passen: Die Laufzeit ist schnell und zuverlässig.

Customer (Strings & Integer)

Nachdem wir gesehen haben, wie die Systeme große Mengen festgroßer Typen handhaben, ist es Zeit, einige variabel große Typen zu sehen! Für unser erstes Experiment auf der Tabelle customer wählen wir alle Spalten und ordnen sie entweder nach 3 Integer-Spalten (c_birth_year, c_birth_month, c_birth_day) oder nach 2 String-Spalten (c_first_name, c_last_name). Strings zu vergleichen ist viel, viel schwieriger als Integer zu vergleichen, weil Strings variable Größen haben können und Byte für Byte verglichen werden müssen, während Integer immer denselben Vergleich haben.

Wie erwartet ist das Ordnen nach Strings teurer als das Ordnen nach Integern, außer bei HyPer, das beeindruckend ist. Pandas hat nur einen etwas größeren Unterschied zwischen dem Ordnen nach Integern und dem Ordnen nach Strings als ClickHouse und DuckDB. Dieser Unterschied erklärt sich durch einen teuren Comparator zwischen Strings. Pandas nutzt NumPys Sort, das effizient in C umgesetzt ist. Wenn das jedoch Strings sortiert, muss es virtuelle Funktionsaufrufe nutzen, um ein Python-String-Objekt zu vergleichen, was langsamer ist als ein einfaches „<“ zwischen Integern in C. Dennoch läuft Pandas auf der Tabelle customer gut.

Im nächsten Experiment schauen wir, wie der Payload-Typ die Leistung beeinflusst. customer hat 10 Integer-Spalten und 8 String-Spalten. Wir wählen entweder alle Integer-Spalten oder alle String-Spalten und ordnen jedes Mal nach (c_birth_year, c_birth_month, c_birth_day).

Wie erwartet braucht das Umordnen von Strings viel mehr Zeit als das Umordnen von Integern. Pandas hat hier einen Vorteil, weil es die Strings bereits im Speicher hat und höchstwahrscheinlich nur Pointer auf diese Strings umordnen muss. Die Datenbanksysteme müssen Strings zweimal kopieren: Einmal beim Lesen der Eingabetabelle und erneut beim Anlegen der Ausgabetabelle. Profiling in DuckDB zeigt, dass das tatsächliche Sortieren bei SF300 weniger als eine Sekunde braucht und die meiste Zeit mit dem (De-)Serialisieren von Strings verbracht wird.

Fazit

DuckDBs neue parallele Sortierimplementierung kann effizient mehr Daten sortieren, als in den Speicher passen, und nutzt die Geschwindigkeit moderner SSDs. Wo andere Systeme abstürzen, weil ihnen der Speicher ausgeht, oder zu einer externen Sortierstrategie wechseln, die viel langsamer ist, lässt DuckDBs Leistung elegant nach, wenn es das Speicherlimit überschreitet.

Der Code, der für die Experimente genutzt wurde, liegt auf GitHub. Wenn wir Fehler gemacht haben, lassen Sie es uns bitte wissen!

DuckDB ist ein freies und quelloffenes Datenbankmanagementsystem (MIT-Lizenz). Es will das SQLite für Analysen sein und bietet ein schnelles, effizientes Datenbanksystem ohne externe Abhängigkeiten. Es ist nicht nur für Python verfügbar, sondern auch für C/C++, R, Java und mehr.

Diesen Beitrag auf Hacker News diskutieren

Unser Paper zu Sortieren auf der ICDE ’23 lesen

Hören Sie Laurens’ Auftritt im Disseminate-Podcast:

• Spotify
• Google
• Apple

Anhang A: Predication

Eine weitere Technik, die wir genutzt haben, um Merge Sort zu beschleunigen, ist Predication. Mit dieser Technik machen wir Code mit if/else-Zweigen zu Code ohne Zweige. Moderne CPUs versuchen vorherzusagen, ob der if- oder der else-Zweig genommen wird. Ist das schwer vorherzusagen, kann das den Code verlangsamen. Schauen Sie sich das Beispiel von Pseudo-Code mit Zweigen unten an.

// continue until merged
while (l_ptr && r_ptr) {
  // check which side is smaller
  if (memcmp(l_ptr, r_ptr, entry) < 0) {
    // copy from left side and advance
    memcpy(result_ptr, l_ptr, entry);
    l_ptr += entry;
  } else {
    // copy from right side and advance
    memcpy(result_ptr, r_ptr, entry);
    r_ptr += entry;
  }
  // advance result
  result_ptr += entry;
}

Wir mergen die Daten aus den linken und rechten Blöcken in einen Ergebnisblock, einen Eintrag nach dem anderen, indem wir Pointer vorrücken. Dieser Code kann branchless gemacht werden, indem das Vergleichs-Boolean als 0 oder 1 genutzt wird, gezeigt im Pseudo-Code unten.

// continue until merged
while (l_ptr && r_ptr) {
  // store comparison result in a bool
  bool left_less = memcmp(l_ptr, r_ptr, entry) < 0;
  bool right_less = 1 - left_less;
  // copy from either side
  memcpy(result_ptr, l_ptr, left_less * entry);
  memcpy(result_ptr, r_ptr, right_less * entry);
  // advance either one
  l_ptr += left_less * entry;
  l_ptr += right_less * entry;
  // advance result
  result_ptr += entry;
}

Wenn left_less wahr ist, ist es gleich 1. Das bedeutet, right_less ist falsch und daher gleich 0. Wir nutzen das, um entry Bytes von der linken Seite und 0 Bytes von der rechten Seite zu kopieren und die linken und rechten Pointer entsprechend zu erhöhen.

Mit predicated Code muss die CPU nicht vorhersagen, welche Anweisungen auszuführen sind, das bedeutet weniger Instruction-Cache-Misses!

Anhang B: Zickzack

Ein einfacher Trick, I/O zu reduzieren, ist das Zickzack-Durchlaufen der zu mergenden Blockpaare im kaskadierten Merge Sort. Das ist im Bild unten illustriert (gestrichelte Pfeile zeigen an, in welcher Reihenfolge die Blöcke gemerged werden).

Indem wir zickzack durch die Blöcke gehen, beginnen wir eine Iteration, indem wir die letzten Blöcke mergen, die in der vorherigen Iteration gemerged wurden. Diese Blöcke sind wahrscheinlich noch im Speicher, was uns einige kostbare Read/Write-Operationen spart.

Anhang C: x86-Experiment

Wir haben das Experiment catalog_sales SF100 auch auf einer Maschine mit x86-CPU-Architektur ausgeführt, um einen faireren Vergleich mit HyPer zu bekommen (ohne Rosetta-2-Emulation). Die Maschine hat eine Intel(R) Xeon(R) W-2145 CPU @ 3.70 GHz mit 8 Kernen (bis zu 16 virtuellen Threads) und 128 GB RAM, diesmal passen die Daten also vollständig in den Speicher. Wir haben die Zahl der Threads, die DuckDB und ClickHouse nutzen, auf 8 gesetzt, weil wir jenseits von 8 keine sichtbare Leistungsverbesserung gesehen haben.

Pandas läuft im Vergleich zum MacBook schlechter, weil es eine einthreadige Implementierung hat und diese CPU eine niedrigere Einthread-Leistung hat. Wieder stürzt Pandas mit einem Fehler ab (diese Maschine erhöht Swap nicht dynamisch):

numpy.core._exceptions.MemoryError: Unable to allocate 6.32 GiB for an array with shape (6, 141430723) and data type float64

DuckDB, HyPer und ClickHouse nutzen alle die mehr verfügbaren Threads gut und sind deutlich schneller als auf dem MacBook.

Ein interessantes Muster in diesem Plot: DuckDB und HyPer skalieren sehr ähnlich mit zusätzlichen Payload-Spalten. Obwohl DuckDB beim Sortieren schneller ist, scheint das Umordnen der Payload für beide Systeme etwa dasselbe zu kosten. Daher ist es wahrscheinlich, dass HyPer ebenfalls ein Zeilenlayout nutzt.

ClickHouse skaliert schlechter mit zusätzlichen Payload-Spalten. ClickHouse nutzt kein Zeilenlayout und muss daher die Kosten wahlfreien Zugriffs zahlen, wenn jede Spalte nach dem Sortieren umgeordnet wird.