2023-09-15

DuckDBs AsOf-Joins: Fuzzy temporale Lookups

Richard Wesley

Haben Sie Zeitreihendaten, die Sie joinen wollen, aber die Zeitstempel passen nicht ganz? Oder wollen Sie einen Wert nachschlagen, der sich über die Zeit ändert, anhand der Zeiten in einer anderen Tabelle? Und sind Sie bei verschlungenen (und langsamen) Inequality-Joins gelandet? Dann ist dieser Beitrag für Sie!

Was ist ein AsOf-Join?

Zeitreihendaten sind nicht immer perfekt ausgerichtet. Uhren können leicht danebenliegen, oder zwischen Ursache und Wirkung liegt eine Verzögerung. Zwei geordnete Datensätze zu verbinden kann dadurch schwierig werden. AsOf-Joins sind ein Werkzeug für genau das und ähnliche Probleme.

Eines der Probleme, die AsOf-Joins lösen, ist der Wert einer sich ändernden Eigenschaft zu einem bestimmten Zeitpunkt. Dieser Anwendungsfall ist so häufig, dass der Name daher kommt: Gib mir den Wert der Eigenschaft as of this time – zu diesem Zeitpunkt.

Allgemeiner verkörpern AsOf-Joins gängige Semantik temporaler Analyse, die in Standard-SQL umständlich und langsam umzusetzen ist.

Portfolio-Beispiel

Beginnen wir mit einem konkreten Beispiel. Angenommen, wir haben eine Tabelle mit Aktien-prices und Zeitstempeln:

Eine weitere Tabelle enthält Portfolio-holdings zu verschiedenen Zeitpunkten:

Den Wert jeder Position zu diesem Zeitpunkt berechnen wir, indem wir den jüngsten Preis vor dem Zeitstempel der Position per AsOf-Join holen:

SELECT h.ticker, h.when, price * shares AS value
FROM holdings h ASOF JOIN prices p
  ON h.ticker = p.ticker
 AND h.when >= p.when;

Das hängt den Wert der Position zu diesem Zeitpunkt an jede Zeile:

Im Kern wird eine Funktion ausgeführt, die nahe Werte in der Tabelle prices nachschlägt. Fehlende ticker-Werte haben keinen Match und erscheinen nicht in der Ausgabe.

Äußere AsOf-Joins

Weil AsOf höchstens einen Match von der rechten Seite liefert, wächst die linke Tabelle durch den Join nicht, sie kann aber schrumpfen, wenn rechts Zeiten fehlen. Dafür gibt es einen äußeren AsOf-Join:

SELECT h.ticker, h.when, price * shares AS value
FROM holdings h ASOF LEFT JOIN prices p
  ON h.ticker = p.ticker
 AND h.when >= p.when
ORDER BY ALL;

Wie zu erwarten, entstehen NULL-Preise und -Werte statt fallender linker Zeilen, wenn der Ticker fehlt oder die Zeit vor dem Beginn der Preise liegt.

Alternative mit Windowing

Standard-SQL kann diesen Join umsetzen, braucht dafür aber eine Window-Funktion und einen Inequality-Join. Beides kann teuer sein; die Abfrage sähe so aus:

WITH state AS (
    SELECT
        ticker,
        price,
        "when",
        lead("when", 1, 'infinity')
            OVER (PARTITION BY ticker ORDER BY "when") AS end
    FROM prices
)
SELECT h.ticker, h.when, price * shares AS value
FROM holdings h
INNER JOIN state s
        ON h.ticker = s.ticker
      AND h.when >= s.when
      AND h.when < s.end;

Der Standardwert infinity sorgt dafür, dass die letzte Zeile einen vergleichbaren Endwert hat. So sieht die CTE state in unserem Beispiel aus:

Ohne Gleichheitsbedingung müsste der Planner einen Inequality-Join nutzen, der sehr teuer sein kann. Selbst mit Gleichheitsbedingung kann der Hash-Join lange Ketten identischer ticker-Schlüssel erzeugen, die alle matchen und beschnitten werden müssen.

Warum AsOf?

Wenn SQL AsOf-Joins schon berechnen kann, warum ein neuer Join-Typ? Zwei große Gründe: Ausdrucksstärke und Leistung. Die Windowing-Alternative ist weitschweifiger und schwerer zu verstehen als die AsOf-Syntax; leichter zu sagen, was Sie tun, hilft anderen (oder Ihnen selbst), zu verstehen, was passiert.

Die Syntax macht es DuckDB auch leichter, zu verstehen, was Sie wollen, und Ergebnisse schneller zu liefern. Die Variante mit Window und Inequality-Join verliert die wertvolle Information, dass sich die Intervalle nicht überlappen. Außerdem kann der Optimizer den Join nicht verschieben, weil SQL Windowing nach Joins verlangt. Behandeln wir die Operation als Join mit bekannten Datenconstraints, kann DuckDB den Join aus Leistungsgründen verschieben und einen maßgeschneiderten Algorithmus nutzen. Wir sortieren die rechte Tabelle und machen eine Art Merge-Join mit den linken Werten. Anders als ein Standard-Merge-Join kann AsOf die Suche beim ersten Match beenden, weil es höchstens einen gibt.

State-Tabellen

Vielleicht fragen Sie sich, warum die Common Table Expression in der Klausel WITH state hieß. Die Tabelle prices ist ein Beispiel für das, was in der temporalen Analyse eine Event-Tabelle heißt. Die Zeilen einer Event-Tabelle enthalten Zeitstempel und das, was zu diesem Zeitpunkt passiert ist (Events). Die Events in prices sind Preisänderungen einer Aktie. Ein anderes gängiges Beispiel ist eine strukturierte Logdatei: Jede Zeile hält fest, wann etwas „passiert“ ist – meist eine Änderung an einem Teil des Systems.

Event-Tabellen sind unhandlich, weil jede Tatsache nur die Startzeit hat. Ob die Tatsache noch gilt (oder zu einem bestimmten Zeitpunkt galt), braucht auch die Endzeit. Eine Tabelle mit Start- und Endzeit heißt State-Tabelle. Event-Tabellen in State-Tabellen zu wandeln ist eine gängige Vorbereitung temporaler Daten; die Windowing-CTE oben zeigt, wie das allgemein in SQL geht.

Sentinel-Werte

Eine Grenze des Windowing-Ansatzes: Der Ordnungstyp braucht einen Sentinel-Wert, falls er kein infinity kennt – entweder einen ungenutzten Wert oder NULL.

Beide Varianten sind potenziell problematisch. Im ersten Fall ist ein oberer Sentinel nicht immer leicht zu finden (was, wenn die Ordnung eine String-Spalte ist?). Im zweiten Fall müsste die Bedingung h.when < s.end OR s.end IS NULL heißen; ein solches OR in einer Join-Bedingung macht Vergleiche langsam und schwer zu optimieren. Nutzt die Ordnungsspalte NULL schon für fehlende Werte, fällt diese Option weg.

Für die meisten State-Tabellen gibt es passende Werte (z. B. große Daten). Ein Vorteil von AsOf: Eine State-Tabelle muss nicht entworfen werden, wenn die Analyse sie nicht braucht.

Varianten von Event-Tabellen

Bisher nutzen wir eine Standard-Event-Tabelle, bei der die Zeitstempel als Beginn der Zustandsübergänge gelten. AsOf kann jetzt beliebige Ungleichungen nutzen und damit andere Event-Tabellen-Typen.

Zur Erkundung zwei sehr einfache Tabellen ohne Gleichheitsbedingungen. Die Build-Seite hat nur vier Integer-„Zeitstempel“ mit alphabetischen Werten:

Die Probe-Tabelle sind die Zeitwerte plus die Mittelpunkte. Eine Tabelle zeigt, welcher Wert jede Probe-Zeit bei größer-gleich trifft:

Das Intervall, das ein Probe-Wert trifft, ist das halboffene Intervall [Tn, Tn+1).

Was passiert bei streng größer als als Ungleichung:

Das getroffene Intervall ist jetzt das halboffene Intervall (Tn, Tn+1]. Der einzige Unterschied: Das Intervall ist am Ende geschlossen statt am Anfang. Bei diesem Ungleichungstyp gehört die Zeit also nicht zum Intervall.

Was, wenn die Ungleichung in die andere Richtung geht, etwa kleiner-gleich?

Wieder halboffene Intervalle, diesmal das vorherige Intervall (Tn-1, Tn]. Eine Deutung: Die Zeiten in der Build-Tabelle sind das Ende des Intervalls, nicht der Beginn. Anders als bei größer-gleich ist das Intervall am Ende geschlossen statt am Anfang. Zusammen mit dem Ergebnis für streng größer als heißt das: Lookup-Zeiten gehören zum Intervall, wenn nicht-strenge Ungleichungen genutzt werden.

Prüfen wir das an der letzten Ungleichung: streng kleiner als:

Die passenden Intervalle sind [Tn-1, Tn). Strenge Ungleichung: Die Tabellenzeit liegt nicht im Intervall. Kleiner-als: Die Zeit ist das Ende des Intervalls.

Zusammengefasst die volle Liste:

Zwei natürliche Deutungen der Ungleichungen:

• Größer- (bzw. kleiner-)als heißt: Die Zeit ist Beginn (bzw. Ende) des Intervalls.
• Strenge (bzw. nicht-strenge) Ungleichungen heißen: Die Zeit ist ausgeschlossen (bzw. eingeschlossen).

Wissen wir, ob die Zeit Start oder Ende des Events markiert und ob sie eingeschlossen oder ausgeschlossen ist, wählen wir die passende AsOf-Ungleichung.

Nutzung

Bisher haben wir die AsOf-Bedingungen explizit angegeben. SQL hat aber eine vereinfachte Join-Bedingungssyntax für den häufigen Fall gleicher Spaltennamen in beiden Tabellen. Die Syntax nutzt das Schlüsselwort USING und listet die Felder für den Gleichheitsvergleich. AsOf unterstützt das ebenfalls, mit zwei Einschränkungen:

• Das letzte Feld ist die Ungleichung
• Die Ungleichung ist >= (der häufigste Fall)

Unsere erste Abfrage lässt sich dann so schreiben:

SELECT ticker, h.when, price * shares AS value
FROM holdings h
ASOF JOIN prices p
    USING(ticker, "when");

Achtung: Listen Sie die Spalten im SELECT nicht explizit auf, ist der Wert des Ordnungsfelds der Probe-Wert, nicht der Build-Wert. Bei einem Natural Join ist das egal, weil alle Bedingungen Gleichheiten sind; bei AsOf muss eine Seite gewählt werden. AsOf lässt sich als Lookup-Funktion sehen; natürlicher ist es, die „Funktionsargumente“ zurückzugeben als die Innereien.

Unter der Haube

Ein AsOf-Join erlaubt es, eine Event-Tabelle für Join-Operationen wie eine State-Tabelle zu behandeln. Weil die Join-Semantik bekannt ist, muss keine volle State-Tabelle entstehen; das ist effizienter als ein allgemeiner Inequality-Join.

Zuerst die Windowing-Variante. Zur Erinnerung: Diese Abfrage wandelte die Event-Tabelle in eine State-Tabelle:

WITH state AS (
    SELECT
        ticker,
        price,
        "when",
        lead("when", 1, 'infinity')
            OVER (PARTITION BY ticker ORDER BY "when") AS end
    FROM prices
);

Die State-Tabellen-CTE entsteht durch Hash-Partitionierung der Tabelle nach ticker, Sortierung nach when und eine weitere Spalte, die when um eins nach unten verschiebt. Der Join ist dann ein Hash-Join auf ticker plus zwei Vergleiche auf when.

Ohne Spalte ticker (z. B. Preise für ein einzelnes Item) würde der Join über unseren Inequality-Join-Operator laufen, der beide Seiten materialisiert und sortiert, weil er nicht weiß, dass die Bereiche disjunkt sind.

Der AsOf-Operator nutzt alle drei Operator-Pipeline-APIs, um Zeilen zu bündeln und zu sammeln. In der Phase sink hash-partitioniert und sortiert AsOf die rechte Seite zu einer temporären State-Tabelle. (Tatsächlich derselbe Code wie Window, ohne die Endspalte unnötig zu materialisieren.) In der Phase operator filtert er Zeilen heraus (oder gibt sie zurück), die wegen NULL in den Prädikaten nicht matchen können, dann hash-partitioniert und sortiert er die restlichen Zeilen in einen Cache. In der Phase source matcht er Hash-Partitionen und macht einen Merge-Join der sortierten Werte je Partition.

Benchmarks

Weil AsOf-Joins sich auf verschiedene Weise in Standard-SQL umsetzen lassen, geht es beim Benchmarking um den Vergleich der Alternativen.

Eine Alternative ist ein Debugging-PRAGMA für AsOf namens debug_asof_iejoin, das den Join über Window und IEJoin umsetzt. So lassen sich Implementierungen umschalten und Laufzeiten vergleichen.

Andere Alternativen kombinieren Equi-Joins und Window-Funktionen. Der Equi-Join setzt die Gleichheitsbedingungen um, das Window wählt die nächste Ungleichung. Wir schauen uns zwei Windowing-Techniken an und vergleichen die Leistung. Wer den Abschnitt überspringen will: Manchmal sind sie etwas schneller, der AsOf-Join hat aber das konsistenteste Verhalten aller Algorithmen.

Window als State-Tabelle

Der erste Benchmark vergleicht einen Hash-Join mit einer State-Tabelle. Er probed eine 5-Mio.-Zeilen-Tabelle aus 100.000 Zeitstempeln und 50 Partitionierungsschlüsseln über einen Self-Join, bei dem nur 50 % der Schlüssel vorkommen und die Zeitstempel auf die Mitte zwischen den Originalen verschoben sind:

CREATE OR REPLACE TABLE build AS (
    SELECT k, '2001-01-01 00:00:00'::TIMESTAMP + INTERVAL (v) MINUTE AS t, v
    FROM range(0, 100_000) vals(v), range(0, 50) keys(k)
);

CREATE OR REPLACE TABLE probe AS (
    SELECT k * 2 AS k, t - INTERVAL (30) SECOND AS t
    FROM build
);

Die Tabelle build sieht so aus:

und die Probe-Tabelle so (nur gerade Werte für k):

Der Benchmark macht nur den Join und summiert die Spalte v:

SELECT sum(v)
FROM probe
ASOF JOIN build USING(k, t);

Das Debugging-PRAGMA erlaubt keinen Hash-Join, wir können die State-Tabelle aber wieder in einer CTE anlegen und einen Inner Join nutzen:

-- Hash Join implementation
WITH state AS (
    SELECT k,
      t AS begin,
      v,
      lead(t, 1, 'infinity'::TIMESTAMP) OVER (PARTITION BY k ORDER BY t) AS end
    FROM build
)
SELECT sum(v)
FROM probe p
INNER JOIN state s
        ON p.t >= s.begin
       AND p.t < s.end
       AND p.k = s.k;

Das funktioniert, weil der Planner Gleichheitsbedingungen für selektiver hält als Ungleichungen und einen Hash-Join mit Filter erzeugt.

Der Benchmark ergibt ungefähr:

Die Laufzeitverbesserung von AsOf gegenüber IEJoin liegt hier bei etwa 9×. Die miserable Leistung des Hash-Joins kommt von den langen (100.000) Bucket-Ketten in der Hash-Tabelle.

Der zweite Benchmark testet den Fall, dass die Probe-Seite etwa 10× kleiner ist als die Build-Seite:

CREATE OR REPLACE TABLE probe AS
    SELECT k,
      '2021-01-01T00:00:00'::TIMESTAMP +
          INTERVAL (random() * 60 * 60 * 24 * 365) SECOND AS t,
    FROM range(0, 100_000) tbl(k);

CREATE OR REPLACE TABLE build AS
    SELECT r % 100_000 AS k,
      '2021-01-01T00:00:00'::TIMESTAMP +
          INTERVAL (random() * 60 * 60 * 24 * 365) SECOND AS t,
      (random() * 100_000)::INTEGER AS v
    FROM range(0, 1_000_000) tbl(r);

SELECT sum(v)
FROM probe p
ASOF JOIN build b
  ON p.k = b.k
 AND p.t >= b.t

-- Hash Join Version
WITH state AS (
  SELECT k,
    t AS begin,
    v,
    lead(t, 1, 'infinity'::TIMESTAMP)
        OVER (PARTITION BY k ORDER BY t) AS end
  FROM build
)
SELECT sum(v)
FROM probe p
INNER JOIN state s
        ON p.t >= s.begin
       AND p.t < s.end
       AND p.k = s.k;

Jetzt ist die Laufzeitverbesserung von AsOf gegenüber IEJoin enorm (~500×), weil die Partitionierung fast alle Gleichheits-Mismatches eliminiert.

Die Hash-Join-Implementierung schneidet hier viel besser ab, weil der Optimizer merkt, dass die Probe-Seite kleiner ist, und die Hash-Tabelle auf der „Probe“-Tabelle baut. Außerdem sind die Probe-Werte hier eindeutig, die Hash-Ketten sind minimal.

Window mit Ranking

Eine andere Nutzung des Window-Operators:

• Tabellen über die Gleichheitsprädikate joinen
• Auf Paare filtern, bei denen die Build-Zeit vor der Probe-Zeit liegt
• Das Ergebnis nach Gleichheitsschlüsseln und Probe-Zeitstempel partitionieren
• Die Partitionen nach Build-Zeitstempel absteigend sortieren
• Alle Werte außer Rang 1 filtern (also die größte Build-Zeit <= Probe-Zeit)

Die Abfrage sieht so aus:

WITH win AS (
    SELECT p.k, p.t, v,
        rank() OVER (PARTITION BY p.k, p.t ORDER BY b.t DESC) AS r
    FROM probe p INNER JOIN build b
      ON p.k = b.k
    AND p.t >= b.t
    QUALIFY r = 1
)
SELECT k, t, v
FROM win;

Der Vorteil dieser Windowing-Abfrage: keine Sentinel-Werte, sie funktioniert mit jedem Datentyp. Der Nachteil: deutlich mehr Partitionen, weil beide Zeitstempel dabei sind – komplexeres Sortieren. Und weil das Window nach dem Join kommt, können riesige Zwischenstände entstehen, externes Sortieren und teure Out-of-Memory-Operationen.

Für diesen Benchmark drei Build-Tabellen und zwei Probe-Tabellen, alle mit 10.000 Integer-Gleichheitsschlüsseln. Die Probe-Tabellen haben 1 oder 15 Zeitstempel je Schlüssel:

CREATE OR REPLACE TABLE probe15 AS
    SELECT k, t
    FROM range(10_000) cs(k),
         range('2022-01-01'::TIMESTAMP, '2023-01-01'::TIMESTAMP, INTERVAL 26 DAY) ts(t);

CREATE OR REPLACE TABLE probe1 AS
    SELECT k, '2022-01-01'::TIMESTAMP t
    FROM range(10_000) cs(k);

Die Build-Tabellen sind viel größer und haben etwa das 10-/100-/1000-Fache der Einträge der 15-Element-Tabellen:

-- 10:1
CREATE OR REPLACE TABLE build10 AS
    SELECT k, t, (random() * 1000)::DECIMAL(7, 2) AS v
    FROM range(10_000) ks(k),
         range('2022-01-01'::TIMESTAMP, '2023-01-01'::TIMESTAMP, INTERVAL 59 HOUR) ts(t);

-- 100:1
CREATE OR REPLACE TABLE build100 AS
    SELECT k, t, (random() * 1000)::DECIMAL(7, 2) AS v
    FROM range(10_000) ks(k),
         range('2022-01-01'::TIMESTAMP, '2023-01-01'::TIMESTAMP, INTERVAL 350 MINUTE) ts(t);

-- 1000:1
CREATE OR REPLACE TABLE build1000 AS
    SELECT k, t, (random() * 1000)::DECIMAL(7, 2) AS v
    FROM range(10_000) ks(k),
         range('2022-01-01'::TIMESTAMP, '2023-01-01'::TIMESTAMP, INTERVAL 35 MINUTE) ts(t);

Die AsOf-Join-Abfragen:

-- AsOf/IEJoin
SELECT p.k, p.t, v
FROM probe p ASOF JOIN build b
  ON p.k = b.k
 AND p.t >= b.t
ORDER BY 1, 2;

-- Rank
WITH win AS (
    SELECT p.k, p.t, v,
        rank() OVER (PARTITION BY p.k, p.t ORDER BY b.t DESC)  AS r
    FROM probe p INNER JOIN build b
      ON p.k = b.k
    AND p.t >= b.t
    QUALIFY r = 1
)
SELECT k, t, v
FROM win
ORDER BY 1, 2;

Die Ergebnisse:

(Median von 5, außer Rank/15/1000).

• Bei allen Verhältnissen mit 15 Probes ist AsOf am schnellsten.
• Bei kleinen Verhältnissen mit 15 Probes schlägt Rank IEJoin (beide mit Windowing), ab 100:1 explodiert es aber.
• Bei Single-Element-Probes ist Rank am wirksamsten, der Vorsprung gegenüber AsOf liegt selbst im großen Maßstab nur bei etwa 50 %.

AsOf ließe sich also noch verbessern – vorherzusagen, wo, wäre aber schwierig, und danebenzuliegen hätte enorme Kosten.

Ausblick

DuckDB kann AsOf-Joins jetzt für alle Ungleichungstypen mit vernünftiger Leistung ausführen. In manchen Fällen liegt der Gewinn bei mehreren Größenordnungen gegenüber den Standard-SQL-Varianten – auch mit unserem schnellen Inequality-Join-Operator.

Der aktuelle AsOf-Operator ist vollständig allgemein, ein paar Planungsoptimierungen wären aber möglich.

• Bei selektiven Gleichheitsbedingungen wäre ein Hash-Join mit Filter gegen eine materialisierte State-Tabelle wahrscheinlich deutlich schneller. Erkennt der Planner das und gibt es passende Sentinel-Werte, könnte er statt der Standard-AsOf-Implementierung einen Hash-Join wählen.
• Es gibt auch Fälle, in denen die Probe-Tabelle viel kleiner ist als die Build-Tabelle, plus Gleichheitsbedingungen; ein Hash-Join gegen die Probe-Tabelle könnte deutlich schneller sein.

SQL hat den Vorteil, deklarativ zu sein: Sie sagen was Sie wollen, die Datenbank kümmert sich um das wie. Jetzt, da die Semantik des AsOf-Joins festliegt, können Sie Abfragen schreiben, die sagen, was Sie wollen – und wir können das wie weiter verbessern!

Viel Spaß beim Joinen!

Eines der spannendsten Dinge an der Arbeit an DuckDB: Es dehnt das klassische SQL-Modell ungeordneter Daten. DuckDB macht es leicht, geordnete Datensätze wie DataFrames und Parquet-Dateien abzufragen – und bei solchen Daten erwartet man geordnete Analyse! Schnelles Sortieren, schnelles Windowing und schnelle AsOf-Joins setzen diese Erwartung um.