2022-05-27

Range Joins in DuckDB

Richard Wesley

Range-Intersection-Joins sind eine wichtige Operation in Bereichen wie temporaler Analytik und treten auf, wenn zwei Ungleichheitsbedingungen in einem Join-Prädikat stehen. Datenbankimplementierungen greifen für diese Operationen oft auf langsame O(N^2)-Algorithmen zurück, die jedes Zeilenpaar vergleichen. DuckDB nutzt stattdessen seine schnelle Sortierlogik, um zwei hochoptimierte parallele Join-Operatoren für solche Range-Prädikate umzusetzen – mit 20–30× schnelleren Abfragen. Mit diesen Operatoren lässt sich DuckDB effektiv in stärker zeitreihenorientierten Einsatzfällen nutzen.

Einleitung

Das zeilenweise Joinen von Tabellen ist eine der grundlegenden und unterscheidenden Operationen des relationalen Modells. Ein Join verbindet zwei Tabellen horizontal über eine boolesche Bedingung, das Prädikat. Das klingt geradlinig, aber wie schnell der Join ausgeführt werden kann, hängt von den Ausdrücken im Prädikat ab. Das hat zur Entstehung verschiedener Join-Algorithmen geführt, die für unterschiedliche Prädikattypen optimiert sind.

In diesem Beitrag erklären wir mehrere Join-Algorithmen und ihre Fähigkeiten. Insbesondere beschreiben wir einen neu hinzugekommenen „Range Join“-Algorithmus, der das Verbinden von Tabellen über überlappende Zeitintervalle oder mehrere Ordnungsbedingungen deutlich schneller macht.

Flugdaten

Nein, dieser Teil handelt nicht von Enten, sondern von Flugstatistiken der Air Group aus dem Battlestar-Galactica-Reboot. Wir haben ein paar Tabellen, die wir nutzen: Pilots, Crafts, Missions und Battles. Einige Daten gingen verloren, als die Flotte sich zerstreute, aber hoffentlich reicht das für ein paar „reale“ Beispiele!

Die Tabelle Pilots enthält die Piloten und ihre unveränderlichen Daten (Name, Callsign, Seriennummer):

Die Tabelle Crafts enthält alle verschiedenen Kampfschiffe (das „Schiff des Theseus“-Problem recycelter Teile ignorieren wir!):

Die Tabelle Missions enthält alle von Piloten geflogenen Missionen. Missionen haben eine begin- und end-Zeit, die beim Flight Deck geloggt wird. Wir nutzen ein paar übliche Paarungen (und eine ungewöhnliche Mission am Ende, bei der Commander Adama seine alte Viper flog):

Die Tabelle Battles enthält das Zeitfenster jeder Schlacht mit den Zylonen.

Diese letzten beiden Tabellen (Missions und Battles) sind Beispiele für Zustandstabellen. Ein Objekt in einer Zustandstabelle hat einen Zustand, der zwischen zwei Zeitpunkten läuft. Bei den Schlachten ist der Zustand einfach ja/nein. Bei den Missionen ist der Zustand eine Kombination aus Pilot und Schiff.

Gleichheitsprädikate

Der häufigste Join-Typ vergleicht ein oder mehrere Ausdruckspaare auf Gleichheit, oft einen Primärschlüssel und einen Fremdschlüssel. Wollen wir zum Beispiel eine Liste der von den Piloten geflogenen Schiffe, können wir die Tabelle Pilots über die Tabelle Missions mit der Tabelle Craft joinen:

SELECT callsign, count(*), tailno
FROM Pilots p, Missions m, Crafts c
WHERE p.id = m.pid
  AND c.id = m.cid
GROUP BY ALL
ORDER BY 2 DESC;

Das ergibt eine Tabelle wie:

Range-Prädikate

Auffällig an diesem Beispiel ist, dass die Bedingungen, die die Tabellen joinen, Gleichheiten sind, die mit AND verbunden sind. Relationale Joins können aber mit jedem booleschen Prädikat definiert werden – auch solchen ohne Gleichheit oder AND.

Eine häufige Operation in temporalen Datenbanken ist das Schneiden zweier Zustandstabellen. Angenommen, wir wollen die Zeitintervalle finden, in denen jeder Pilot im Kampf war, damit wir Kampfstunden für die Seniorität berechnen können? Viper werden schnell gestartet, aber nicht bevor die Schlacht begonnen hat, und es kann Fehlfunktionen geben oder Piloten können sich verspäten, bis sie zum Flight Deck kommen.

SELECT callsign, battle,
    greatest(m.begin, b.begin) AS begin,
    least(m.end, b.end) AS end
FROM Pilots p, Missions m, Crafts c, Battles b
WHERE m.begin < b.end
  AND b.begin < m.end
  AND p.id = m.pid
  AND c.id = m.cid;

Dieser Join erzeugt eine Menge von Datensätzen mit Callsign und Kampfzeitraum für jeden Piloten. Er behandelt den Fall, dass ein Pilot für ein neues Schiff zurückkehrt, schließt Patrouillenflüge aus und behandelt sogar die Situation, wenn ein Patrouillenflug zum Kampf wird! Das liegt daran, dass das Schneiden von Zustandstabellen auf diese Weise eine gemeinsame Zustandstabelle erzeugt – eine wichtige Operation temporaler Datenbanken. Hier ein paar Zeilen aus dem Ergebnis:

Apollo war bereits in der Luft, als der erste Zylonenangriff kam, deshalb setzt die Abfrage seine begin-Zeit für die Schlacht auf den Beginn der Schlacht, nicht auf den Start zum Decommissioning-Überflug. Starbuck und Boomer wurden nach Beginn der Schlacht alarmiert, aber Boomer kehrte erst zurück, als die Schlacht faktisch vorbei war, deshalb wird ihre end-Zeit auf das offizielle Ende der Schlacht zurückgesetzt.

Wichtig ist hier, dass die Join-Bedingung zwischen der Relation Pilot/Mission/Schiff und der Schlachtentabelle keine Gleichheiten enthält. Diese Art von Join ist traditionell sehr teuer zu berechnen, aber wie wir sehen werden, gibt es Wege, sie zu beschleunigen.

Unendliche Zeit

Ein häufiges Problem beim Befüllen von Zustandstabellen ist, wie man die offenen Ränder darstellt. Zum Beispiel ist die Begin-Zeit des ersten Zustands möglicherweise unbekannt, oder der aktuelle Zustand ist noch nicht beendet.

Oft werden solche Werte durch NULLs dargestellt, das verkompliziert aber die Schnittabfrage, weil der Vergleich mit NULL NULL ergibt. Das lässt sich umgehen, indem man coalesce(end, <großer Zeitstempel>) nutzt, das fügt aber eine Berechnung zu jeder Zeile hinzu, von denen die meisten sie nicht brauchen. Ein anderer Ansatz ist, <großen Zeitstempel> direkt statt NULL zu nutzen, was das Ausdrucksberechnungsproblem löst, aber einen willkürlichen Zeitwert einführt. Dieser Wert kann bei Berechnungen merkwürdige Ergebnisse liefern.

DuckDB bietet eine dritte Alternative aus Postgres, die für diese Situationen genutzt werden kann: unendliche Zeitwerte. Unendliche Zeitwerte vergleichen sich wie erwartet, aber Arithmetik mit ihnen erzeugt NULLs oder Unendlichkeiten und zeigt damit an, dass die Berechnung nicht wohldefiniert ist.

Übliche Join-Algorithmen

Um zu sehen, warum diese Joins teuer sein können, schauen wir uns zuerst die zwei häufigsten Join-Algorithmen an.

Hash Joins

Joins mit mindestens einer Gleichheitsbedingung, die mit AND an den Rest der Bedingungen gehängt ist, heißen Equi-Joins. Sie werden üblicherweise mit einer Hash-Tabelle umgesetzt, etwa so:

hashes = {}
for b in build:
    hashes[b.pk] = b

result = []
for p in probe:
    result.append((p, hashes[p.fk], ))

Die Ausdrücke einer Seite (der Build-Seite) werden berechnet und gehasht, dann werden die entsprechenden Ausdrücke der anderen Seite (der Probe-Seite) in der Hash-Tabelle nachgeschlagen und auf einen Match geprüft.

Wir können das etwas anpassen, wenn nur einige der mit AND verbundenen Bedingungen Gleichheiten sind, indem wir die anderen Bedingungen prüfen, sobald wir die Gleichheiten in der Hash-Tabelle gefunden haben. Der wichtige Punkt ist, dass wir eine Hash-Tabelle nutzen können, um die Join-Laufzeit auf O(N) zu bringen. Diese Anpassung ist eine allgemeine Technik, die mit jedem Join-Algorithmus genutzt werden kann, der die möglichen Matches reduziert.

Nested Loop Joins

Da relationale Joins mit jedem booleschen Prädikat definiert werden können – auch einem ohne Gleichheit oder AND –, funktionieren Hash Joins nicht immer. Der Join-Algorithmus der letzten Wahl in diesen Situationen heißt Nested Loop Join (oder kurz NLJ) und besteht einfach darin, jede Zeile der Probe-Seite mit jeder Zeile der Build-Seite zu vergleichen:

result = []
for p in probe:
    for b in build
        if compare(p, b):
            result.append((p, b, ))

Das ist O(M x N) in der Zahl der Zeilen und kann sehr langsam sein, wenn die Tabellen groß sind. Schlimmer noch: Die meisten praktischen analytischen Abfragen (wie das Kampfstunden-Beispiel oben) liefern bei weitem nicht so viele Ergebnisse, sodass viel Aufwand verschwendet werden kann. Aber ohne einen Algorithmus, der auf eine Art von Prädikat zugeschnitten ist, müssten wir das nutzen.

Range Joins

Haben wir einen Range-Vergleich (eines von <, <= >, >=) als eine der Join-Bedingungen, können wir die implizierte Ordnung nutzen, indem wir die Eingaberelationen nach einigen der Join-Bedingungen sortieren. Sortieren ist O(N log N), was nahelegt, dass das schneller sein könnte als ein NLJ – und das stellt sich tatsächlich als der Fall heraus.

Piecewise Merge Join

Bevor Hash Joins aufkamen, sortierten Datenbanken die Join-Eingaben oft, um Matches zu finden. Bei Equi-Joins fand dann eine wiederholte binäre Suche die passenden Werte auf der Build-Seite in O(M log N) Zeit. Das heißt Merge Join und läuft schneller als O(M x N), aber nicht so schnell wie die O(N)-Zeit eines Hash Joins. Haben wir aber einen einzelnen Range-Vergleich, lässt uns die binäre Suche den ersten Match für einen Probe-Wert finden. Die restlichen Matches finden wir dann, indem wir nach dem ersten weitersuchen.

Sortieren wir auch die Probe-Seite, wissen wir sogar, wo wir die Suche für den nächsten Probe-Wert starten, weil er nach der Stelle liegen wird, an der wir den vorherigen Wert gefunden haben. So funktioniert Piecewise Merge Join (PWMJ): Wir sortieren die Build-Seite so, dass die Werte nach dem Prädikat geordnet sind (entweder ASC oder DESC), dann sortieren wir jeden Probe-Chunk auf dieselbe Weise, sodass wir schnell durch Wertemengen scannen können, um mögliche Matches zu finden. Das kann für diese Art von Abfragen deutlich schneller sein als NLJ. Gibt es weitere Join-Bedingungen, können wir die erzeugten Matches dann prüfen, um sicherzustellen, dass alle Bedingungen erfüllt sind, weil das Sortieren die Zahl der nötigen Prüfungen wieder deutlich reduziert hat.

Inequality Join (IEJoin)

Für zwei Range-Bedingungen (wie die Kampfgeld-Abfrage) gibt es noch schnellere Algorithmen. Wir haben kürzlich einen neuen Join namens IEJoin hinzugefügt, der nach zwei Prädikaten sortiert, um die Sache wirklich zu beschleunigen.

IEJoin funktioniert so: Zuerst werden beide Tabellen nach den Werten der ersten Bedingung sortiert und die beiden Sortierschlüssel in eine kombinierte Tabelle gemerged, die die Zeilennummern der beiden Eingabetabellen mitführt. Als Nächstes werden die Positionen in der kombinierten Tabelle nach der zweiten Range-Bedingung sortiert. Dann kann schnell nach Matches gescannt werden, die beide Bedingungen erfüllen. Und genau wie bei Hash Joins können wir alle restlichen Bedingungen prüfen, weil wir hoffentlich die Zahl der zu testenden Paare deutlich reduziert haben.

Walkthrough

Weil der Algorithmus etwas knifflig ist, gehen wir ein kleines Beispiel durch. (Wenn Sie das Paper lesen: Das ist eine vereinfachte Version der „Union Arrays“-Optimierung aus §4.3, aber ich finde diese Version des Algorithmus deutlich leichter verständlich als die Version in §3.1.) Wir schauen uns Qp aus dem Paper an, einen Self Join auf der Tabelle „West“:

Wir suchen Paare von Billing-IDs, bei denen die zweite ID eine kürzere Zeit als die erste hatte, aber höhere Kosten:

SELECT s1.t_id, s2.t_id AS t_id2
FROM west s1, west s2
WHERE s1.time > s2.time
  AND s1.cost < s2.cost;

Es gibt zwei Paare, die diese Kriterien erfüllen:

(Das ist ein Beispiel für eine andere Art von Double-Range-Abfrage, bei der wir nach Anomalien suchen.)

Zuerst sortieren wir beide Eingabetabellen nach dem Schlüssel der ersten Bedingung (time). (Wir sortieren DESC, weil die Werte die Join-Bedingung (>) von links nach rechts erfüllen sollen.)

Weil sie auf dieselbe Weise sortiert sind, können wir die Bedingungsschlüssel aus den sortierten Tabellen in eine neue Tabelle namens L1 mergen, nachdem wir jede Zeile mit der Tabelle markiert haben, aus der sie stammt (mit negativen Zeilennummern für die rechte Tabelle):

Die Spalte rid lässt uns Zeilen in L1 auf die ursprüngliche Tabelle zurückabbilden.

Als Nächstes bauen wir eine zweite Tabelle L2 mit dem Schlüssel der zweiten Bedingung (cost) und den Zeilenpositionen (P) von L1 (nicht den Zeilennummern der ursprünglichen Tabellen!) Wir sortieren L2 nach cost (diesmal wieder DESC, weil die Join-Bedingung jetzt von rechts nach links gelten soll):

Die sortierte Spalte der L1-Zeilenpositionen heißt Permutation Array, und wir können sie nutzen, um die entsprechende Position des time-Werts für ein gegebenes cost zu finden.

An diesem Punkt haben wir zwei Tabellen (L1 und L2), jede sortiert nach einer der Join-Bedingungen und zurückzeigend auf die Tabellen, aus denen sie abgeleitet wurde. Außerdem wurden die Sortierreihenfolgen so gewählt, dass die Bedingung von links nach rechts (bzw. rechts nach links) gilt. Da die Bedingungen transitiv sind, bedeutet das: Wann immer wir einen Wert haben, der eine Bedingung an einem Punkt in der Tabelle erfüllt, erfüllt er sie auch für alles rechts (bzw. links)!

Mit diesem Setup können wir L2 von links nach rechts scannen und mit zwei Indizes nach Zeilen suchen, die beide Bedingungen erfüllen:

• i iteriert über L2 von links nach rechts;
• off2 folgt i und dient dazu, costs zu identifizieren, die die Join-Bedingung im Vergleich zu i erfüllen. (Bei lockeren Ungleichungen könnte das rechts von i liegen);

Wir nutzen eine Bitmap B, um zu tracken, welche Zeilen in L1 der L2-Scan bereits als die cost-Bedingung im Vergleich zur L2-Scan-Position i erfüllend identifiziert hat.

Weil wir nur Matches zwischen einer linken und einer rechten Zeile wollen, können wir Matches überspringen, bei denen die rids unterschiedliche Vorzeichen haben. Um diese Beobachtung zu nutzen, verarbeiten wir nur Werte von i, die in der linken Tabelle liegen (rid[P[i]] ist positiv), und markieren Bits nur für Zeilen in der rechten Tabelle (rid[P[i]] ist negativ). In diesem Beispiel sind die rechten Zeilen die ungeraden Werte in P (die bequemerweise auch die ungeraden Werte von i sind), was sie im Beispiel leicht nachverfolgbar macht.

Für die anderen Zeilen passiert Folgendes:

Wann immer wir costs finden, die die Bedingung links der Scan-Position erfüllen (zwischen off2 und i), nutzen wir P[off2], um die Bits in B zu markieren, die den Positionen in L1 entsprechen, die rechte Zeilen referenzieren. Das hält fest, dass die cost-Bedingung für diese Zeilen erfüllt ist. Haben wir dann eine Position P[i] in L1, können wir B nach rechts scannen, um Werte zu finden, die ebenfalls die cost-Bedingung erfüllen. Das funktioniert, weil alles rechts von P[i] in L1 die price-Bedingung erfüllt – dank der Sortierreihenfolge von L1 und der Transitivität der Vergleichsoperationen.

Im Detail:

1. Wenn i und off2 0 sind, ist die cost-Bedingung < nicht erfüllt, also passiert nichts;
2. Wenn i 1 ist, schauen wir auf eine Zeile von der rechten Seite des Joins, also überspringen wir sie und machen weiter;
3. Wenn i 2 ist, schauen wir jetzt auf eine Zeile von der linken Seite, also holen wir off2 nach vorne, bis die cost-Bedingung fehlschlägt, und markieren B dort, wo sie bei P[1] = [1] erfolgreich ist;
4. Dann scannen wir die time-Werte in L1 rechts von Position P[i=2] = 6 und finden keine Matches in B;
5. Wenn i 4 ist, holen wir off2 wieder nach vorne und markieren B bei P[3] = [7];
6. Dann scannen wir time ab Position 2 und finden Matches bei [6,7], von denen einer (6) aus der rechten Tabelle stammt;
7. Wenn i 6 ist, holen wir off2 wieder nach vorne und markieren B bei P[5] = [3];
8. Dann scannen wir time ab Position 4 und finden wieder Matches bei [6,7];
9. Wenn i schließlich über das Ende läuft, haben wir keine neuen cost-Werte, also passiert nichts;

Was das schnell macht, ist, dass wir nur ein paar Bits prüfen müssen, um die Matches zu finden. Wenn wir Vergleiche durchführen müssen, können wir den schnellen Radix-Vergleichscode aus unserem Sortiercode nutzen, der keine speziellen templatisierten Versionen für jeden Datentyp braucht. Das reduziert nicht nur Codegröße und -komplexität, es macht ihn auch „zukunftssicher“ gegen neue Datentypen.

Weitere Details

Dieser Walkthrough ist eine etwas vereinfachte, Single-Thread-Version des tatsächlichen Algorithmus. Es gibt ein paar weitere Details, die interessant sein können:

• Das Scannen großer, überwiegend leerer Bitmaps kann langsam sein, deshalb nutzen wir die Bloom-Filter-Optimierung aus §4.2.
• Der veröffentlichte Algorithmus nimmt an, dass es keine doppelten L1-Werte in keiner der Tabellen gibt. Um den allgemeinen Fall zu behandeln, nutzen wir eine exponentielle Suche, um den ersten L1-Wert zu finden, der das Prädikat bezüglich der aktuellen Position erfüllt, und scannen von dort nach rechts;
• Wir haben auch den verteilten Algorithmus 3 aus §5 angepasst, indem wir Paare der vom Sortiercode erzeugten sortierten Blöcke auf getrennten Threads joinen. So können wir den Operator vollständig parallelisieren: zuerst durch paralleles Sortieren und dann durch das Aufteilen des Joins in unabhängige Stücke;
• Das Aufteilen der Stücke für parallele Ausführung erlaubt uns außerdem, Join-Blöcke, die gerade nicht verarbeitet werden, auf die Platte auszulagern, wodurch der Join skalierbar wird.

Spezielle Joins

Eines der Schönen an IEJoin ist, dass es sehr allgemein ist und eine Reihe spezialisierterer Join-Typen recht effizient umsetzt. Zum Beispiel ist die Zustandsschnitt-Abfrage oben ein Beispiel für einen Interval Join, bei dem wir über die Schnittmenge zweier Intervalle joinen.

Ein weiterer spezialisierter Join, der mit IEJoin beschleunigt werden kann, ist ein Band Join. Damit kann man Werte joinen, die „nah“ beieinanderliegen:

SELECT r.id, s.id
FROM r, s
WHERE r.value - s.value BETWEEN a AND b;

Das übersetzt sich in eine doppelte Ungleichheits-Join-Bedingung:

SELECT r.id, s.id
FROM r, s
WHERE s.value + a <= r.value AND r.value <= s.value + b;

genau die Art von Join-Ausdruck, die IEJoin behandelt.

Performance

Wie schnell ist der IEJoin also? Er ist so schnell, dass es schwierig ist, ihn mit den vorherigen Range-Join-Algorithmen zu vergleichen, weil die Verbesserungen so groß sind, dass die anderen Algorithmen in angemessener Zeit nicht fertig werden!

Einfache Messungen

Als Beispiel hier die Laufzeiten für einen 100K-Self-Join einiger Mitarbeiter-Steuer- und Gehaltsdaten, bei dem das Ziel ist, die 1001 Paare von Mitarbeitern zu finden, bei denen einer ein höheres Gehalt hat, der andere aber einen höheren Steuersatz:

SELECT
    r.id,
    s.id
FROM Employees r
JOIN Employees s
    ON r.salary < s.salary
    AND r.tax > s.tax;

Ein weiteres Beispiel ist ein Self Join, um 3772 überlappende Events in einer 30K-Event-Tabelle zu finden:

SELECT
    r.id,
    s.id
FROM events r
JOIN events s
    ON r.start <= s.end
    AND r.end >= s.start
    AND r.id <> s.id;

In beiden Fällen sehen wir Performance-Verbesserungen von 20–100×, was sehr hilfreich ist, wenn man viele solcher Abfragen ausführt!

Optimierungsmessungen

Ein drittes Beispiel zeigt die Bedeutung der Join-Paar-Filterung und der exponentiellen Suche. Die Daten sind eine Zustandstabelle von Bibliotheksausleihdaten aus einem anderen Interval-Join-Paper, und die Abfrage ist eine Point-in-Period-temporale Abfrage, die für Abbildung 4d genutzt wurde:

SELECT x, count(*) AS y
FROM books,
    (SELECT x FROM range('2013-01-01'::TIMESTAMP, '2014-01-01'::TIMESTAMP, INTERVAL 1 DAY) tbl(x)) dates
WHERE checkout <= x AND x <= return
GROUP BY ALL
ORDER BY 1;

Das Ergebnis ist eine Zählung der Zahl der Bücher, die um Mitternacht an jedem Tag ausgeliehen waren. Das sind die Laufzeiten auf einem 18-Kern-iMac Pro:

Die Abfrage joint eine 35M-Zeilen-Tabelle mit einer 365-Zeilen-Tabelle, der Großteil der Daten kommt also von der linken Seite. Indem wir das Setzen von Bits für die passenden Zeilen in der linken Tabelle vermeiden, eliminieren wir fast alle L1-Prüfungen. Das reduziert die Laufzeit dramatisch und verbessert die CPU-Auslastung.

Die Daten haben außerdem eine große Zahl von Zeilen, die Büchern entsprechen, die zu Jahresbeginn ausgeliehen waren und alle dasselbe checkout-Datum haben. Linear nach links im ersten Block zu suchen, um den ersten Match für den Scan zu finden, führte zu wiederholten Läufen von ~120K Vergleichen. Dadurch wurde die Laufzeit vollständig vom Verarbeiten des ersten Blocks dominiert. Indem wir die Zahl der Vergleiche für diese Zeilen von durchschnittlich ~60K auf 16 reduziert haben, fiel die Laufzeit um den Faktor 10, und die CPU-Auslastung verdoppelte sich.

Fazit und Feedback

In diesem Blogpost haben wir die neuen DuckDB-Range-Join-Verbesserungen durch den neuen IEJoin-Operator erklärt. Das sollte die Antwortzeit von Zustandstabellen-Joins und Anomalieerkennungs-Joins deutlich verbessern. Wir hoffen, das macht Ihre DuckDB-Erfahrung noch besser – und lassen Sie uns bitte wissen, wenn Sie auf Probleme stoßen! Melden Sie sich gerne auf unserer GitHub-Seite oder auf unserem Discord-Server.