2021-10-13

Windowing in DuckDB

Richard Wesley

Window-Funktionen (mit der Klausel OVER) sind wichtige Werkzeuge für die Analyse von Datenreihen, können aber langsam sein, wenn sie nicht sorgfältig implementiert sind. In diesem Beitrag schauen wir uns an, wie DuckDB Windowing umsetzt. Wir sehen auch, wie DuckDB seine Aggregate-Architektur nutzt, um nützliche gleitende Aggregationen wie gleitende Interquartilsabstände (IQRs) zu berechnen.

Jenseits von Mengen

Das ursprüngliche Relationenmodell, das Codd in den 1970er Jahren entwickelte, behandelte Relationen als ungeordnete Mengen von Tupeln. Das war praktisch für theoretische Informatik, ignorierte aber, wie Menschen mit physischen Analogien denken (das Modell des „verkörperten Gehirns“ aus der Neurowissenschaft). Menschen ordnen Daten natürlich, um sie zu verstehen und mit ihnen zu arbeiten. SQL nutzt dafür die Klausel SELECT für das horizontale Layout und ORDER BY für das vertikale.

Die Ordnungen, die Menschen Daten geben, sind oft mehr als neurologische Krücken. Zeit etwa legt eine natürliche Ordnung auf Messungen, und starke Ausschläge können selbst wichtige Daten sein oder anzeigen, dass die Daten durch Glätten bereinigt werden müssen. Trends können vorhanden sein, oder relative Änderungen sind für die Analyse wichtiger als Rohwerte. Um solche Fragen zu beantworten, führte SQL 2003 analytische (oder Window-) Funktionen ein.

Window-Funktionen

Windowing zerlegt eine Relation in unabhängige Partitionen, ordnet diese Partitionen und definiert dann verschiedene Funktionen, die für jede Zeile anhand der benachbarten Werte berechnet werden können. Dazu gehören alle Aggregatfunktionen (etwa sum und avg) sowie einige window-spezifische Funktionen (etwa rank() und nth_value(<expression>, <N>)).

Manche Window-Funktionen hängen nur von der Partitionsgrenze und der Ordnung ab, einige (einschließlich aller Aggregationen) nutzen zusätzlich einen Frame. Frames werden als Anzahl von Zeilen auf beiden Seiten (preceding oder following) der aktuellen Zeile angegeben. Die Distanz kann als Anzahl von rows oder als range von Werten anhand des Ordnungswerts der Partition und einer Distanz angegeben werden.

Framing ist der verwirrendste Teil der Windowing-Umgebung, schauen wir uns daher ein sehr einfaches Beispiel an und ignorieren Partitionierung und Ordnung zunächst.

SELECT points,
    sum(points) OVER (
        ROWS BETWEEN 1 PRECEDING
                 AND 1 FOLLOWING) AS we
FROM results;

Diese Abfrage berechnet die sum jedes Punkts und der Punkte zu beiden Seiten:

Am Rand der Partition werden nur zwei Werte addiert.

Beispiel Stromerzeugung

Nun ein konkretes Beispiel einer Window-Funktionsabfrage. Angenommen, wir haben Erzeugungsdaten eines Kraftwerks:

Die Daten sind verrauscht, daher wollen wir für jedes Kraftwerk einen 7-Tage-gleitenden Mittelwert berechnen. Dafür können wir diese Window-Abfrage nutzen:

SELECT "Plant", "Date",
    avg("MWh") OVER (
        PARTITION BY "Plant"
        ORDER BY "Date" ASC
        RANGE BETWEEN INTERVAL 3 DAYS PRECEDING
                  AND INTERVAL 3 DAYS FOLLOWING)
        AS "MWh 7-day Moving Average"
FROM "Generation History"
ORDER BY 1, 2;

Diese Abfrage berechnet den siebentägigen gleitenden Mittelwert der erzeugten Energie jedes Kraftwerks an jedem Tag. Die Klausel OVER ist die SQL-Art, eine Funktion in einem Window zu berechnen. Sie partitioniert die Daten nach Plant (um die Kraftwerke zu trennen), ordnet jede Partition nach Date (um die Energiemessungen nebeneinanderzulegen) und nutzt einen RANGE-Frame von drei Tagen zu beiden Seiten jedes Tages für avg (um fehlende Tage abzufangen). Hier das Ergebnis:

Sie können mehrere verschiedene OVER-Klauseln in demselben SELECT anfordern, jede wird getrennt berechnet. Häufig wollen Sie dasselbe Window für mehrere Funktionen nutzen, dann können Sie mit einer WINDOW-Klausel ein benanntes Window definieren:

SELECT "Plant", "Date",
    avg("MWh") OVER seven AS "MWh 7-day Moving Average"
FROM "Generation History"
WINDOW seven AS (
    PARTITION BY "Plant"
    ORDER BY "Date" ASC
    RANGE BETWEEN INTERVAL 3 DAYS PRECEDING
              AND INTERVAL 3 DAYS FOLLOWING)
ORDER BY 1, 2;

Das wäre nützlich, wenn man zusätzlich das 7-Tage-gleitende min und max haben wollte, um die Grenzen der Daten zu zeigen.

Unter den Federn

Das ist eine lange Liste komplizierter Funktionalität! Damit das alles relativ schnell läuft, gehören viele Teile dazu, schauen wir uns an, wie sie in DuckDB umgesetzt sind.

Pipeline Breaking

Als Erstes fällt auf, dass Windowing ein „Pipeline Breaker“ ist. Der Operator Window muss alle Eingaben lesen, bevor er eine Funktion berechnen kann. Gibt es einen anderen Weg, etwas zu berechnen, kann eine andere Technik deutlich schneller sein.

Eine häufige analytische Aufgabe ist, den letzten Wert in einer Gruppe zu finden. Angenommen, wir wollen die zuletzt aufgezeichnete Leistung jedes Kraftwerks. Verlockend ist, die Window-Funktion rank() mit umgekehrter Sortierung zu nutzen:

SELECT "Plant", "MWh"
FROM (
    SELECT "Plant", "MWh",
        rank() OVER (
            PARTITION BY "Plant"
            ORDER BY "Date" DESC) AS r
    FROM table) t
WHERE r = 1;

Das erfordert jedoch, die gesamte Tabelle zu materialisieren, zu partitionieren, die Partitionen zu sortieren und dann eine einzelne Zeile aus diesen Partitionen zu ziehen. Viel schneller ist ein Self-Join, der die Tabelle auf den letzten (max) Wert des Felds DATE filtert:

SELECT table."Plant", "MWh"
FROM table,
    (SELECT "Plant", max("Date") AS "Date"
     FROM table GROUP BY 1) lasts
WHERE table."Plant" = lasts."Plant"
  AND table."Date" = lasts."Date";

Diese Join-Abfrage braucht zwei Scans der Tabelle, aber das einzige materialisierte Datum ist die Filtertabelle (die wahrscheinlich viel kleiner ist als die Originaltabelle), und es wird gar nicht sortiert.

Diese Art von Abfrage tauchte in einem Nutzerblog auf, und wir fanden, dass die Join-Abfrage auf ihrem Datensatz über 20-mal schneller war:

Natürlich erfordern die meisten analytischen Aufgaben mit Windowing doch den Operator Window, und DuckDB nutzt eine Reihe von Techniken, um die Leistung so hoch wie möglich zu halten.

Partitionierung und Sortierung

Früher wurde Windowing umgesetzt, indem nach Partition und Ordnungsfeldern sortiert und dann die Partitionsgrenzen gefunden wurden. Das ist ressourcenintensiv, weil die gesamte Relation sortiert werden muss und Sortieren O(N log N) in der Größe der Relation ist. Glücklicherweise gibt es schnellere Wege für diesen Schritt.

Um den Ressourcenverbrauch zu senken, nutzt DuckDB das Partitionierungsschema aus Leis et al., Efficient Processing of Window Functions in Analytical SQL Queries, und zerlegt die Partitionen in 1024 Chunks mit O(N)-Hashing. Die Chunks müssen trotzdem nach allen Feldern sortiert werden, weil es Hash-Kollisionen geben kann, aber jede Partition kann jetzt 1024-mal kleiner sein, was die Laufzeit deutlich senkt. Außerdem können die Partitionen leicht extrahiert und parallel verarbeitet werden.

Sortieren in DuckDB hat kürzlich einen großen Leistungsschub bekommen, zusammen mit der Fähigkeit, Partitionen zu verarbeiten, die größer als der Speicher sind. Diese Funktionalität wurde auch in den Operator Window übernommen, was im Last-in-Group-Beispiel 33 % Verbesserung brachte:

Als letzte Optimierung sammelt DuckDB Funktionen mit derselben Partitionierung und Ordnung, auch wenn Sie mehrere Window-Funktionen anfordern, und teilt das Datenlayout zwischen diesen Funktionen.

Aggregation

Die meisten allgemeinen Window-Funktionen sind einfach zu berechnen, aber windowierte Aggregatfunktionen können teuer sein, weil sie für jede Zeile mehrere Werte betrachten müssen. Oft müssen sie denselben Wert mehrfach betrachten oder wiederholt eine große Zahl von Werten ansehen, daher wurden über die Jahre mehrere Ansätze zur Leistungssteigerung verfolgt.

Naive windowierte Aggregation

Bevor wir erklären, wie DuckDB windowierte Aggregation umsetzt, ein kurzer Umweg über gewöhnliche Aggregationen. Aggregat-„Funktionen“ werden mit drei erforderlichen Operationen und einer optionalen umgesetzt:

• Initialize – Erzeugt einen Zustand, der aktualisiert wird. Für sum ist das die laufende Summe, beginnend bei NULL (weil die Summe von null Elementen NULL ist, nicht null).
• Update – Aktualisiert den Zustand mit einem neuen Wert. Für sum wird der Wert zum Zustand addiert.
• Finalize – Erzeugt den endgültigen Aggregatwert aus dem Zustand. Für sum wird einfach die laufende Summe kopiert.
• Combine – Kombiniert zwei Zustände zu einem. Combine ist optional, erlaubt aber, die Aggregation parallel zu berechnen. Für sum entsteht ein neuer Zustand mit der Summe der beiden Eingabewerte.

Der einfachste Weg, einen einzelnen windowierten Aggregatwert zu berechnen, ist, einen Zustand zu initialisieren, den Zustand mit allen Werten im Window-Frame zu aktualisieren und dann mit finalize den Wert der windowierten Aggregation zu erzeugen. Dieser naive Algorithmus funktioniert immer, ist aber ziemlich ineffizient. Eine laufende Summe addiert zum Beispiel alle Werte vom Beginn der Partition für jede laufende Summe erneut, mit einer Laufzeit von O(N^2).

Um das zu verbessern, fügen manche Datenbanken zusätzliche „Moving-State“-Operationen hinzu, die einzelne Werte inkrementell hinzufügen oder entfernen können. Das reduziert die Berechnung in manchen häufigen Fällen, kann aber nur für bestimmte Aggregationen genutzt werden. Für min funktioniert es zum Beispiel nicht, weil man nicht weiß, ob es mehrere gleiche Minima gibt. Und wenn sich die Frame-Grenzen stark bewegen, kann es immer noch zu O(N^2) degenerieren.

Segment-Tree-Aggregation

Statt mehr Funktionen zu ergänzen, nutzt DuckDB den Segment-Tree-Ansatz aus Leis et al. Dabei wird ein Baum über die gesamte Partition gebaut, mit den aggregierten Werten unten. Werte werden in Zustände an Knoten darüber kombiniert, bis es eine einzelne Wurzel gibt:

Um einen Wert zu berechnen, erzeugt der Algorithmus Zustände für die unregelmäßigen Enden des Frames, kombiniert Zustände im Baum über den Werten im Frame und finalisiert das Ergebnis aus dem letzten verbliebenen Zustand. Im Beispiel oben (Abbildung 5 aus Leis et al.) müssen nur drei Werte statt 7 addiert werden. Diese Technik funktioniert für alle kombinierbaren Aggregationen.

Allgemeine windowierte Aggregation

Der größte Nachteil von Segment Trees ist die Verwaltung einer potenziell großen Zahl von Zwischenzuständen. Für die einfachen Zustände standardmäßiger distributiver Aggregationen wie sum ist das kein Problem, weil die Zustände klein sind, der Baum die Zahl der Zustände logarithmisch niedrig hält und der Zustand zur Berechnung jedes Werts ebenfalls günstig ist.

Für manche Aggregationen ist der Zustand jedoch nicht klein. Typischerweise sind das sogenannte holistische Aggregationen, bei denen der Wert von allen Werten des Frames abhängt. Beispiele sind mode und quantile, bei denen jeder Zustand eine Kopie aller bisher gesehenen Werte enthalten kann. Segment Trees können genutzt werden, um gleitende Versionen jeder kombinierbaren Aggregation umzusetzen, das kann für große, komplexe Zustände aber teuer sein – und das war nicht das ursprüngliche Ziel des Algorithmus.

Um dieses Problem zu lösen, nutzen wir den Ansatz aus Wesley und Xu, Incremental Computation of Common Windowed Holistic Aggregates, der Segment Trees auf aggregatspezifische Datenstrukturen verallgemeinert. Die Aggregation kann eine fünfte optionale window-Operation definieren, der die Unterseite des Baums und die Grenzen des aktuellen und vorherigen Frames übergeben werden. Die Aggregation kann dann eine passende Datenstruktur für ihre Implementierung anlegen.

Die Funktion mode hält zum Beispiel eine Hash-Tabelle von Zählern, die sie effizient aktualisieren kann, und quantile hält eine teilweise sortierte Liste von Frame-Indizes. Außerdem können die quantile-Funktionen ein Array von Quantilwerten entgegennehmen, was die Leistung weiter steigert, indem die teilweise geordneten Ergebnisse zwischen den verschiedenen Quantilwerten geteilt werden.

Weil diese Aggregationen in einem Windowing-Kontext nutzbar sind, lässt sich das gleitende-Mittelwert-Beispiel oben leicht zu einem gleitenden Interquartilsabstand umbauen:

SELECT "Plant", "Date",
    quantile_cont("MWh", [0.25, 0.5, 0.75]) OVER seven
        AS "MWh 7-day Moving IQR"
FROM "Generation History"
WINDOW seven AS (
    PARTITION BY "Plant"
    ORDER BY "Date" ASC
    RANGE BETWEEN INTERVAL 3 DAYS PRECEDING
              AND INTERVAL 3 DAYS FOLLOWING)
ORDER BY 1, 2;

Gleitende Quantile wie diese sind robuster gegenüber Ausreißern, was sie zu einem wertvollen Werkzeug für die Analyse von Datenreihen macht, in den meisten Datenbanksystemen sind sie aber nicht allgemein umgesetzt. Es gibt Ansätze, die in manchen Query-Engines genutzt werden können, aber ohne eine allgemeine Architektur für gleitende Aggregationen können diese Lösungen unnatürlich oder komplex sein. DuckDBs Implementierung nutzt die Standard-Window-Notation, Sie müssen also keine neue Syntax lernen oder die Daten in ein anderes Werkzeug ziehen.

Ordered-Set-Aggregationen

Window-Funktionen werden oft eng mit einigen speziellen „Ordered-Set-Aggregationen“ des SQL-Standards verbunden. Manche Datenbanken setzen diese Funktionen mit dem Operator Window um, das ist aber eher ineffizient, weil das Sortieren der Daten (eine O(N log N)-Operation) nicht nötig ist – es reicht, Hoares O(N)- FIND-Algorithmus, wie er in der STL- std::nth_element genutzt wird. DuckDB übersetzt diese Ordered-Set-Aggregationen auf die schnelleren regulären Aggregatfunktionen quantile_cont, quantile_disc und mode und vermeidet so Windowing vollständig.

Erweiterungen

Diese Architektur bedeutet auch, dass jede neue Aggregation, die wir ergänzen, von der bestehenden Windowing-Infrastruktur profitieren kann. DuckDB ist ein Open-Source-Projekt, und wir freuen uns über Einreichungen nützlicher Aggregatfunktionen – oder Sie können eigene domänenspezifische in Ihrem Fork anlegen. Irgendwann hoffen wir auf eine UDF-Architektur, die Plug-in-Aggregationen erlaubt, und die Einfachheit und Kraft der Schnittstelle wird diesen Plugins die notationelle Einfachheit und Laufzeitleistung der internen Funktionen geben.

Fazit

DuckDBs Windowing-Implementierung nutzt verschiedene Techniken, um zu beschleunigen, was der langsamste Teil einer analytischen Abfrage sein kann. Sie ist gut in das Sortier-Subsystem und die Aggregate-Architektur integriert, was fortgeschrittene gleitende Aggregationen natürlich und effizient ausdrückbar macht.

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.