It's the Memory Latency, Stupid

When you see your databases CPU at 100%, what is it actually doing? We akin to act as although CPUs run code linearly for simplicity's sake, but in fact CPUs are incredibly complex beasts; they can possibly do many instructions at once, or sometimes none at all (despite our console showing 100%). Because CPUs can do so many things, one CPU at 100% may be doing much small than another at 100%. A CPU is considered "idle" lone if no project (Linux task) needs CPU time; this does not average the CPU itself is being completely saturated. So why would a CPU not be completely utilized, equal if it appears to be at 100%?

The answer is normally memory.

Memory accesses are orders of dimension slower than inner CPU operations. To provision a fine image of how much so, a contemporary CPU power be capable to procedure location in the realm of 4-8 instructions per nanosecond. A recollection admission to RAM can obtain location in the realm of 50-100ns. To comfort your mathing, that method a contemporary CPU can do about 400-800 instructions for all one recollection admission from RAM.

Given this, CPUs are built from the dirt up to activity their way about this. The most apparent way this is dealt alongside is CPU caches. CPU caches keep recollection nearer to the CPU than RAM, exactly baked into the CPU chip. The closest cache (the L1 cache) takes about 1ns per recollection operation- that's 50-100x faster than RAM! The "problem" current of way is that the CPU caches are fairly small; the L1 cache for a contemporary CPU is likely about ~64KB, during the farther caches max out at a few hundred MB for top end processors. A fine repository tries manipulating data in small "chunks" to obtain complete advantage of caches, and you'll frequently comprehend cache use as being among the top concerns of achievement engineering.

Let's appearance at GROUP BY aggregations to get a improved grip on cache utilization. In GROUP BY, we're looking to obtain a dataset, collection it by a certain pillar (or more), and execute certain aggregations (for example, count, or sum on another column). The "naive" method power be to open a HashTable keyed by the collection key that grows always larger. For all new row, you would inspect if that key exists- if it does, go to it's corresponding aggregation and update the result. Simple, right? The issue is that if you have millions of distinctive groups, you're not using your cache at all! Nearly all HashTable admission volition be a "cold" hit that goes to RAM. This method that although your CPU may /look/ akin it's 100% utilized, in fact it'll continually be waiting for chilly recollection accesses.

To display this, we'll obtain our state-of-the-art Pivot motor and ravage it! We'll eliminate all our complex logic and substitute it alongside simplicity:

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() { // If the Group Key doesn't exist- set it to 0 // Afterwards, add 1 to the group- this volition provision us a count(*) GROUP BY key_column *self.counts.entry(key).or_insert(0) += 1; } }

We'll afterward run a uncomplicated query SELECT count(*), UserID FROM hits GROUP BY UserID on the "hits" dataset ClickBench provides (99m rows). This query volition provision us a figure per UserID, definition we'll have an admission in our HashMap per UserID- I'll provision a spoiler current that there are ~17m distinctive UserID.
Our achievement checks volition be on a naked metal element server from our fine allies complete at CherryServers (AMD EPYC 9124 Genoa, 16 cores, 128 GB DDR5). We unfortunately can't use a average AWS/GCP device since we desire admission to counters that aren't exposed there for safety reasons (for example, we desire to see how much recollection is passing from DRAM controller to us). We'll confine ourselves to a sole core, since this naive collection by can't run on many cores.

To run it:

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)"

This takes... a whopping 3.9 seconds. Is that good? Is that bad? Everything in this earth is relative, or so they say. If we run the query again (without restarting the server), it takes 3.6s. The second run is the one we're additional curious in- there's no disk engaged as everything is already sitting in our page-cache.

You'll notice we did taskset -c. This pins the procedure to a stated CPU; this volition authorize us to run perf stat on that CPU and see what's going on internally. We'll measure two metric groups- percent of CPU slots stalling since of recollection (and not doing item in the meanwhile), and percent of CPU slots really retiring a micro-op:

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)"

and we get rear this

Performance oppose 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 period elapsed

This tells us that 50.1% of the time, the CPU couldn't do item since it was stalling on memory! And lone 16.1% of the period the CPU was really retiring (aka doing) operations.

Another engaging stat to get is how many of our CPU recollection loads are coming from DRAM (and not from 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)"

And the result:

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

Every burden is 64 bytes (the minimum amount you can bring, a cache line), so we can multiply by 64 to see we pulled 4.5GB during this query from DRAM on CPU loads (this is without any prefetching).

Let's now try to enhance our algorithm to obtain advantage of the CPU caches. To do this, we'll use a method called radix partitioning. The idea is to divided the merge into two distinct phases:

Phase 1: In this phase, we're merely going to distinct the keys into 4096 distinct partitions (simply a Vec of SlabVecs- a SlabVec is a data construction akin to a Vec that is complete recollection that is already paged in. We don't use Vecs for raw recollection in Pivot since it creates a lot of variance; not item for this blog post!). The partition volition be decided by hashing the key and taking it's top 12 bits (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); // Since we desire top 12 bits, we change the hash (a u64) 52 bits to the right let partition = (hash >> 52) as usize; self.partitions[partition].push(&mut self.allocator, (hash, key)); } Ok(()) }

Phase 2: When all data has arrive through, go partition by partition, all period creating a HashMap, aggregating the partition into it, and afterward outputting it. This phase is extremely akin to our first algorithm, but we're going to use the hash from phase 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))) }

At archetypal glance, this resolution may appear fairly odd. We're doing more operations here, not less! Every row, in the two solutions, is hashed and sent into a HashMap. But in this solution, we're archetypal pushing rows into apparently meaningless vecs.

The key current (no pun intended) is to comprehend why we've done this; it's to create the aggregating HashMaps smaller. By dividing our collection by into 4096 pieces, we've made our operating HashMaps much likelier to fit into our CPU caches. Every period we do a lookup on our HashMap, our CPU is additional apt to discover the applicable recollection sitting in it's caches, which saves a big chunk of time. But why rely me? Let's run it!

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)"

And... Boom! 3.3 seconds, and afterward on second run 2.7 seconds! We've certainly made an improvement. But to really run this item home, let's run perf again:

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)"

And our results:

 Performance oppose 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 period elapsed

Now we can see the image has entirely changed- 23.5% of the period is spent on waiting for recollection (as opposed to 50% before), 32% for really retiring operations. You may notice we're really doing more work, not less- in this run, we did 19350459182 operations (the ex_ret_ops counter), whereas before the repository did 11431618219 operations. This is value it though, since as we stated before- if we can translate chilly DRAM accesses into CPU instructions, it's nearly continually a fine commerce to make. In this case, we went downward from 8364109137 CPU slots stalled since of recollection to 3790955223.

To complete the image and really see we're pulling small recollection from the DRAM, let's again inspect how many our recollection loads arrive from DRAM:

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)"

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

Whoo! That method that on petition we pulled 1.5GB from DRAM, alternatively of 4.5GB in the naive version.

That, of course, isn't the bounds of how much we can utilize our CPU cache for GROUP BYs. And using our CPU cache isn't the lone way we can "hide" DRAM latency; another trick is prefetching recollection to our cache before we need it, and yet another trick is creating algorithms that authorize maximization of recollection throughput- but that's not for this blog! We'll go into that in our next blog anywhere we appearance at our /seemingly extremely weird/ implementation of JOIN probing.

For now, without explaining too much, I'll depart you alongside Pivot's speed for this query on a sole core:

Time: 1848.353 ms (00:01.848)

and the amount of RAM we bring:

 Performance oppose stats for 'CPU(s) 3': 14917958 ls_dmnd_fills_from_sys.dram_io_near 2.182674864 seconds period elapsed

Which comes out to 954 MB.

Of course, Pivot knows to run in parallel on our 16 cores (where there are many another engaging cases), in which case it comes out to:

nice :)

Try out Pivot if you desire to have excellent speeds complete parquets (OSS fun!), see you next time.