Parallelizing MySQL Snapshots

A key customer recently came to us because loading data from MySQL into Materialize would run for over 40 hours and then fail. Materialize ingests data by first “snapshotting” existing data, then switching to a stream of diffs from MySQL’s binlog. The initial snapshot took so long that by the time it finished, MySQL couldn’t provide diffs going back to when the snapshot had started.

Even with enough disk to retain days’ worth of diffs, the initial ingestion time is also the disaster recovery time, so an over 40-hour recovery time would be unacceptable.

The customer’s data skewed towards a single table with over two billion rows and two terabytes of data. Our Postgres integration can handle this scale and shape of data because it supports parallelizing snapshots within a single table, but the MySQL integration wasn’t as mature. We knew the solution was to parallelize the initial snapshot for MySQL, but we didn’t know how to partition the table efficiently so parallel readers could share the load.

We started with a Claude-generated PR, optimistic we could unblock the customer in a week. In reality, getting to a performant solution took over a month of iteration as we learned about the customer’s workload, ran scale tests, investigated bottlenecks, and fixed correctness issues caused by the complexities of MySQL string sort ordering. This post shares the missteps we made and the lessons learned along the way.

In our largest internal scale tests we saw anywhere from a 2.2 to 6.9x improvement, depending on data shape and MySQL instance size:

In all of these tests Materialize saturated MySQL’s input/output operations per second (IOPS) at full parallelism, so we likely could have seen more improvement with more provisioned IOPS. Our target was a 4x improvement, so we were happy with the top line numbers.

For our customer, we brought the snapshot for their largest table down to just three hours! They reported a roughly 10x speedup based on their estimate that the largest table had previously taken around 30 hours to snapshot.

Finding the right values to evenly partition the customer’s primary key space turned out to be a core challenge.

Months before we started pushing on this in earnest we actually had a large Claude-generated PR for parallelized snapshotting. It just needed a review. The PR was tantalizing because all we needed to do was read the code, approve, and then we’re good to go, right? We’re good to go, right??

The initial approach was straightforward and only supported INT and BIGINT primary keys. You select the MIN and MAX and then interpolate. Unfortunately, we quickly learned the customer we wanted to support was using CHAR(n) primary keys, so we focused on getting that to work out of the box first. One major idea from the Claude-generated solution that did work was partitioning along the primary key.

In InnoDB, MySQL’s default storage engine, primary keys are clustered B+ tree indexes. "Clustered" in this context means that all columns for each row are stored inline. This is different from non-clustered indices that only store the columns specified for the index and the primary key as a pointer back to the full row. The primary key’s B+ tree index stores the rows sorted by the primary key in the leaf pages of the index. Having the primary key index means

For example, imagine the primary keys were common key shapes like ULIDs or UUIDs. The underlying B+ tree would look something like this:

MySQL loads full pages at a time from storage if they aren’t in the cache. Because of this, reading just one row or a subset of rows in a page isn’t cheaper than reading all rows from the page in terms of pages read from storage. Similarly, when reading from a clustered index where all columns are inlined, selecting a subset of the columns doesn’t reduce the pages to be read.

So assuming perfect partitioning for four workers, with this example we’d issue 4 queries:

In this case we’d only read each leaf page once, resulting in excellent efficiency. Ok great, the B+ tree has the rows nicely sorted and we can just portion out equal-sized regions of the keyspace to our workers. It sounds simple, but how do we select the right boundary keys?

One approach we considered and thankfully excluded early was hashing the primary key to distribute work across Materialize workers (i.e. SELECT * FROM table WHERE CRC32(id) % <worker_count> = <worker_index>). The rows each worker would read would be spread throughout the pages, likely resulting in all workers reading all pages from the database. If we had 16 workers we might end up pushing up to roughly 16x load into the upstream MySQL, which as we’ll see is a big problem.

We had a general sense that strings in MySQL are complex, so we wanted to avoid directly handling, generating, or sorting strings in Materialize as much as possible.

Our first approach to partitioning that merged was selecting primary keys from the table spaced according to the number of rows each worker should own. We had a sense up front that string sorting (a.k.a. collation) in MySQL was complex, so we wanted to avoid directly sorting or generating strings in Materialize as the original PR had done with integer interpolation.

For example, for four workers on the table above:

This had the major advantage of not relying on understanding string sort order in Materialize. We were already running SELECT COUNT(*) to report table size statistics, so it was “just” an extra pass over the primary key index to compute the boundary keys.

Performance testing against 200M-row table on an AWS RDS MySQL instance looked great. On a 400cc (8-worker) instance we observed a 3x speedup (~11m → ~3.5m) with minimal load on the RDS instance. This “confirmed” our initial hypothesis that Materialize was the bottleneck, and we were excited to ship.

Unfortunately, it wouldn’t be this clean and easy. Once we started testing against larger tables where we needed parallelism the most, it became apparent that MySQL was the bottleneck and not Materialize.

Hints we had a problem surfaced as we set up a narrow 1B row table to test against. Somewhere between 200M and 500M rows performance fell off a cliff, going from around nine seconds to around nine minutes for running a SELECT COUNT(*):

-- AWS RDS db.r6g.2xlarge

2 mysql> select count(*) from t;

3 +-----------+

4 | count(*) |

5 +-----------+

6 | 200000000 |

7 +-----------+

8 1 row in set (9.50 sec)

9

10 mysql> select count(*) from t;

11 +-----------+

12 | count(*) |

13 +-----------+

14 | 530000000 |

15 +-----------+

16 1 row in set (8 min 52.83 sec)

Copy sql

Both SELECT COUNT(*) and SELECT id .... OFFSET ... work by traversing all rows in an index. The slowdown was accompanied by metrics showing an increase in IOPS which shows how many reads and writes MySQL was running against the disk.

We concluded that MySQL had been caching the dataset in memory and that the performance degraded once the dataset exceeded the available cache. The “cache” in InnoDB is called the “buffer pool.” The AWS RDS MySQL instance we had set up at the time was a db.r6g.2xlarge with 64GB of memory. RDS automatically configures the buffer pool to 75% of memory (48GB) so it makes sense that at 500M rows and ~85GB it was no longer cached like it was at 200M rows and ~34GB.

Before we even ran the tests my heart sank. I knew from playing with MySQL at this point that COUNT(*) performance scaled similarly to the OFFSET queries we were running, so we might be SOL.

The legacy snapshot of 1B rows took 1h50m, while the parallelized version took 2h. Back to the drawing board.

We could clearly see the different phases of the snapshot in the graphs of IOPS. It turns out that COUNT(*) will run with parallelism in MySQL and we had the default minimum parallelism of 4.

We could also see that partition computation phase where we run SELECT ... OFFSET ... took just as long to run as the legacy serial worker’s snapshot! This indicated that the bottleneck was MySQL. You can see in each case we’re stuck at ~1500 IOPS for ~1h30m at steady state.

The one silver lining was that we could see the parallelized snapshot itself was quite effective for this test case.

We didn’t give up immediately because the results weren’t all strictly worse. It made significant improvements when the full dataset was cached. It also worked well against the test cases with wider tables (2.2B rows with 2.45TB data instead of 1B rows with 171GB). The OFFSET query on a large instance peaked at nearly 40k IOPS, while the legacy single-threaded snapshot topped out around 10k IOPS, so there was some room for optimization. On the largest cluster, we got a ~2x performance improvement. Things were better, and we tried to tell ourselves that this would be good enough for the customer, but deep down we knew that there was more work to do.

At this point we decided to scrap this approach and look for something that could get us a true 4x+ improvement without increasing upstream load. We couldn’t come up with anything.

Our manager Nick brought up the idea of benchmarking against MyDumper because it’s a best-in-class implementation of MySQL export. He thought it would crush our performance and give us some inspiration. On the off chance MyDumper didn’t have a better approach, that might be a signal to set the project aside.

We quickly found a couple of novel ideas in MyDumper. They did have some form of string primary key partitioning implemented!

Specifically, a couple of query fragments that jumped out were:

The high level idea this sparked was trying to break down a table with a string primary key into ranges based on the key prefixes. We jumped right at trying to replicate this approach.

At a high level, the algorithm we came up with inspired by MyDumper’s approach is broken down into a couple of steps:

We had learned from the late-in-the-game failure of the offset walking approach and went straight to trying out the queries borrowed from MyDumper. Unfortunately this query hung on our test dataset:

SELECT LEFT(id, 1) FROM t WHERE id > '0' AND id NOT LIKE '0%' ORDER BY id ASC LIMIT 1;

Copy sql

For context, our test table was populated with 2.2B rows worth of ULIDs, which were all prefixed with “01K” before any divergence in the characters. This query for finding the next prefix after “0” was triggering a full table scan. We had been running on fumes, propelled by the hope we could rip off MyDumper, but instead we found ourselves tumbling back down into the pits of despair.

While the humans were feeling emotions, Claude was still churning in the background firing off queries at a local MySQL instance running in docker. Claude would bail us out.

We were able to keep lookups fast for our test cases using two queries instead of the single query (id > '<prefix>' AND id NOT LIKE '<prefix>%') MyDumper used:

The relevant term here is “sargability” (Search ARGument ABLE), which just means the database can do a quick index dive (traversing down the B+ tree in logarithmic time) rather than a slow linear scan evaluating each row individually. The relevant MySQL docs are under “range optimization”. They don’t directly describe why NOT LIKE isn’t sargable while LIKE is, but it certainly wasn’t.

The final algorithm was composed from these sargable queries:

So here’s a fresh example of the prefix probing approach now with the queries plugged in:

We knew going into this that EXPLAIN was approximate, but it was both much better and much worse than expected. It was much better than expected because it actually based estimates on index dives down to samples of live data. It was much worse than expected because we saw major underestimates for queries touching a lot of data and significant over-estimates in some cases.

For example, we knew our table had 2.2B rows, and the information_schema.tables size estimate gave us this:

mysql> SELECT table_name, tabl