replicators: Allow configuring max_parallel_snapshot_tables for MySQL.

This commit allows configuring the max_parallel_snapshot_tables
parameter for MySQL. It was fixed to 8, but now it can be configured
via cmd line or env variable. Defaults to number of cores -1.

Release-Note-Core: Allow configuring the number of parallel tables
  for MySQL snapshots via MAX_PARALLEL_SNAPSHOT_TABLES or
  --max-parallel-snapshot-tables.

Fixes: REA-5785
Closes: #1529
Change-Id: I7f18faf945b2a8c13cbbf879df7dfbb12bd274b2
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9715
Reviewed-by: Michael Zink <michael.z@readyset.io>
Tested-by: Buildkite CI

Author: altmannmarcelo

database-utils/src/lib.rs

@@ -135,7 +135,7 @@ pub struct UpstreamConfig {
     pub snapshot_report_interval_secs: u16,
 
     /// The maximum number of relations that will be snapshotted in parallel from the upstream
-    #[arg(long, hide = true)]
+    #[arg(long, env = "MAX_PARALLEL_SNAPSHOT_TABLES")]
     #[serde(default = "default_max_parallel_snapshot_tables")]
     pub max_parallel_snapshot_tables: Option<usize>,
 

replicators/src/mysql_connector/snapshot.rs

@@ -39,8 +39,6 @@ use std::collections::HashSet;
 
 const RS_BATCH_SIZE: usize = 1000; // How many queries to buffer before pushing to ReadySet
 
-const MAX_SNAPSHOT_BATCH: usize = 8; // How many tables to snapshot at the same time
-
 /// A list of databases MySQL uses internally, they should not be replicated
 pub const MYSQL_INTERNAL_DBS: &[&str] =
     &["mysql", "information_schema", "performance_schema", "sys"];
@@ -558,14 +556,15 @@ impl MySqlReplicator<'_> {
         db_schemas: &mut DatabaseSchemas,
         snapshot_report_interval_secs: u16,
         full_snapshot: bool,
-        _max_parallel_snapshot_tables: usize, // TODO: limit parallelism for MySQL
+        max_parallel_snapshot_tables: usize,
     ) -> ReadySetResult<()> {
         let result = self
             .replicate_to_noria_with_table_locks(
                 noria,
                 db_schemas,
                 snapshot_report_interval_secs,
                 full_snapshot,
+                max_parallel_snapshot_tables,
             )
             .await;
 
@@ -585,6 +584,7 @@ impl MySqlReplicator<'_> {
         db_schemas: &mut DatabaseSchemas,
         snapshot_report_interval_secs: u16,
         full_snapshot: bool,
+        max_parallel_snapshot_tables: usize,
     ) -> ReadySetResult<()> {
         // NOTE: There are two ways to prevent DDL changes in MySQL:
         // `FLUSH TABLES WITH READ LOCK` or `LOCK INSTANCE FOR BACKUP`. Both are not
@@ -628,6 +628,7 @@ impl MySqlReplicator<'_> {
             table_list,
             &replication_offsets,
             snapshot_report_interval_secs,
+            max_parallel_snapshot_tables,
         )
         .await
     }
@@ -678,6 +679,7 @@ impl MySqlReplicator<'_> {
         mut table_list: Vec<Relation>,
         replication_offsets: &ReplicationOffsets,
         snapshot_report_interval_secs: u16,
+        max_parallel_snapshot_tables: usize,
     ) -> ReadySetResult<()> {
         let mut replication_tasks = FuturesUnordered::new();
         let mut compacting_tasks = FuturesUnordered::new();
@@ -698,7 +700,7 @@ impl MySqlReplicator<'_> {
                 );
             }
 
-            if replication_tasks.len() == MAX_SNAPSHOT_BATCH {
+            if replication_tasks.len() == max_parallel_snapshot_tables {
                 break;
             }
         }
@@ -745,7 +747,7 @@ impl MySqlReplicator<'_> {
             }
 
             // If still have tables to snapshot add them to the task list
-            while replication_tasks.len() < MAX_SNAPSHOT_BATCH && !table_list.is_empty() {
+            while replication_tasks.len() < max_parallel_snapshot_tables && !table_list.is_empty() {
                 let table = table_list.pop().expect("Not empty");
                 if replication_offsets.has_table(&table) {
                     info!(