sources: Speed up MySQL ingestion by parallelizing#35569
Conversation
|
Thanks for opening this PR! Here are a few tips to help make the review process smooth for everyone. PR title guidelines
Pre-merge checklist
|
9c63b2a to
f7effd1
Compare
def-
changed the title
2ee8d42 to
2e9f0da
Compare
def-
changed the title
def-
changed the title
db20263 to
f97084a
Compare
f97084a to
96687b4
Compare
11d7b39 to
a7eed48
Compare
Partition each table's primary-key range across timely workers so they read disjoint PK ranges of the initial snapshot concurrently, reducing initial-load time for large tables with a single-column integer primary key. A snapshot leader establishes the consistent point and broadcasts SnapshotInfo to all workers over a timely feedback loop; each worker then reads its range under a CONSISTENT SNAPSHOT transaction. Tables without a suitable PK fall back to single-worker-per-table mode. Also adds a MySqlInitialLoadMultiWorker feature benchmark. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Generalize PK-range snapshot partitioning beyond single-column integer primary keys to also cover non-integer (char/varchar/text) and composite PKs, so large tables keyed by strings (e.g. ULIDs) or multi-column keys also load their initial snapshot in parallel across workers instead of falling back to a single worker. Single-column integer PKs keep the cheap MIN/MAX split. Other supported PKs sample partition boundaries by keyset pagination over the primary-key index (one forward pass), reading each boundary key back already rendered as a SQL literal (QUOTE() for text, CAST(.. AS CHAR) for numeric) so range predicates compare under each column's own collation, matching the ORDER BY used to pick them. Composite keys use row-value comparisons. Every boundary is found with a strict `pk > prev`, so boundaries strictly increase and the resulting half-open ranges are disjoint and cover the entire key domain: each row is read by exactly one worker regardless of data distribution. Boundary computation runs before LOCK TABLES so the O(rows) sampling scan for non-integer PKs does not extend the table-lock window; correctness does not depend on boundaries matching the exact snapshot point. Also fix a snapshot duplication bug: when a partitioned table has fewer partitions than the cluster has workers, the worker responsible for the table could fall through to the `responsible_for` path and read the whole table in addition to the range-owning workers, duplicating every row. A read plan now distinguishes range-owning workers, the single responsible worker of an unpartitioned table, and a responsible worker that owns no range (which reads nothing but still emits the table's rewind request). Adds char and composite PK correctness cases to mysql-cdc.td and a test/mysql-pk-benchmark composition that demonstrates the speedup under a bandwidth-limited MySQL link (the regime where parallel range reads help). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Now also implementation and testing for non-integer and composite keys, numbers from this CI run: https://buildkite.com/materialize/nightly/builds/17115: |
|
Ran with parallelism 2: commit tweaking test: 5517b8a |
peterdukelarsen
left a comment
peterdukelarsen
left a comment
There was a problem hiding this comment.
Let's simplify this a bit by focusing just on a single-column primary key and using just one method for computing partitions (the one that supports the char(26) type). Let's flesh out the serde testing for the various types a little more aggressively to try to root out issues with round-tripping the stringified value -- including testing different character sets.
| let row: Option<MySqlRow> = conn | ||
| .query_first(format!( | ||
| "SELECT {cols_literal} FROM {table}{predicate} \ | ||
| ORDER BY {cols_ident} LIMIT 1 OFFSET {offset}" |
There was a problem hiding this comment.
I think the offset results in an O(offset) scan, so across all partitions you'll have an O(N) scan here. Discussed with some S&S folks and we don't think this is a strict blocker because the decoding is a big part.
| return ( | ||
| "CREATE TABLE pk_table " | ||
| "(region CHAR(2) NOT NULL, id BIGINT NOT NULL, f2 BIGINT, " | ||
| "PRIMARY KEY (region, id));" |
There was a problem hiding this comment.
would be interesting to try:
PRIMARY KEY (region ASC, id DESC));
I suspect this would cripple performance when running ORDER BY region, id (which I think defaults to ORDER BY region ASC, id ASC) on a large table because it will trigger a Filesort.
| f"--binlog-row-metadata={binlog_row_metadata}", | ||
| f"--server-id={server_id}", | ||
| "--max-connections=500", | ||
| f"--max-connections={max_connections}", |
There was a problem hiding this comment.
Any sense of how risky this is for customer mysqls? Maybe we want a docs change here. I did a little legwork and most clusters cap out at about 64 workers.
| //! We have therefore succeeded in starting a transaction at a known point in time! | ||
| //! For tables with a suitable primary key, the leader computes `worker_count - 1` boundary keys | ||
| //! that split the key domain into disjoint half-open ranges, and broadcasts them. Each worker | ||
| //! reads only its assigned range. A single-column integer PK uses a cheap `MIN(pk)`/`MAX(pk)` |
There was a problem hiding this comment.
I think we should remove this feature for now and stick with the table-scanning approach for simplicity. The table-scanning approach will be much more resilient to uneven distributions. Let's leave adding this as an optional optimization to consider in the future.
| } | ||
|
|
||
| /// Classify a scalar type for range splitting, or `None` if unsupported. | ||
| fn pk_col_kind(scalar_type: &SqlScalarType) -> Option<PkColKind> { |
There was a problem hiding this comment.
Why stick to such a limited set of types?
There was a problem hiding this comment.
Did a little claude research and it sounds like float/double conversions to string could be lossy. DECIMAL should work. BINARY and similar types likely won't cast cleanly to a String. ENUM may not compare the same way and could be risky. There were some other types considered too, but generally makes sense that round-tripping to String is not guaranteed to work for all types and that there may be more complex concerns than just round-tripping to String for other types (like ENUMs).
Would be interested if there was a way we could avoid round-tripping to a string to make support easier to extend correctly? i.e. can we keep it in the Value type of the my_sql library?
There was a problem hiding this comment.
Moving to the Value type definitely not required and might lose us some protection of string conversion -- namely enforcing UTF8. Regardless, we should test with a couple of different charsets to make sure this is resilient to that -- or at least isn't a regression.
| .iter() | ||
| .map(|(c, kind)| match kind { | ||
| PkColKind::Numeric => format!("CAST({c} AS CHAR)"), | ||
| PkColKind::Text => format!("QUOTE({c})"), |
There was a problem hiding this comment.
Can we avoid to the string casting and round-trip of strings by holding onto the typed value as a parameter? This will add some complexity to the PkSplits broadcast because now we'll need to serde the full type rather than a simple string, but it could make supporting other types simpler and less error prone.
| // stay correct for any data distribution (see `PkSplits`). | ||
| // `None` means single-worker fallback. | ||
| let mut pk_bounds_map = BTreeMap::new(); | ||
| for (table, outputs) in &tables_to_snapshot { |
There was a problem hiding this comment.
Let's open up a transaction for the computation of all of the partitioned tables here and make sure it's not closed until after the locks are acquired, the other workers have snapshotted, and the locks are released.
The concern driving this suggestion is around collation changes to the char primary key. If collation is altered between computing the partition boundaries and executing our snapshot there's a small risk we could have duplicate or dropped data.
| let splits = if config.worker_count < 2 { | ||
| None | ||
| } else if let Some(pk_col) = integer_pk_col(desc) { | ||
| compute_integer_splits( |
There was a problem hiding this comment.
Given the linear cost of reading these tables being imposed on the leader, I'm slightly concerned we could have a regression in snapshot performance, i.e. imagine the case where snapshotting a small table would have been executed successfully within a minute or two but is now blocked behind computing the partitions for the largest table.
One option for improving on this could be some version of concurrency control locally since all the work happens in mysql, i.e. we could submit up to #workers of parallel tasks to compute the partitions which would bring us closer to historical performance.
| .responsible_for(table) | ||
| .then_some(ReadPlan::RewindOnly), | ||
| }, | ||
| _ => config |
There was a problem hiding this comment.
I think we should error on the None case here because we insert something for each table up top.
| ) | ||
| .await?; | ||
| // Phase G: The leader publishes the full snapshot size so the summed | ||
| // worker-local gauges reflect the upstream total without double-counting. |
There was a problem hiding this comment.
code style nit: I have found these phases somewhat useful, but I wonder if we could get similar benefit with less verbose comments by factoring out some functions to function a little smaller
|
Closing in favor of #37642 |
### 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. --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2 participants
On a cluster with 8 workers, running on my 8 core / 16 thread dev server:
Co-written with Claude 🤖