Het is de memory latency, idioot: GROUP BY versnellen door meer (CPU!) werk te verrichten

Het antwoord is meestal het geheugen.

Toegang tot het geheugen is orden van grootte langzamer dan interne CPU-operaties. Om een goed beeld te geven van het verschil: een moderne CPU kan ongeveer 4 tot 8 instructies per nanoseconde verwerken. Een geheugentoegang tot het RAM-geheugen kan echter 50 tot 100 nanoseconden duren. Om de rekensom makkelijker te maken: een moderne CPU kan ongeveer 400 tot 800 instructies uitvoeren voor elke enkele geheugentoegang vanuit het RAM.

ComponentLatentieCapaciteit / Efficiëntie
CPU core4–8 instr / ns-
L1 cache~1 ns~64 KB · ~4–8 instructies per toegang
L2 cache~3–4 ns~1 MB · ~15–30 instructies per toegang
L3 cache~10–15 nstot enkele honderden MB · ~50–100 instructies per toegang
DRAM~50–100 nsGB's · ~400–800 instructies per toegang

Gezien dit feit zijn CPU's van grond tot top ontworpen om dit probleem te omzeilen. De meest voor de hand liggende oplossing zijn de CPU-caches. Deze houden het geheugen dichter bij de CPU dan het RAM; ze zijn letterlijk in de CPU-chip ingebouwd. De dichtstbijzijnde cache (de L1-cache) kost ongeveer 1 ns per geheugenoperatie—dat is 50 tot 100 keer sneller dan RAM! Het "probleem" is uiteraard dat de CPU-caches vrij klein zijn; de L1-cache van een moderne CPU is waarschijnlijk ongeveer 64 KB, terwijl de verder gelegen caches bij high-end processors maximaal enkele honderden MB bedragen. Een goede database probeert data in kleine "chunks" te manipuleren om optimaal gebruik te maken van caches. Cache-benutting is dan ook een van de belangrijkste aandachtspunten bij performance engineering.

GROUP BY aggregaties en cache-benutting

Laten we kijken naar GROUP BY aggregaties om een beter grip te krijgen op cache-benutting. Bij een GROUP BY nemen we een dataset, groeperen we deze op basis van één of meerdere kolommen en voeren we aggregaties uit (bijvoorbeeld een count of sum op een andere kolom).

De "naïve" methode zou zijn om een HashTable te openen met de groepsleutel als key, die steeds groter wordt. Voor elke nieuwe rij controleer je of die sleutel al bestaat; zo ja, dan ga je naar de bijbehorende aggregatie en update je het resultaat. Simpel, toch?

Het probleem is dat als je miljoenen unieke groepen hebt, je totaal geen gebruik maakt van je cache. Bijna elke toegang tot de HashTable zal een "koude hit" zijn die naar het RAM moet. Dit betekent dat hoewel je CPU eruitziet alsof hij voor 100% benut wordt, hij in werkelijkheid constant staat te wachten op koude geheugentoegangen.

Om dit aan te tonen, nemen we de Pivot-engine en strippen we deze van alle complexe logica, om deze te vervangen door eenvoud:

fn consume(
    &mut self,
    batch: RecordBatch,
    _sender: &mut dyn Sender<RecordBatch>,
) -> unary::Result<()> {
    let group_keys = batch.column(self.key_column).as_primitive::<Int64Type>();
    for &key in group_keys.values() {
        // Als de Group Key niet bestaat, zet deze dan op 0
        // Voeg daarna 1 toe aan de groep - dit geeft ons een count(*) GROUP BY key_column
        *self.counts.entry(key).or_insert(0) += 1;
    }
}

We voeren vervolgens een simpele query uit: SELECT count(*), UserID FROM hits GROUP BY UserID op de "hits" dataset van ClickBench (99 miljoen rijen). Deze query geeft een telling per UserID, wat betekent dat we een vermelding in onze HashMap per UserID hebben. Ter spoiler: er zijn ongeveer 17 miljoen unieke UserID's.

De prestatiecontroles worden uitgevoerd op een bare-metal server (AMD EPYC 9124 Genoa, 16 cores, 128 GB DDR5). We gebruiken geen standaard AWS/GCP-machine omdat we toegang willen tot counters die daar om veiligheidsredenen verborgen blijven (bijvoorbeeld om te zien hoeveel geheugen er van de DRAM-controller naar ons stroomt). We beperken ons tot één enkele core, aangezien deze naïve group by niet op meerdere cores kan draaien.

Om dit uit te voeren:

cat > config.yaml <<'EOF'
server:
  bind: 127.0.0.1:5440
  workers: 1
datastores:
  default:
    kind: pivotlake
    location: ./data
    default: true
users:
  pivot:
    auth:
      method: trust
EOF

taskset -c 3 pivot server --config config.yaml &
taskset -c 7 psql -h 127.0.0.1 -p 5440 -U pivot -c '\timing on' -c "SELECT count(*) FROM (SELECT COUNT(*), UserID FROM hits GROUP BY UserID)"

Dit duurt... maar liefst 3,9 seconden. Is dat goed? Is dat slecht? Alles is relatief. Als we de query opnieuw uitvoeren (zonder de server te herstarten), duurt het 3,6 s. De tweede run is de belangrijkste; er is geen schijf involved omdat alles al in onze page-cache staat.

We hebben taskset -c gebruikt om het proces aan een specifieke CPU te pinnen. Hierdoor kunnen we perf stat op die CPU draaien om intern te zien wat er gebeurt. We meten twee metriekgroepen: het percentage CPU-slots dat vastloopt (stalling) vanwege het geheugen, en het percentage CPU-slots dat daadwerkelijk een micro-op voltooit (retiring):

sudo perf stat -C 3 -M backend_bound_memory,retiring -- taskset -c 7 psql -h 127.0.0.1 -p 5440 -U pivot -c "SELECT count(*) FROM (SELECT count(*), UserID FROM hits GROUP BY UserID)"

Het resultaat is als volgt:

Performance counter stats for 'CPU(s) 3':
 8364109137      ex_no_retire.load_not_complete   #     50.1 %  backend_bound_memory
 38233214040      de_no_dispatch_per_slot.backend_stalls
 8977647849      ex_no_retire.not_complete
 11861387798      ls_not_halted_cyc
 11431618219      ex_ret_ops                       #     16.1 %  retiring
 11858248045      ls_not_halted_cyc
 4.030148989 seconds time elapsed

Dit vertelt ons dat de CPU 50,1% van de tijd niets kon doen omdat hij stond te wachten op het geheugen! Slechts 16,1% van de tijd was de CPU daadwerkelijk operaties aan het voltooien (retiring).

Een andere interessante statistiek is hoeveel van onze CPU-geheugenladingen afkomstig zijn uit DRAM (en niet uit de caches):

sudo perf stat -C 3 -e ls_dmnd_fills_from_sys.dram_io_near -- taskset -c 7 psql -h 127.0.0.1 -p 5440 -U pivot -c "SELECT count(*) FROM (SELECT count(*), UserID FROM hits GROUP BY UserID)"

Het resultaat:

Performance counter stats for 'CPU(s) 3':
 76190837      ls_dmnd_fills_from_sys.dram_io_near
 3.996301634 seconds time elapsed

Elke load is 64 bytes (de minimale hoeveelheid die je kunt ophalen, een cache-line). Als we dit vermenigvuldigen met 64, zien we dat we tijdens deze query 4,5 GB uit het DRAM hebben getrokken via CPU-loads (zonder prefetching).

De oplossing: Radix Partitioning

Laten we nu proberen het algoritme te verbeteren om optimaal gebruik te maken van de CPU-caches. Hiervoor gebruiken we een methode genaamd radix partitioning. Het idee is om de merge op te splitsen in twee aparte fasen:

Fase 1

In deze fase scheiden we de sleutels in 4096 verschillende partities (een Vec van SlabVecs). De partitie wordt bepaald door de sleutel te hashen en de bovenste 12 bits te nemen ($2^{12} = 4096$).

fn consume(
    &mut self,
    batch: RecordBatch,
    _sender: &mut dyn Sender<RecordBatch>,
) -> unary::Result<()> {
    let keys = batch.column(self.key_column).as_primitive::<Int64Type>();
    for &key in keys.values() {
        let hash = self.hasher.hash_one(key);
        // Omdat we de bovenste 12 bits willen, verschuiven we de hash (een u64) 52 bits naar rechts
        let partition = (hash >> 52) as usize;
        self.partitions[partition].push(&mut self.allocator, (hash, key));
    }
    Ok(())
}

Visualisatie van Fase 1:

  • Input: UserID → hash(UserID)
  • Proces: partition = hash >> 52 (de eerste 3 hex-cijfers)
  • Resultaat:
  • Partitie 0 (0x000...): 4242 items
  • Partitie 1 (0x001...): 5 items
  • Partitie 2 (0x002...): 77997 items
  • ...
  • Partitie 4095 (0xFFF...): 13 items

Fase 2

Wanneer alle data is verwerkt, gaan we partitie voor partitie te werk. Per partitie maken we een HashMap, aggregeren we de partitie daarin en geven we deze vervolgens uit. Deze fase lijkt erg op ons oorspronkelijke algoritme, maar we gebruiken de hash uit fase 1.

fn into_outputter(self) -> unary::Result<Option<Self::Outputter>> {
    let mut groups = Vec::new();
    for rows in &self.partitions {
        let mut table: HashTable<(u64, i64, u64)> = HashTable::new();   // (hash, key, count)
        rows.for_each(|(hash, key)| {
            match table.entry(hash, |group| group.1 == key, |group| group.0) {
                Entry::Occupied(mut group) => group.get_mut().2 += 1,
                Entry::Vacant(slot) => { slot.insert((hash, key, 1)); }
            }
        });
        groups.extend(table.into_iter().map(|(_, key, count)| (key, count)));
    }
    Ok(Some(Outputter::new(groups)))
}

Op het eerste gezicht lijkt deze oplossing vreemd. We voeren immers méér operaties uit, niet minder. In beide oplossingen wordt elke rij gehasht en naar een HashMap gestuurd. In deze oplossing pushen we de rijen echter eerst naar schijnbaar betekenisloze vectors.

De sleutel tot succes is het begrijpen van het waarom: we doen dit om de aggregerende HashMaps kleiner te maken. Door onze group by in 4096 stukken te verdelen, is de kans veel groter dat onze actieve HashMaps in de CPU-caches passen. Elke keer dat we een lookup doen in onze HashMap, is de kans groter dat de CPU de relevante data in de caches vindt, wat enorm veel tijd bespaart.

Resultaten en Analyse

Laten we de nieuwe methode testen:

taskset -c 7 psql -h 127.0.0.1 -p 5440 -U pivot -c '\timing on' -c "SELECT count(*) FROM (SELECT COUNT(*), UserID FROM hits GROUP BY UserID)"

Het resultaat: 3,3 seconden, en bij de tweede run 2,7 seconden! Er is een duidelijke verbetering. Om dit te bewijzen, draaien we perf opnieuw:

sudo perf stat -C 3 -M backend_bound_memory,retiring -- taskset -c 7 psql -h 127.0.0.1 -p 5440 -U pivot -c "SELECT count(*) FROM (SELECT count(*), UserID FROM hits GROUP BY UserID)"

Nieuwe resultaten:

Performance counter stats for 'CPU(s) 3':
 3790955223      ex_no_retire.load_not_complete   #     23.5 %  backend_bound_memory
 20376118172      de_no_dispatch_per_slot.backend_stalls
 5442424751      ex_no_retire.not_complete
 10083617764      ls_not_halted_cyc
 19350459182      ex_ret_ops                       #     32.2 %  retiring
 10023435553      ls_not_halted_cyc
 3.575992317 seconds time elapsed

Het beeld is volledig veranderd: slechts 23,5% van de tijd wordt besteed aan wachten op het geheugen (tegenover 50% hiervoor), en 32% aan het daadwerkelijk voltooien van operaties. Opvallend is dat we daadwerkelijk méér werk doen: in deze run voerden we 19.350.459.182 operaties uit (exretops), terwijl de database daarvoor 11.431.618.219 operaties deed. Dit is echter een goede ruil, want als we koude DRAM-toegangen kunnen vertalen naar CPU-instructies, is dat bijna altijd winst. We zijn gedaald van 8,3 miljard CPU-slots die vastliepen door geheugen naar 3,7 miljard.

Om het plaatje compleet te maken, controleren we hoeveel geheugenladingen er uit DRAM komen:

sudo perf stat -C 3 -e ls_dmnd_fills_from_sys.dram_io_near -- taskset -c 7 psql -h 127.0.0.1 -p 5440 -U pivot -c "SELECT count(*) FROM (SELECT count(*), UserID FROM hits GROUP BY UserID)"

Resultaat:

Performance counter stats for 'CPU(s) 3':
 25688280      ls_dmnd_fills_from_sys.dram_io_near
 3.009547012 seconds time elapsed

Dit betekent dat we on-demand 1,5 GB uit het DRAM hebben getrokken, in plaats van 4,5 GB in de naïve versie.

Conclusie

Het optimaal benutten van de CPU-cache is niet de enige manier om DRAM-latentie te "verstoppen". Andere technieken zijn prefetching (geheugen naar de cache laden voordat het nodig is) en het creëren van algoritmen die de geheugen-throughput maximaliseren.

Ter illustratie, dit is de snelheid van Pivot voor deze query op een enkele core: Tijd: 1848,353 ms Geheugen uit DRAM: 954 MB

En wanneer Pivot parallel draait op 16 cores, komt de tijd uit op: Tijd: 171,773 ms