2025-02-14

Durch Windows fliegen

Richard Wesley

Einleitung

Im vorherigen Beitrag bin ich auf einige neue Windowing-Funktionalität in DuckDB eingegangen, die über SQL verfügbar ist. Es gibt aber weitere Änderungen, die unsere Ressourcennutzung (etwa Speicher) verbessern, ohne neue Funktionalität hinzuzufügen. Schauen wir uns also „unter den Federn“ diese Änderungen an.

Wir haben uns zuvor auf einem window-funktionslastigen Workload selbst gebenchmarkt, der große Performance-Verbesserungen über die Zeit zeigte. Die in diesem Blogbeitrag vorgestellten Optimierungen treiben die Performance von DuckDBs Window-Operator noch weiter.

Segment-Tree-Vektorisierung

Eine wichtige Verbesserung im Sommer 2023 war, den Segment-Tree-Auswertungscode von Einzelwert-Auswertung auf vektorisierte Auswertung umzustellen. Sie fragen sich vielleicht, warum das in einer „vektorisierten relationalen Datenbank“ (!) nicht von Anfang an so war, aber die Antwort liegt im Nebel der Zeit. Meine beste Vermutung: Entweder war der veröffentlichte Algorithmus für Werte geschrieben, oder die Aggregat-API war noch nicht festgezurrt (oder beides).

In der alten Version nutzten wir die APIs update oder combine des Aggregats, aber nur mit den Werten und Tree-States für eine einzelne Zeile. Um die Segment-Tree-Aggregation zu vektorisieren, sammeln wir Vektoren von Blattwerten und Tree-States und flushen sie in den State jeder Ausgabezeile, wenn wir die Vektorkapazität von 2048 Zeilen erreichen. Etwas Sorgfalt war nötig, um ordnungssensitive Aggregate zu behandeln, indem Werte in der richtigen Reihenfolge gesammelt werden. Klauseln wie FILTER und EXCLUDE sorgten ebenfalls für Unterhaltung, aber die Segment Trees sind jetzt vollständig vektorisiert. Die Performance-Gewinne hier lagen bei etwa Faktor vier (von „Baseline“ zu „Fan Out“).

Das Diagramm unten zeigt die durch die Vektorisierungsverbesserungen erreichten Speedups:

Sobald Segment Trees vektorisiert waren, konnten wir denselben Ansatz nutzen, um DISTINCT-Aggregate mit Merge Sort Trees zu implementieren. Es könnte sich lohnen, die Custom-Window-API irgendwann auf Vektorisierung umzustellen, denn obwohl die meisten Custom-Window-Aggregate ziemlich langsam sind (z. B. quantile, mad und mode), ist count(*) ebenfalls als Custom Aggregate implementiert und würde wahrscheinlich von einer vektorisierten Implementierung profitieren.

Konstantenaggregation

Viele Window-Berechnungen sind Aggregate über Frames, und eine gängige analytische Aufgabe mit diesen Ergebnissen ist, ein Teilaggregat mit demselben Aggregat über die gesamte Partition zu vergleichen. Diesen Wert wiederholt zu berechnen, ist teuer und potenziell speicherverschwenderisch (die alte Implementierung würde zum Beispiel einen Segment Tree bauen, obwohl nur ein Wert nötig war).

Der bisherige Performance-Workaround dafür war, das Aggregat in einer Subquery zu berechnen und es über die Partition Keys zu joinen – das war, nun ja, unfreundlich. Stattdessen haben wir eine Optimierung ergänzt, die auf partitionsweite Aggregate prüft und diesen Wert einmal pro Partition berechnet. Das reduziert nicht nur Speicher und Rechenzeit für das Aggregat selbst, sondern wir können oft einen konstanten Vektor zurückgeben, der die Werte über alle Zeilen in einem Chunk teilt, was Kopierkosten und potenziell sogar Downstream-Auswertungskosten senkt.

Einen konstanten Vektor zurückzugeben, kann überraschend große Speicher- und Performance-Vorteile bringen. In dem Issue, das diese Verbesserung angetrieben hat, konstruierte der Nutzer eine konstante 100k-Elemente-Liste (!) und berechnete dann den Median mit einem List-Aggregation-Lambda. Indem wir eine einzelne konstante Liste zurückgeben, bauen und reduzieren wir diese Liste nur einmal statt einmal pro Zeile!

Streaming Windows

Window-Funktionen zu berechnen, ist meist ziemlich teuer! Die gesamte Relation muss materialisiert, in Partitionen zerlegt und jede Partition sortiert werden.

Aber was, wenn es keine Partitionierung oder Ordnung gibt? Das heißt einfach, dass die Window-Funktion über die gesamte Relation in der „natürlichen Ordnung“ berechnet wird, mit einem Frame, der bei der ersten Zeile beginnt und bis zur aktuellen Zeile weitergeht. Beispiele wären, Zeilennummern zuzuweisen oder eine laufende Summe zu berechnen. Das ist einfach genug, dass wir die Auswertung der Funktion auf einem einzelnen Thread streamen können.

Treten wir zuerst einen Schritt zurück und sprechen über den Window-Operator. Während Parsing und Optimierung einer Query werden alle Window-Funktionen an einen einzelnen logischen Window-Operator gehängt. Wenn es Zeit wird, die Query zu planen, gruppieren wir die Funktionen, die gemeinsame Partitionen und „kompatible“ Ordnungen haben (siehe Cao et al., Optimization of Analytic Window Functions für mehr Informationen) und übergeben jede Gruppe an einen eigenen physischen Window-Operator, der diese Partitionierung und Ordnung behandelt. Um die „natürliche Ordnung“ zu nutzen, müssen wir die Funktionen gruppieren, die gestreamt werden können, und sie zuerst ausführen (sonst wäre die Ordnung zerstört!) und sie an den streaming physischen Window-Operator übergeben.

Welche Window-Funktionen können wir also streamen? Es sind ziemlich viele:

• Aggregate BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW (wir updaten einfach das Aggregat)
• first_value – es ist immer dasselbe
• percent rank – es ist immer 0
• rank – es ist immer 0
• dense_rank – es ist immer 0
• row_number – wir zählen einfach die Zeilen
• lead und lag – wir halten einfach einen Buffer

Es gibt ein paar weitere Einschränkungen:

• IGNORE NULLS, EXCLUDE und ORDER BY-Argumente sind nicht erlaubt
• lead- und lag-Distanzen sind auf eine Konstante innerhalb ±2048 beschränkt (ein Vektor). Das ist keine große Sache, weil die Distanz meist 1 ist.

Die Verbesserungen für streaming LEAD waren ziemlich dramatisch:

SELECT setseed(0.8675309);
CREATE OR REPLACE TABLE df AS
    SELECT
        random() AS a,
        random() AS b,
        random() AS c,
    FROM range(10_000_000);

SELECT sum(a_1 + a_2 + b_1 + b_2)
FROM (
    SELECT
        lead(a, 1) OVER () AS a_1,
        lead(a, 2) OVER () AS a_2,
        lead(b, 1) OVER () AS b_1,
        lead(b, 2) OVER () AS b_2
    FROM df
) t;

Das Diagramm unten zeigt die Performance-Verbesserungen.

• Baseline – Wo wir angefangen haben
• Shift – Blöcke von Zeilen innerhalb der nicht-streaming Implementierung kopieren, statt eine nach der anderen
• Streaming – Erste Implementierung von streaming LEAD
• Partials – Kopieren vermeiden, wenn der gesamte Chunk einfach referenziert werden kann

Hinweis: Die x-Achse zeigt die Speedups gegenüber der Baseline (1×), und die absoluten Laufzeiten (in Sekunden) stehen als Labels.

In Zukunft können wir das Ende des Frames vielleicht auf eine konstante Distanz von der aktuellen Zeile lockern, die in die Bufferlänge passt (z. B. BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING), das wurde aber noch nicht untersucht.

Partition-Major-Auswertung

Window-Partitionen sind vollständig unabhängig, sie getrennt auszuwerten ist also attraktiv. Unsere erste Implementierung nutzte das und wertete jede Partition auf einem eigenen Thread aus. Das funktioniert gut, wenn Sie mehr Partitionen als Threads haben und sie alle ungefähr gleich groß sind, sind die Partitionsgrößen aber schief oder gibt es nur eine Partition (eine häufige Situation), dann bleiben die meisten Kerne idle, was den Durchsatz senkt.

Um die CPU-Auslastung zu verbessern, haben wir das Ausführungsmodell für v1.1 geändert, um Partitionen parallel auszuwerten. Die Partitionen werden von der größten zur kleinsten ausgewertet, und wir verteilen dann jede Partition über so viele Kerne wie möglich, während wir den Zugriff auf geteilte Datenstrukturen synchronisieren. Das war deutlich anspruchsvoller als die unabhängige Single-Thread-Auswertung von Partitionen, und wir hatten einige Synchronisierungsprobleme (hoffentlich jetzt alle behoben!), die in den v1.1.x-Releases behandelt wurden. Aber wir haben jetzt eine deutlich bessere Kernauslastung, besonders für unpartitionierte Daten. Als Nebenvorteil konnten wir den Speicher-Footprint reduzieren, weil weniger Partitionen gleichzeitig im Speicher waren.

Ein verbleibendes Issue ist die Grobkörnigkeit der Subpartitionen. Derzeit nutzen sie, um Kopieren zu vermeiden, die Blöcke, die der Sortiercode erzeugt, die oft größer sind, als uns lieb wäre. Die Größe dieser Chunks zu reduzieren, ist zukünftige Arbeit, aber hoffentlich kommen wir im Rahmen einiger vorgeschlagener Änderungen am Sortiercode dazu.

Out-of-Memory-Betrieb

Weil Windowing die gesamte Relation materialisiert, war es sehr leicht, das Speicherbudget einer Query zu sprengen. Für v1.2 haben wir von der Materialisierung aktiver Partitionen im Speicher auf eine pageable Collection umgestellt. Jetzt haben wir nicht nur weniger aktive Partitionen (dank der Partition-Major-Auswertung oben), sondern diese Partitionen selbst können jetzt auf Platte spulen. Das senkt den Speicherdruck bei der Auswertung großer Partitionen weiter.

Windowing ist aber so komplex, dass noch einige große Datenstrukturen übrig sind. Die Segment Trees und Merge Sort Trees zur Beschleunigung der Aggregation sind noch im Speicher, besonders die Zwischen-Aggregatzustände in der Mitte der Bäume. Das vollständig zu lösen, erfordert eine allgemeine Methode, Aggregatzustände auf Platte zu serialisieren, die wir noch nicht haben. Die meisten Aggregate lassen sich aber als Binärdaten ohne Sonderbehandlung serialisieren, kurzfristig können wir also wahrscheinlich viele Fälle mit der aktuellen Aggregationsinfrastruktur abdecken, genau wie bei GROUP BY.

Geteilte Ausdrücke

Window-Ausdrücke werden unabhängig ausgewertet, teilen sich aber oft Ausdrücke. Manche dieser Ausdrücke sind teuer auszuwerten, andere müssen über die gesamte Partition materialisiert werden. Als Beispiel für Letzteres können Aggregatfunktionen Werte überall in der Partition referenzieren. Das könnte dazu führen, dieselben Werte mehrfach zu berechnen und zu materialisieren:

-- Compute the moving average and range of x over a large window
SELECT
    x,
    min(x) OVER w AS min_x,
    avg(x) OVER w AS avg_x,
    max(x) OVER w AS max_x,
FROM data
WINDOW w AS (
    PARTITION BY p
    ORDER BY s
    ROWS BETWEEN 1_000_000 PRECEDING and 1_000_000 FOLLOWING
);

Die Daten für x zu pagieren, reduziert den Speicher-Footprint, aber die Segment Trees zur Auswertung der drei Aggregate enthalten doppelte Kopien von x – zusammen mit dem Operator selbst (der x zurückgeben muss). Mit v1.2 haben wir einen Mechanismus ergänzt, um die Auswertung solcher Ausdrücke zwischen Funktionen zu teilen. Das reduziert nicht nur Speicher, sondern in diesem Beispiel reduzieren wir auch Disk-Paging, weil alle drei Funktionen auf dieselben Werte zugreifen.

Es gibt eine Reihe von Stellen, an denen wir Ausdrücke teilen, darunter ORDER BY-Argumente, range-Ausdrücke und „Wert“-Funktionen wie lead, lag und nth_value, und wir halten immer Ausschau nach mehr (etwa Frame-Grenzen – oder sogar Segment Trees).

Zukünftige Arbeit

Ich habe eine Reihe von Dingen erwähnt, zu denen wir in Zukunft kommen möchten, eines, das in keines der bisherigen Themen schön passt, ist Query Rewriting. Es stellt sich heraus, dass manche Window-Funktionen mit anderen Techniken ausgewertet werden können, etwa Self-Joins und einigen unserer smarteren Aggregate (wie arg_max). Diese alternativen Query-Pläne zu erzeugen, kann große Performance-Vorteile haben, und wir planen, sie zu untersuchen.

Fazit

Wie Sie sehen, ist Windowing ein großes haariges Biest! Es gibt auch nicht viel veröffentlichte Forschung zu wirksamen Algorithmen (ich habe so ziemlich alles verlinkt), deshalb müssen wir oft selbst etwas erfinden oder auf extrem einfache und langsame Ansätze zurückfallen. Aber ich hoffe, dass viele von Ihnen etwas Neues und Spannendes in dem finden, womit wir uns die letzten 2–3 Jahre beschäftigt haben – und ich werde versuchen, zukünftige Window-Verbesserungen zeitnäher zu bloggen.