2025-09-24

DuckDBs Sort erneut neu entworfen

Laurens Kuiper

DuckDB v1.4.0 ist gerade erschienen und enthält einen kompletten Redesign von DuckDBs Sort-Implementierung. Wir haben DuckDBs Sort vor gerade vier Jahren neu entworfen, sodass DuckDB mehr Daten sortieren konnte, als in den Hauptspeicher passen, parallel, mit hocheffizienten Vergleichen. Diese Implementierung hat uns gut gedient, seitdem haben wir aber Larger-than-Memory-Query-Processing für weitere Operatoren umgesetzt, etwa den Hash Join und die Hash Aggregation, die beide ein neues, verbessertes spillbares Page-Layout nutzen. Dieses Layout haben wir in einem früheren Blogbeitrag vorgestellt. Wir haben uns entschieden, dieses Layout in DuckDBs Sort zu integrieren und die Implementierung komplett neu zu entwerfen.

Kein Interesse an der Implementierung? Springen Sie direkt zum Benchmark!

Two-Phase Sorting

DuckDB setzt parallele Query-Ausführung mit Morsel-Driven Parallelism um. In DuckDBs Implementierung dieses Frameworks haben Blocking-Operatoren, also Operatoren, die die gesamte Eingabe lesen müssen, bevor sie ausgeben können, wie Hash Aggregation und Sort, die folgenden Phasen:

1. Sink: Thread-lokale Akkumulation von Daten aus einer Pipeline
2. Combine: Signalisiert, dass ein Thread seine Sink-Phase beendet
3. Finalize: Einmal aufgerufen, wenn alle Threads Combine aufgerufen haben
4. GetData: Daten an die nächste Pipeline ausgeben

Seit vielen Jahrzehnten ist die bevorzugte Option, Larger-than-Memory-Sorting in Datenbanksystemen umzusetzen, mehrere sortierte Runs zu erzeugen, gefolgt von einem Merge Sort. Konkret erzeugt ein k-way Merge Sort die geringste I/O-Menge beim Larger-than-Memory-Sorting, was für die Performance kritisch ist. Dieser Ansatz mappt gut auf Morsel-Driven Parallelism: DuckDB führt thread-lokales Sortieren in der Sink-Phase aus, gefolgt von einem parallelen Merge Sort in der Finalize- oder GetData-Phase. Beide wurden für DuckDB v1.4.0 neu entworfen. Wir diskutieren zuerst die neue thread-lokale Sort-Implementierung, bevor wir das neue Merge-Design vorstellen.

Thread-lokales Sortieren

Sortierte Runs werden thread-lokal in der Sink-Phase erzeugt. Die Art, wie DuckDB das parallelisiert, hat sich in v1.4.0 nicht geändert: Threads erzeugen sortierte Runs unabhängig, parallel. Geändert hat sich die physische Sort-Implementierung.

Key Normalization

Datenbanksysteme, die die benötigten Typen nicht in den Query-Plan kompilieren – z. B. DuckDB – leiden unter Interpretation Overhead, besonders beim Vergleichen von Tuples beim Sortieren. Ein Weg, das zu umgehen, ist Key Normalization. DuckDBs Sort nutzte schon vor v1.4.0 eine Ad-hoc-Version davon, die neue Implementierung nutzt aber die generischere Funktion create_sort_key, die über SQL verfügbar ist.

Diese Funktion nimmt eine beliebige Zahl von Eingaben und Sort-Bedingungen und konstruiert ein BLOB-Feld, das die angegebene Ordnung erzeugt. Ein Beispiel aus der Beschreibung des PRs, der create_sort_key umgesetzt hat:

SELECT
    s,
    create_sort_key(s, 'asc nulls last') AS k1,
    create_sort_key(s, 'asc nulls first') AS k2
FROM
    (VALUES ('hello'), ('world'), (NULL)) t(s);

┌─────────┬───────────────┬───────────────┐
│    s    │      k1       │      k2       │
│ varchar │     blob      │     blob      │
├─────────┼───────────────┼───────────────┤
│ hello   │ \x01ifmmp\x00 │ \x02ifmmp\x00 │
│ world   │ \x01xpsme\x00 │ \x02xpsme\x00 │
│ NULL    │ \x02          │ \x01          │
└─────────┴───────────────┴───────────────┘

Wegen der binär-vergleichbaren Natur des konstruierten BLOBs sind die folgenden Queries äquivalent:

SELECT * FROM tbl
ORDER BY x DESC NULLS LAST, y ASC NULLS FIRST;

SELECT * FROM tbl
ORDER BY create_sort_key(x, 'DESC NULLS LAST', y, 'ASC NULLS FIRST');

Das behebt das Problem des Interpretation Overheads beim Vergleichen von Tuples, weil wir jetzt nur noch BLOBs vergleichen müssen statt beliebiger Typkombinationen in einer ORDER BY-Klausel.

Statische Integer-Vergleiche

Es ist bekannt, dass die Verarbeitung von Strings deutlich langsamer ist als die Verarbeitung fester Typen wie Integers. Würden wir immer die Funktion create_sort_key nutzen, auch für Integers, würden wir viel Performance liegen lassen. Wenn wir aber die Größe des resultierenden BLOB kennen, können wir es zurück in ein oder mehrere unsigned Integers wandeln und Integer-Vergleiche nutzen.

Haben wir zum Beispiel die folgende Query:

SELECT *
FROM tbl
ORDER BY
    c0::INTEGER ASC NULLS LAST,
    c1::DOUBLE ASC NULLS LAST;

Das resultierende BLOB aus create_sort_key(c0::INTEGER, 'ASC NULLS LAST', c1::DOUBLE, 'ASC NULLS LAST') ist kleiner als 16 Bytes, deshalb tauscht die neue Sort-Implementierung die Bytes (für Big-Endian-Integer-Vergleiche) und speichert sie in zwei 64-Bit-Unsigned-Integers. Eine vereinfachte Version der Datenstruktur, die wir in C++ nutzen:

struct FixedSortKeyNoPayload {
    uint64_t part0;
    uint64_t part1;
};
struct FixedSortKeyPayload {
    uint64_t part0;
    uint64_t part1;
    data_ptr_t payload;
};

Die sich so vergleichen lässt:

bool LessThan(const FixedSortKeyPayload &lhs, const FixedSortKeyPayload &rhs) {
    return lhs.part0 < rhs.part0 || (lhs.part0 == rhs.part0 && lhs.part1 < rhs.part1);
}

Das Feld payload ist nur vorhanden, wenn mehr Spalten selektiert werden, d. h.:

SELECT ⟨many columns⟩
FROM tbl
ORDER BY ⟨a few columns⟩;

Werden nur Spalten selektiert, die auch in der ORDER BY-Klausel vorkommen, wird das Feld payload nicht gebraucht, weil DuckDB die normalisierten Keys decoden kann.

Nicht-kontinuierliche Iteration

Vor v1.4.0 nutzte DuckDB Sort Keys fester Größe, ihre Größe war aber erst zur Query-Ausführung bekannt. Das macht dynamisches Vergleichen und Verschieben von Sort Keys beim Sortieren nötig, was deutlich weniger effizient ist als statisch kompilierter Code. Das C++-struct, das DuckDB jetzt nutzt, oben gezeigt, ist zur Compile-Zeit bekannt, sodass es mit Sortieralgorithmen sortiert werden kann, die das C++-std::iterator-Interface umsetzen. Das heißt, DuckDB muss keinen Sortieralgorithmus mehr selbst implementieren: Es kann eine fertige C++-Implementierung nehmen!

C++ kommt mit std::iterator-Implementierungen für verschiedene Datenstrukturen wie std::array und std::vector. Diese Datenstrukturen erfordern aber, alle Daten in einem zusammenhängenden Speicherblock zu halten. DuckDB nutzt eine Standard-Page-Allokation (= zusammenhängender Speicherblock) von 256 KiB. Das oben gezeigte FixedSortKeyPayload ist 24 Bytes, also passen nur etwa 10k Tuples in eine Page. Wir wollen, dass sortierte Runs deutlich länger sind (aus Performancegründen, auf die wir in diesem Beitrag nicht eingehen). Um längere sortierte Runs zu erzeugen, haben wir einen std::iterator umgesetzt, der über nicht-zusammenhängende Speicherblöcke iterieren kann:

Dieser Iterator ist großartig für sequenziellen Zugriff, manche Sortieralgorithmen brauchen aber Random Access. Mit diesem Design können wir nicht einfach einen Offset zu einem Pointer addieren, um die Adresse eines Tuples zu bekommen. Stattdessen berechnen wir den Page-Index und den Offset innerhalb der Page mit Integer-Division/Modulo, weil die Zahl der Tuples pro Page immer gleich ist (außer der letzten Page). Integer-Division/Modulo ist aber nicht günstig im Vergleich zur einfachen Pointer-Arithmetik, die für zusammenhängende Speicherblöcke genutzt werden kann, deshalb nutzen wir fastmod, um die Kosten zu senken.

Sortieralgorithmus

Mit den bisher beschriebenen Komponenten können wir große sortierte Runs erzeugen, die page-by-page ins Storage gespillt werden können, statt alles-oder-nichts. Wir nutzen eine Kombination aus drei Sortieralgorithmen, um gute Sortierperformance und hohe Adaptivität an vorsortierte Daten zu erreichen:

1. Vergesort
2. Ska Sort
3. Pattern-defeating quicksort

Vergesort erkennt und merget Runs von (fast) sortierten Daten, was den Aufwand stark reduziert, z. B. bei Zeitreihendaten, die oft schon sortiert gespeichert sind. Kann Vergesort keine Muster erkennen, fällt es auf Ska Sort zurück, das einen adaptiven Most-Significant-Digit-(MSD-)Radix Sort auf dem ersten 64-Bit-Integer des Sort Keys ausführt. Werden Radix-Partitionen in der Rekursion zu klein, oder sind die Daten nach dem ersten 64-Bit-Integer nicht vollständig sortiert, fällt es auf Pattern-defeating Quicksort zurück.

Mergen

Vor v1.4.0 hat DuckDB die vollständig gemergten Daten materialisiert. Mit einem k-way Merge ist es aber möglich, Chunks sortierter Daten direkt aus den sortierten Runs auszugeben, in einer Art Streaming-Manier. Das heißt, Daten können ausgegeben werden, bevor der volle Merge berechnet ist. Wir visualisieren das für vier sortierte Runs:

Chunk 1 kann an die nächste Pipeline ausgegeben werden, bevor alle sortierten Runs gemerged sind. Einer der Gründe, warum das nützlich ist, sind große ORDER BY ... LIMIT ...-Queries. Ist das LIMIT klein, nutzt DuckDB einen Min-Heap, der deutlich schneller ist als das Sortieren der gesamten Eingabe. Bei großen LIMITs wird der Min-Heap-Ansatz aber schlechter als vollständiges Sortieren und dann Anwenden des LIMIT. Mit einem k-way Merge kann der Merge jederzeit durch ein LIMIT gestoppt werden, das heißt, die Kosten eines vollständigen Merges der sortierten Runs fallen nie an.

Traditionell wird der k-way Merge sequenziell mit einem Tournament Tree ausgewertet. Mit modernen Multi-Core-CPUs lässt das aber viel Performance liegen. Die Frage ist: Wie machen wir das parallel?

K-Way Merge Path

Es gibt verschiedene Algorithmen, um Merge Sort zu parallelisieren, etwa Merge Path, das DuckDBs Sort vor v1.4.0 nutzte, und Bitonic Merge Sort. Diese Algorithmen parallelisieren aber einen kaskadierenden Two-Way Merge Sort, keinen k-way Merge Sort. Sie sind also parallel und schiefresistenz, für Larger-than-Memory-Sorting aber unattraktiv, weil sie deutlich mehr I/O erzeugen.

Für k-way Merging gibt es weniger Parallelisierungsoptionen. Die Arbeit kann mit value-based Splitting aufgeteilt werden. Es ist aber leicht zu sehen, dass die Parallelität zusammenbricht, wenn die Eingabeverteilung extrem schief ist, z. B. wenn die Hälfte der Eingabe denselben Wert hat, weil es keinen Splitting-Wert gibt, der die Arbeit in gleich große Tasks teilen kann. Nach einer Websuche ist der einzige schiefresistente parallele k-way Merge, den wir finden konnten, eine Bachelorarbeit von 2014. Wir wollten einen sehr feingranularen Ansatz, deshalb haben wir stattdessen Merge Path auf k sortierte Runs verallgemeinert.

In der vorherigen Abbildung gibt es in jedem sortierten Run eine horizontale Linie, die anzeigt, wie viel von jedem sortierten Run in den Ausgabe-Chunk gegangen ist. Die Grundidee von Merge Path, erklärt in unserem Blogbeitrag zum Sortieren vor vier Jahren, ist, zu berechnen, wo diese Linien liegen, d. h., wo sich die sortierten Runs schneiden. Merge Path tut das effizient für das Mergen zweier sortierter Runs mit binärer Suche.

Wir verallgemeinern diesen Ansatz auf k sortierte Runs, sodass wir eine beliebige Ausgabe-Chunk-Größe wählen und berechnen können, wo sich die sortierten Runs so schneiden, dass beim Mergen der resultierende Chunk die gewählte Größe hat. Das erlaubt sehr feingranulare, schiefresistente Parallelität, die nicht möglich ist, wenn man spezifische Splitting-Werte wählt, weil die Größe der so erzeugten Chunks von der Datenverteilung abhängt. Das ist der Pseudocode für k-way Merge Path:

def compute_intersections(sorted_runs, chunk_size):
    intersections = [0 for _ in range(len(sorted_runs))]
    while chunk_size != 0:
        delta = ceil(chunk_size / len(sorted_runs))
        min_idx = 0
        min_val = sorted_runs[0][intersections[0] + delta]
        for run_idx in range(1, len(sorted_runs)):
            val = sorted_runs[run_idx][intersections[run_idx] + delta]
            if val < min_val:
                min_idx = run_idx
                min_val = val
        intersections[min_idx] += delta
        chunk_size -= delta
    return intersections

Das ist stark vereinfacht, weil es keine Edge Cases oder Out-of-Bounds auf den sortierten Runs berücksichtigt. Die Grundidee: Wir schieben in jeder Iteration der while-Schleife die untere Schranke für die Intersection eines sortierten Runs nach oben. Das hat eine schlechtere Komplexität als die binäre Suche im originalen Merge Path, muss aber auch seltener aufgerufen werden, weil ein k-way Merge alle sortierten Runs in einem einzigen Durchlauf mergen kann, statt in vielen Durchläufen. Profiling zeigt, dass diese Berechnung nur 1–2 % der gesamten Ausführungszeit ausmacht.

Threads können die Intersections unabhängig und damit parallel berechnen. Sobald Threads die Intersections berechnet haben, können sie die Daten zwischen den Intersections frei mergen, weil garantiert ist, dass die Daten sich nicht mit denen anderer Threads überlappen. Die gemergten Chunks können dank DuckDBs order-preserving Parallelism sofort parallel ausgegeben werden.

Benchmark

Wie performt die neue Sort-Implementierung im Vergleich zur alten? Wir führen ein paar Experimente auf meinem Laptop aus (M1-Max-MacBook Pro mit 10 Threads und 64 GB Speicher).

Rohe Performance

Zuerst benchmarken wir die rohe Integer-Sortierperformance. Wir haben drei Arten von Eingaben (aufsteigend vorsortiert, absteigend vorsortiert und zufällig geordnet) in drei Größen (10, 100 und 1000 Millionen Zeilen). Die Daten haben wir mit den folgenden Queries erzeugt:

CREATE TABLE ascending10m AS
    SELECT range AS i FROM range(10_000_000);

CREATE TABLE descending10m AS
    SELECT range AS i FROM range(9_999_999, 0, -1);

CREATE TABLE random10m AS
    SELECT range AS i FROM range(10_000_000) ORDER BY random();

-- and so on for 100m and 1000m

Wir haben den Median von 5 Läufen jeder dieser Queries für jede Tabellengröße genommen:

SELECT any_value(i)
FROM (FROM ascending10m ORDER BY i);

SELECT any_value(i)
FROM (FROM descending10m ORDER BY i);

SELECT any_value(i)
FROM (FROM random10m ORDER BY i);

-- and so on for 100m and 1000m

Diese Query veranlasst DuckDB, den gesamten Sort auszuwerten, ohne die ganze Tabelle als Query-Ergebnis zu materialisieren. So können wir die Performance der Sort-Implementierung besser isolieren.

Ergebnisse

Das zeigt, dass die neue Implementierung hoch adaptiv an vorsortierte Daten ist: Sie ist etwa 10× schneller bei auf-/absteigenden Daten als die alte Implementierung. Sie hat deutlich bessere rohe Sortierperformance: Sie ist mehr als 2× schneller beim Sortieren zufällig geordneter Daten (bei 1000 Millionen).

Wir plotten die Ergebnisse auch auf einer log-log-Skala:

Hier sieht man, dass die neue Implementierung deutlich besser skaliert: Die Ausführungszeit der neuen Implementierung steigt weniger steil mit der Eingabegröße als die der alten Implementierung.

Breite Tabelle

Der erste Benchmark hat die rohe Sortierperformance bewertet. In diesem nächsten Benchmark sortieren wir eine breite Tabelle, d. h., wir selektieren viele Spalten, die von der ORDER BY-Klausel sortiert werden. Wir sortieren die Tabelle lineitem aus TPC-H mit 15 Spalten nach der Spalte l_shipdate, bei Scale Factors 1 (etwa 6 Millionen Zeilen), 10 (etwa 60 Millionen Zeilen) und 100 (etwa 600 Millionen Zeilen), erzeugt mit DuckDBs TPC-H-Extension.

Wir haben die mediane Ausführungszeit von 5 Läufen dieser Query für jeden Scale Factor genommen:

SELECT any_value(COLUMNS(*))
FROM (FROM lineitem ORDER BY l_shipdate);

Ergebnisse

Wir haben das Speicherlimit auf 30 GB gesetzt, die Daten passen bei Scale Factor 100 also nicht mehr in den Speicher. Die neue Implementierung ist bei Scale Factors 1 und 10 etwa 2× schneller und bei Scale Factor 100 mehr als 3× schneller. Das zeigt, dass der neue k-way Merge Sort Datenbewegung und I/O reduziert und breite Tabellen deutlich effizienter sortiert als der alte kaskadierende 2-Way Merge Sort.

Wieder plotten wir die Ergebnisse auf einer log-log-Skala:

Und man sieht, dass die neue Implementierung deutlich besser skaliert, besonders wenn die Daten nicht mehr in den Hauptspeicher passen.

Thread-Skalierung

Schließlich benchmarken wir, wie gut die Sort-Implementierung mit Threads skaliert. Wir sortieren die Tabelle mit 100 Millionen zufällig geordneten Integers von vorher, mit 1, 2, 4 und 8 Threads. Wir nutzen dieselben Daten und dieselbe Query wie im ersten Benchmark und nehmen den Median von fünf Läufen.

Ergebnisse

Wie man sieht, ist die neue Single-Thread-Sortierperformance etwa 30 % langsamer als die alte. Das liegt daran, dass die neue Sort-Implementierung einen In-Place-MSD-Radix-Sort nutzt statt eines Out-of-Place-Least-Significant-Digit-(LSD-)Radix-Sort. Dadurch performt die alte Implementierung speziell auf diesem Workload besser, auf Kosten deutlich höheren Speicherverbrauchs.

Erhöhen wir die Thread-Zahl auf 2, ist dieser Vorteil aber schon weg. Bei 8 Threads hat die alte Implementierung nur einen Speedup von etwa 3,5× gegenüber 1 Thread, während dieser Speedup bei der neuen Implementierung etwa 6,5× beträgt.

Wieder plotten wir die Ergebnisse auf einer log-log-Skala:

Das zeigt, dass die parallele Skalierung der neuen Implementierung deutlich besser ist als die der alten.

Fazit

DuckDBs neue Sort-Implementierung hat die Performance gegenüber der alten Sort-Implementierung stark verbessert. Sie ist hoch adaptiv an vorsortierte Daten, erzeugt weniger I/O beim Sortieren von Daten, die nicht in den Hauptspeicher passen, und skaliert deutlich besser mit zusätzlichen Threads.

Wenn Sie auf v1.4.0 upgegradet haben, können Sie die verbesserte Performance bei der Nutzung der ORDER BY-Klausel genießen. Die neue Sort-Implementierung ist bereits in den Window-Operator integriert, wir erwarten also auch Performanceverbesserungen bei der Nutzung der OVER-Klausel. Für v1.5.0 wollen wir die neue Sort-Implementierung in die Joins integrieren, die Sortieren nutzen, etwa den ASOF-Join.