Bestes Vorgehen: Aggregation größerer Tabelle für schnellen Zugriff

_sparrow · 7. Januar 2025 um 21:24

Hallo liebes Forum,

ich habe eine Frage zum besten Vorgehen in folgendem Fall:

Ich habe eine Tabelle mit 40 Millionen Einträge. Ich brauche relativ häufig den neuesten, gültigen Eintrag für bestimmte Geräte darin.

Mit der folgenden beispielhaften Abfrage bekomme ich alle Datensätze:

SELECT
  DISTINCT ON (device_id)
  device_id,
  timestamp,
  type_id,
  value
FROM measurement m
  LEFT JOIN measurementconfig mc ON (m.type_id = mc.id)
WHERE mc.valid = true
ORDER BY device_id, timestamp DESC;

Da bleiben von 40 Millionen noch etwa 2 Millionen übrig.
Wenn ich nur eine Handvoll dieser Werte brauche, funktioniert das ganz gut.

Auf der andern Seite habe ich auch eine Tabelle mit den ungefähr 2 Millionen Devices.

Jetzt kommt die Herausforderung, dass ich aus der Device Tabelle nur die Devices selektieren möchte, bei denen dieser “neueste Wert” bestimmte Eigenschaften hat.
Packe ich die obige Abfrage in einen View und joine die die Devices, funktioniert das zwar, dauert aber schon seine Zeit. Und das Ziel ist es, das Ergebnis möglichst schnell zu bekommen. Und im besten Fall auch die aktuellsten Werte.

Erste Idee:
Aus der obigen Abfrage einen Materialized View aufzubauen dauert zwar 2 Minuten, dafür ist der Join dagegen entsprechend schnell.
Das Problem ist, dass dann die aktuellen Daten fehlen und ständig neue Werte eintreffen. Gefühlt müsste ein Job den MV die ganze Zeit aktuell halten.
Als Idee hatte ich gedacht, mit einem Full Outer Join des MV und nur der neueren Einträge aus measurement (und einem COALESCE der Felder) ‘günstig’ alle Datensätze zu erhalten.
Funktioniert deutlich schlechter als gedacht.

Zweite Idee:
Eine Tabelle statt des Materialized Views verwenden.
Periodisch läuft ein Job, der die neuesten Datensätze aus measurement dort upserted.
Das geht deutlich schneller als dem MV immer wieder aufzubauen.

Dritte Idee:
Ein Trigger, der den eingefügten Datensatz aus measurement auch in die Tabelle für die aggregation schreibt. Die können dort nur leider nicht pauschal hinein geschrieben werden, weil es theoretisch vorkommen kann, dass ein Datensatz mit einem älteren Timestamp ankommt als es bereits gibt.

Übersehe ich etwas?
Könnt ihr mir eine Empfehlung geben?

pogomips · 8. Januar 2025 um 14:42

Leider sehe ich aus Deiner Beschreibung nicht recht, welche Tabelle welches Mengengerüst hat, wo die zusäzlichen, neuen Einschränkungen erfolgen sollen und welche “Handvoll” Daten Du benötigst. Handvoll Datensätze oder Handvoll Spalten, wenn ja welche…

Bei sowas ist es hilfreich, in einem geposteten Select nicht nur im Join die Table Aliase zu verwenden, sondern auch im Select Teil, trotz (für Insider) eindeutigen Spaltennamen. Dann sieht man es auch ohne weitere Detailkenntnis des Datenmodels schon allein am Statement.

Wie auch immer, angenommen

measurementconfig mc ist die kleine Tabelle mit 2 Mio
measurment m die große mit 40
mc enthält Kriterien, die zusätzlich (und zwingend) berücksichtigt werden müssen

Dann

filtere zuerst mc auf möglichst wenig Datensätze
joine das (nun hoffentlich sehr kleine) Ergebnis (also Handvoll) mit m

Falls mein Beispiel nicht zu den Tabellennamen passt, das Prinzip ist, auf
indizierten Feldern größtmögliche Einschränkungen zu erzielen, entweder
durch Filterung oder durch Joinen.
Danach erst exakte Filterung mit komplexen Kriterien. Die dann dank drastischer Mengenreduzierung nicht mehr weh tut.

Dabei ist nicht immer offensichtlich, was “davor” und “danach” oder “zuerst”
tatsächlich in der Formulierung des SQL bedeutet. Hier empfiehlt es
sich, langsame Statements mit exlplain zu analysieren, damit klar wird,
was der Optimizer macht, ob vorhandene Indizes genutzt werden oder nicht
und welche Auswirkungen die Umformulierung des SQL für Änderungen am
Ausführungsplan hat. Bspw. kann man so feste Joins mit Outer Joins
vergleichen usw.

Ohne die genaue Kenntnis von “neueste”:
Falls es hier eine Business Logik oder grobe Logik gibt, die “neueste”
brauchbar einschränkt, z.B. höchstens 1 Woche alt oder einen Monat oder
max 10 Tage oder bestimmte Stichtage (ab roleout, ab
Rechnungsstellungstermin, …), dann den Timestamp mit in den Filter
einbauen, zumindest wenn der Timestamp indiziert ist oder indizierbar
ist. Bei hinreichend kleinen Werten von “neueste”, kann das zu enormen
Mengeneinschränkungen führen, unabhängig von den sonstigen Kriterien.
Das kann man machen, wenn a) sogar gar keine exakte Einschränkung
existiert und b) dennoch ein minimum/maximum definiert ist, das immer(!)
gilt.

Ansonsten wäre Partitionierung der Tabellen vielleicht hilfreich, also das was Du mit Tabellenkopien erreichen willst, aber per “offizieller” Kommandos. Es
setzt aber brauchbare Partitionierungskriterien voraus, scheitert u.U.
auch an den neuen Werten. (Ich habe da gerade spontan nicht den
Überblick, wie smart die Partitionierung mittlerweile definierbar ist)

akretschmer · 8. Januar 2025 um 17:23

zeig mal den genauen Aufbau der Tabellen inkl. Indexe sowie ein explain analyse.

_sparrow · 8. Januar 2025 um 21:24

Gerne!

Hier sind die Tabelle mit Anzahl:

# \d device
                         Table "public.device"
   Column   |           Type           | Collation | Nullable | Default
------------+--------------------------+-----------+----------+---------
 id         | uuid                     |           | not null |
 device_id  | character varying(50)    |           | not null |
 note       | character varying(35)    |           |          |
 created_at | timestamp with time zone |           | not null |
 updated_at | timestamp with time zone |           | not null |
Indexes:
    "device_pkey" PRIMARY KEY, btree (id)
    "device_id_uidx" UNIQUE CONSTRAINT, btree (device_id)

# SELECT COUNT(*) FROM device;
  count
---------
 1527746
(1 row)

# \d measurement
                       Table "public.measurement"
   Column   |           Type           | Collation | Nullable | Default
------------+--------------------------+-----------+----------+---------
 id         | uuid                     |           | not null |
 value      | character varying(35)    |           |          |
 timestamp  | timestamp with time zone |           | not null |
 created_at | timestamp with time zone |           | not null |
 updated_at | timestamp with time zone |           | not null |
 device_id  | character varying(50)    |           | not null |
 type_id    | character varying(3)     |           | not null |
Indexes:
    "measurement_pkey" PRIMARY KEY, btree (id)
    "device_timestamp_desc_idx" btree (device_id, "timestamp" DESC)
    "measurement_device_type_idx" btree (device_id, type_id)
    "measurement_type_id_idx" btree (type_id)

# SELECT COUNT(*) FROM measurement;
  count
----------
 26463363
(1 row)

# \d measurementconfig
                     Table "public.measurementconfig"
    Column    |           Type           | Collation | Nullable | Default
--------------+--------------------------+-----------+----------+---------
 type         | character varying(3)     |           | not null |
 is_not_valid | boolean                  |           | not null |
 mark_1       | character varying(15)    |           | not null |
 mark_2       | character varying(15)    |           | not null |
 mark_3       | character varying(15)    |           | not null |
 mark_4       | character varying(15)    |           | not null |
 mark_5       | character varying(15)    |           | not null |
 mark_6       | character varying(15)    |           | not null |
 created_at   | timestamp with time zone |           | not null |
 updated_at   | timestamp with time zone |           | not null |
Indexes:
    "measurementconfig_key" PRIMARY KEY, btree (type)

# SELECT COUNT(*) FROM measurementconfig;
 count
-------
   142
(1 row)

Und das ist eine Beispielabfrage:

EXPLAIN ANALYZE WITH newest_valid_measurement AS (
    SELECT DISTINCT ON (m.device_id)
        m.device_id,
        m.timestamp,
        m.type_id,
        m.value,
        mc.mark_1,
        mc.mark_2,
        mc.mark_3,
        mc.mark_4,
        mc.mark_5,
        mc.mark_6
    FROM measurement m
        LEFT JOIN measurementconfig mc ON (m.type_id = mc.type)
    WHERE mc.is_not_valid = false
    ORDER BY m.device_id, m.timestamp DESC
)
SELECT
    d.device_id,
    d.note,
    m.value
FROM device d 
    LEFT JOIN newest_valid_measurement m
        ON (d.device_id = m.device_id)
WHERE m.mark_6 = 'ok';
                                                                                  QUERY PLAN
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Nested Loop  (cost=2384370.46..6192248.52 rows=620 width=26) (actual time=46042.793..139855.102 rows=1211585 loops=1)
   ->  Subquery Scan on m  (cost=2384370.03..6187080.62 rows=620 width=16) (actual time=46039.820..64665.217 rows=1272036 loops=1)
         Filter: ((m.mark_6)::text = 'ok'::text)
         Rows Removed by Filter: 332837
         ->  Unique  (cost=2384370.03..6185531.74 rows=123910 width=286) (actual time=46039.773..64388.580 rows=1604873 loops=1)
               ->  Nested Loop  (cost=2384370.03..6124053.14 rows=24591440 width=286) (actual time=46039.772..63420.324 rows=14300167 loops=1)
                     ->  Gather Merge  (cost=2384369.88..5465423.97 rows=26454428 width=27) (actual time=46039.708..57477.615 rows=26463363 loops=1)
                           Workers Planned: 2
                           Workers Launched: 2
                           ->  Sort  (cost=2383369.85..2410926.55 rows=11022678 width=27) (actual time=45979.759..52091.054 rows=8821121 loops=3)
                                 Sort Key: m_1.device_id, m_1."timestamp" DESC
                                 Sort Method: external merge  Disk: 333312kB
                                 Worker 0:  Sort Method: external merge  Disk: 333048kB
                                 Worker 1:  Sort Method: external merge  Disk: 333024kB
                                 ->  Parallel Seq Scan on measurement m_1  (cost=0.00..566591.78 rows=11022678 width=27) (actual time=96.415..21116.229 rows=8821121 loops=3)
                     ->  Memoize  (cost=0.15..0.17 rows=1 width=9) (actual time=0.000..0.000 rows=1 loops=26463363)
                           Cache Key: m_1.type_id
                           Cache Mode: logical
                           Hits: 26463153  Misses: 210  Evictions: 0  Overflows: 0  Memory Usage: 19kB
                           ->  Index Scan using measurementconfig_key on measurementconfig mc  (cost=0.14..0.16 rows=1 width=9) (actual time=0.012..0.012 rows=1 loops=210)
                                 Index Cond: ((type)::text = (m_1.type_id)::text)
                                 Filter: (NOT is_not_valid)
                                 Rows Removed by Filter: 0
   ->  Index Scan using device_id_uidx on device d  (cost=0.43..8.34 rows=1 width=25) (actual time=0.058..0.058 rows=1 loops=1272036)
         Index Cond: ((device_id)::text = (m.device_id)::text)
 Planning Time: 1.625 ms
 JIT:
   Functions: 28
   Options: Inlining true, Optimization true, Expressions true, Deforming true
   Timing: Generation 1.973 ms, Inlining 84.355 ms, Optimization 115.885 ms, Emission 87.947 ms, Total 290.161 ms
 Execution Time: 140040.834 ms
(31 rows)

Die CTE aus dem Beispiel in einen MV zum Test eingedost:

## \d+ newest_valid_measurement_mv
                                   Materialized view "public.newest_valid_measurement_mv"
  Column   |           Type           | Collation | Nullable | Default | Storage  | Compression | Stats target | Description
-----------+--------------------------+-----------+----------+---------+----------+-------------+--------------+-------------
 device_id | character varying(50)    |           |          |         | extended |             |              |
 timestamp | timestamp with time zone |           |          |         | plain    |             |              |
 type_id   | character varying(3)     |           |          |         | extended |             |              |
 value     | character varying(35)    |           |          |         | extended |             |              |
 mark_1    | character varying(15)    |           |          |         | extended |             |              |
 mark_2    | character varying(15)    |           |          |         | extended |             |              |
 mark_3    | character varying(15)    |           |          |         | extended |             |              |
 mark_4    | character varying(15)    |           |          |         | extended |             |              |
 mark_5    | character varying(15)    |           |          |         | extended |             |              |
 mark_6    | character varying(15)    |           |          |         | extended |             |              |
Indexes:
    "newest_valid_measurement_mv_device_idx" btree (device_id)
    "newest_valid_measurement_mv_mark6_idx" btree (mark_6)
View definition:
 SELECT DISTINCT ON (m.device_id) m.device_id,
    m."timestamp",
    m.type_id,
    m.value,
    mc.mark_1,
    mc.mark_2,
    mc.mark_3,
    mc.mark_4,
    mc.mark_5,
    mc.mark_6
   FROM measurement m
     LEFT JOIN measurementconfig mc ON m.type_id::text = mc.type::text
  WHERE mc.is_not_valid = false
  ORDER BY m.device_id, m."timestamp" DESC;
Access method: heap

# EXPLAIN ANALYZE SELECT
    d.device_id,
    d.note,
    m.value
FROM device d
    LEFT JOIN newest_valid_measurement_mv m
        ON (d.device_id = m.device_id)
WHERE m.mark_6 = 'ok';
                                                                   QUERY PLAN
-------------------------------------------------------------------------------------------------------------------------------------------------
 Hash Join  (cost=112081.81..174201.91 rows=1214312 width=26) (actual time=7429.966..11942.748 rows=1211544 loops=1)
   Hash Cond: ((m.device_id)::text = (d.device_id)::text)
   ->  Seq Scan on newest_valid_measurement_mv m  (cost=0.00..35881.91 rows=1272450 width=16) (actual time=5.410..1678.682 rows=1271950 loops=1)
         Filter: ((mark_6)::text = 'ok'::text)
         Rows Removed by Filter: 332923
   ->  Hash  (cost=82467.47..82467.47 rows=1531547 width=25) (actual time=7411.494..7411.494 rows=1527746 loops=1)
         Buckets: 131072  Batches: 16  Memory Usage: 6449kB
         ->  Seq Scan on device d  (cost=0.00..82467.47 rows=1531547 width=25) (actual time=0.167..6238.127 rows=1527746 loops=1)
 Planning Time: 1.348 ms
 JIT:
   Functions: 12
   Options: Inlining false, Optimization false, Expressions true, Deforming true
   Timing: Generation 0.482 ms, Inlining 0.000 ms, Optimization 0.353 ms, Emission 4.945 ms, Total 5.780 ms
 Execution Time: 11970.090 ms
(14 rows)

laurenz · 10. Januar 2025 um 11:44

Die langsame Abfrage könnte schneller werden, wenn man nur für die eine Abfrage enable_nestloop = off setzt.

Sonst wäre vielleicht eine Lösung mit Trigger interessant (die dritte Varainte von oben). Alternativ eine boolean-Spalte is_current und ein Trigger, der bei einer neuen Zeile die vorhergehende auf FALSE setzt.

akretschmer · 10. Januar 2025 um 12:23

um meinen Kollegen zu korrigieren: der Parameter nennt sich ‘enable_nestloop’