[SS-97] MySQL Parallel Snapshot

[peterdukelarsen]

Motivation

Taking over implementation of #35569.

For customers with large tables in MySQL the initial snapshot can be painfully slow. We can improve performance by distributing the initial snapshot across workers.

Description

Previously we would:

1. Partition ownership of tables across workers
2. Each worker would take out a read lock on the tables it owned, grab the gtid, open a txn and then release the readlock to let us get the gtid of the transaction
3. Copy the entirety of the tables it owned one at a time

The problem with this is when one table is huge we are as slow as the single worker copying the whole table.

Now the MySQL snapshot logic at a high level will be:

1. Have a leader compute partitions for tables where possible that can be shared across workers (more on this later)
2. The leader takes a read lock on all tables to snapshot, grabs the current GTID, then broadcasts the gtid and partitions
3. All workers including the leader will connect to mysql and open transactions while the tables are locked, then drop their capability to indicate to the leader they've opened the snapshot
4. The leader will drop the read locks at this point unblocking writes to the source tables
5. All workers will perform snapshots of the tables or sections of tables they own

The leader will parallelize the partition computation using multithreading. The partition computation is much faster than actually reading the full snapshot, but it is slow and scales with the table size. The leader will step through the table along the primary key sort order using offsets of 1 / # workers to compute partitions of approximately the same size to use for partitioning the table.

Verificiation

• Testdrive tests

Benchmark Performance

Some basic benchmarks: https://buildkite.com/materialize/nightly/builds/17371/list. First two show it hasn't gotten slower for many tables while the last demonstrates the case of a single large table being sped up. This shows a 2.7x speedup on a large table without a slowdown on many tables (although it's possible there could be a slowdown of ~10% on perfectly even distribution of tables due to the extra work of partitioning).

NAME                                   | TYPE            |      THIS       |      OTHER      |  UNIT  | THRESHOLD  |  Regression?  | 'THIS' is
--------------------------------------------------------------------------------------------------------------------------------------------
MySqlInitialLoadMultiWorkerSampled     | wallclock       |           2.834 |           3.389 |   s    |    10%     |      no       | better: 16.4% faster

MySqlInitialLoadMultiWorkerSampled     | memory_mz       |        1379.387 |        1106.102 |   MB   |    60%     |      no       | worse:  24.7% more

MySqlInitialLoadMultiWorkerSampled     | memory_clusterd |          86.762 |          64.301 |   MB   |    60%     |      no       | worse:  34.9% more


MySqlInitialLoadMultiWorkerSingleTable | wallclock       |           0.813 |           2.233 |   s    |    10%     |      no       | better:  2.7 times faster

MySqlInitialLoadMultiWorkerSingleTable | memory_mz       |        1125.410 |         958.266 |   MB   |    60%     |      no       | worse:  17.4% more

MySqlInitialLoadMultiWorkerSingleTable | memory_clusterd |         216.004 |         213.613 |   MB   |    60%     |      no       | worse:   1.1% more

Known trade-offs/risk areas

1. Snapshotting may be ~10-20% slower for an existing workload if the work is well-distributed. Walking the tables to compute partition boundaries is an extra step that only improves wallclock performance when there are underutilized workers.
2. MySQL may experience more sustained periods of high load in a customer environment due to the additional table scan up-front and due to the better utilization of workers on previously skewed workloads.
3. Table read locks will be held for longer. The read locks are now all taken out by the leader and transactions need to be started by all workers before we release that read lock. This increases the duration during which the read locks are held to the slowest worker to start a transaction. It also decreases the granularity of read locking, so could result in some tables being locked for longer while we're waiting to acquire the lock on more actively written to tables.
4. For places where we weren't fully utilizing the worker count -- i.e. 16 workers but 2 tables -- we'll be increasing the number of connections to mysql when we distribute the snapshot load across workers. MySQL by default has a relatively low max_connection count and Materialize will fail to connect if MySQL is saturated, potentially resulting in a retry loop.

[peterdukelarsen]

Do we also want hte mysql_source_snapshot_max_execution_time to apply to the PK boundary sampling? Currently it seems not to.

I think we do want to add this too -- making this change now.

[def-]

Given that we found a bunch of bugs, should we maybe have a way to fall back to non-parallel snapshotting with an LD flag, just in case?

[peterdukelarsen]

Yeah, I'm open to a kill switch. I can easily fall back to the old partitioning logic with a flag. It might be a little harder to fall back to the old way of handling locking and opening up read transactions. I'll draft the easier version, but if anyone feels strongly about avoiding or having a kill switch for the new structure I can scope that out.

[peterdukelarsen]

Pulled testdrive changes out here: #37819 to shrink the diff a little bit.

[martykulma]

Looks great @peterdukelarsen! The main concern I have is around all the extra work we now do while holding the lock for all tables. There's some low hanging fruit there in connection setup that we should address.

While not a blocker, I think we should give some serious considering to moving away from COUNT(*) for row count. As it stands, the implementation will perform 2 full index scans (one for the count, one for the offset queries) and then a full table scan (snapshot).

count is used in 2 places:

1. statistics (and PG already uses an estimate for tables with > 1m rows)
2. partitioning

My understanding is that a stale count affects the effectiveness of the partitioning and the accuracy of the statistics reporting, but it won't affect correctness.

[martykulma]

We should try and minimize the time we're holding the lock. Right now, it looks like we pay connection costs for each worker under the lock. We can be more aggressive here and open the connections ahead of time. If there's no work, we'll just close them.

[peterdukelarsen]

That seems reasonable to me! Only downside to call out is we'll now have a higher peak of open connections, ~2x worker count (workers holding open a conn while waiting for the lock signal overlapping with the connections the leader uses to parallelize partition selection). I don't think that's a blocker.

[peterdukelarsen]

Done b06b134

[martykulma]

While not a blocker, I think we should give some serious considering to moving away from COUNT(*) for row count

Forgot to mention that I'm good with adding that in a separate PR to avoid any additional complexity in this one.

[martykulma]

nice - lgtm!

Non-blocking, but small (and probably worth it). Those connections are going to sit idle for the duration of the partitioning. User may have overridden the default, so explicitly setting the wait timeout once those connections are established for defense.

SET @@session.wait_timeout = 28800 (this is the default )

Seeing if this unblocks merging. I've addressed all of Dennis' review comments and he's approved the tests I split out to the next PR.