storage: partition MySQL string PK snapshots by key prefix

Replace the OFFSET-walking boundary discovery with the prefix-based
partitioner in mz-mysql-util, so discovery costs EXPLAIN index dives
instead of an O(rows) index pass.

Only string primary keys are supported. Integer keys, which the OFFSET
walk used to sample, now fall back to a single-worker whole-table read.
Prefixes of a numeric key do not order consistently with its values, so
they would need a separate numeric range splitter.

Boundaries are rendered as SQL literals via the server QUOTE() and
still pass the existing strict-monotonicity verification in each read
transaction.

The new mysql_source_snapshot_partition_min_rows dyncfg (default
50000) stops splitting below a minimum range size. Test configs set it
low so the tiny tables in mysql-cdc testdrive and parallel-workload
still exercise range reads.

Author: peterdukelarsen

misc/python/materialize/mzcompose/__init__.py

@@ -298,6 +298,11 @@ def get_variable_system_parameters(
         VariableSystemParameter(
             "mysql_source_snapshot_parallelism", "true", ["true", "false"]
         ),
+        # Low default so the tiny tables in tests still exercise PK-prefix
+        # range splitting; the production default only splits large tables.
+        VariableSystemParameter(
+            "mysql_source_snapshot_partition_min_rows", "2", ["2", "50000"]
+        ),
         VariableSystemParameter(
             "persist_batch_columnar_format",
             "structured" if version > MzVersion.parse_mz("v0.135.0-dev") else "both_v2",
@@ -701,6 +706,9 @@ def get_default_system_parameters(
     # The estimated path is covered explicitly in mysql-cdc/statistics.td and
     # by parallel-workload.
     "mysql_source_snapshot_exact_count_max_rows",
+    # Not varied here because the 256-prefix floor dominates for test-sized
+    # tables. parallel-workload flips it.
+    "mysql_source_snapshot_partition_probed_prefixes_per_billion_rows",
     "postgres_fetch_slot_resume_lsn_interval",
     "pg_schema_validation_interval",
     "pg_source_validate_timeline",

misc/python/materialize/parallel_workload/action.py

@@ -3046,6 +3046,20 @@ def __init__(
         self.flags_with_values["mysql_source_snapshot_parallelism"] = (
             BOOLEAN_FLAG_VALUES
         )
+        # 2 exercises PK-prefix splitting on workload-sized tables, the
+        # default leaves them in a single bucket.
+        self.flags_with_values["mysql_source_snapshot_partition_min_rows"] = [
+            "2",
+            "50000",
+        ]
+        # 0 leaves only the 256-prefix floor, the default scales with table
+        # size.
+        self.flags_with_values[
+            "mysql_source_snapshot_partition_probed_prefixes_per_billion_rows"
+        ] = [
+            "0",
+            "1000",
+        ]
 
         # If you are adding a new config flag in Materialize, consider using it
         # here instead of just marking it as uninteresting to silence the

src/storage-types/src/dyncfgs.rs

@@ -215,6 +215,26 @@ pub static MYSQL_SOURCE_SNAPSHOT_PARALLELISM: Config<bool> = Config::new(
     "Whether to split MySQL snapshot reads across workers by primary-key ranges.",
 );
 
+/// The smallest estimated row count for which the MySQL snapshot prefix
+/// partitioner keeps splitting a range; also the smallest table worth splitting.
+pub static MYSQL_SOURCE_SNAPSHOT_PARTITION_MIN_ROWS: Config<usize> = Config::new(
+    "mysql_source_snapshot_partition_min_rows",
+    50_000,
+    "Minimum estimated rows per range before MySQL snapshot PK-prefix partitioning \
+     stops splitting; also the smallest table considered worth splitting.",
+);
+
+/// Probed-prefix budget for the MySQL snapshot prefix partitioner, scaled to
+/// the table's estimated size, with a small floor. When a table's budget runs
+/// out, splitting stops early and buckets come out coarser, never incorrect.
+pub static MYSQL_SOURCE_SNAPSHOT_PARTITION_PROBED_PREFIXES_PER_BILLION_ROWS: Config<usize> =
+    Config::new(
+        "mysql_source_snapshot_partition_probed_prefixes_per_billion_rows",
+        1_000,
+        "Cap on MySQL snapshot PK-prefix partitioning probed prefixes per table, per billion \
+     estimated rows; when exhausted, splitting stops early with coarser buckets.",
+    );
+
 /// If the optimizer estimates the table has fewer rows than this, compute the exact row count
 /// with `COUNT(*)`. Otherwise, report the `information_schema` estimate directly.
 pub static MYSQL_SOURCE_SNAPSHOT_EXACT_COUNT_MAX_ROWS: Config<usize> = Config::new(
@@ -438,6 +458,8 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&MYSQL_REPLICATION_HEARTBEAT_INTERVAL)
         .add(&MYSQL_SOURCE_SNAPSHOT_EXACT_COUNT_MAX_ROWS)
         .add(&MYSQL_SOURCE_SNAPSHOT_PARALLELISM)
+        .add(&MYSQL_SOURCE_SNAPSHOT_PARTITION_MIN_ROWS)
+        .add(&MYSQL_SOURCE_SNAPSHOT_PARTITION_PROBED_PREFIXES_PER_BILLION_ROWS)
         .add(&ORE_OVERFLOWING_BEHAVIOR)
         .add(&PG_FETCH_SLOT_RESUME_LSN_INTERVAL)
         .add(&PG_SCHEMA_VALIDATION_INTERVAL)