Module snapshot 

Renders the table snapshot side of the MySqlSourceConnection dataflow.

Snapshot reading

Depending on the source_outputs resume_upper parameters this dataflow decides which tables to snapshot and performs a simple SELECT * FROM table on them in order to get a snapshot. There are a few subtle points about this operation, described below.

It is crucial for correctness that we always perform the snapshot of all tables at a specific point in time. This must be true even in the presence of restarts or partially committed snapshots. The consistent point that the snapshot must happen at is discovered and durably recorded during planning of the source and is exposed to this ingestion dataflow via the initial_gtid_set field in MySqlSourceDetails.

Unfortunately MySQL does not provide an API to perform a transaction at a specific point in time. Instead, MySQL allows us to perform a snapshot of a table and let us know at which point in time the snapshot was taken. Using this information we can take a snapshot at an arbitrary point in time and then rewind it to the desired initial_gtid_set by “rewinding” it. These two phases are described in the following section.

Producing a snapshot at a known point in time.

Ideally we would like to start a transaction and ask MySQL to tell us the point in time this transaction is running at. As far as we know there isn’t such API so we achieve this using table locks instead.

A designated leader worker acquires table locks on all the tables to be snapshotted. By doing so we establish a moment in time where we know no writes are happening to the tables we are interested in. The leader then reads the current upper frontier (snapshot_upper) using the @@gtid_executed system variable and broadcasts it, along with PK-range bounds (see below), to all workers via a timely feedback loop. This frontier establishes an upper bound on any possible write to the tables of interest until the lock is released.

Each worker now starts a transaction via a new connection with ‘REPEATABLE READ’ and ‘CONSISTENT SNAPSHOT’ semantics. Due to linearizability we know that this transaction’s view of the database must some time t_snapshot such that snapshot_upper <= t_snapshot. We don’t actually know the exact value of t_snapshot and it might be strictly greater than snapshot_upper. However, because this transaction will only be used to read the locked tables and we know that snapshot_upper is an upper bound on all the writes that have happened to them we can safely pretend that the transaction’s t_snapshot is equal to snapshot_upper. We have therefore succeeded in starting a transaction at a known point in time!

The leader verifies each output’s schema against the planning-time desc before locking, and each worker re-verifies in its transaction, retrying transiently if the schema drifted since.

Once all workers have started their transactions the leader unlocks the tables. Each worker then reads the snapshot of the tables (or PK ranges) it is responsible for and publishes it downstream.

TODO: Other software products hold the table lock for the duration of the snapshot, and some do not. We should figure out why and if we need to hold the lock longer. This may be because of a difference in how REPEATABLE READ works in some MySQL-compatible systems (e.g. Aurora MySQL).

Parallel PK-range snapshots

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. Ranges are assigned round-robin starting from each table’s legacy single-worker owner, so the open-ended ranges (which absorb any rows written past the last sampled boundary) land on a different worker per table rather than always the last worker. Tables without a suitable PK fall back to single-worker-per-table mode. The mysql_source_snapshot_parallelism dyncfg disables splitting entirely, putting every table in that fallback mode. Workers open their connections while the leader samples and locks, so setup briefly holds up to 2 * worker_count + 1 upstream connections per source, settling to one per ranged worker plus the leader’s lock connection. To handle various charsets and collation gracefully we rely on MySQL’s sort order and never attempt to compare or order strings in Rust. To handle possible races with changes to collation each worker validates in its read transaction that the boundaries are strictly increasing under the table’s current collation, retrying transiently if not. The repeatable read snapshots should then succeed if they start reading from the table before DDL runs, or if DDL does run before one of the workers reads the table, that worker’s transaction should fail with an ER_TABLE_DEF_CHANGED.

Rewinding the snapshot to a specific point in time.

Having obtained a snapshot of a table at some snapshot_upper we are now tasked with transforming this snapshot into one at initial_gtid_set. In other words we have produced a snapshot containing all updates that happened at t: !(snapshot_upper <= t) but what we actually want is a snapshot containing all updates that happened at t: !(initial_gtid <= t).

If we assume that initial_gtid_set <= snapshot_upper, which is a fair assumption since the former is obtained before the latter, then we can observe that the snapshot we produced contains all updates at t: !(initial_gtid <= t) (i.e the snapshot we want) and some additional unwanted updates at t: initial_gtid <= t && !(snapshot_upper <= t). We happen to know exactly what those additional unwanted updates are because those will be obtained by reading the replication stream in the replication operator and so all we need to do to “rewind” our snapshot_upper snapshot to initial_gtid is to ask the replication operator to “undo” any updates that falls in the undesirable region.

This is exactly what RewindRequest is about. It informs the replication operator that a particular table has been snapshotted at snapshot_upper and would like all the updates discovered during replication that happen at t: initial_gtid <= t && !(snapshot_upper <= t). to be cancelled. In Differential Dataflow this is as simple as flipping the sign of the diff field.

The snapshot reader emits updates at the minimum timestamp (by convention) to allow the updates to be potentially negated by the replication operator, which will emit negated updates at the minimum timestamp (by convention) when it encounters rows from a table that occur before the GTID frontier in the Rewind Request for that table.

Structs

PkBoundaries 🔒

PkRange 🔒

SnapshotInfo 🔒

TableStatistics 🔒

Enums

ReadPlan 🔒

What a worker does for one table during the snapshot.

Functions

boundaries_strictly_monotonic 🔒

Whether boundaries are strictly increasing under collation. The half-open PK ranges only partition the table without gaps or overlaps when this holds. The boundaries are already SQL literals (from QUOTE()); each is coerced to charset/collation so the comparison uses the same collation as the column, matching the read predicates. Fewer than two boundaries are trivially monotonic.

build_snapshot_query 🔒

Builds the SQL query to be used for creating the snapshot using the first entry in outputs.

collect_table_statistics 🔒

Row count for the snapshot size gauge. Tables whose optimizer row estimate exceeds exact_count_max_rows report the estimate directly, everything else is counted exactly with COUNT(*). The gauge only drives progress reporting, and an estimate is a fair trade for skipping an O(rows) index walk on a large table.

compute_sampled_splits 🔒

Walks the primary key index in steps of about row_count / worker_count, taking the key at each step’s OFFSET. The per-step OFFSET scans sum to a full index pass, so this function has a time complexity of O(row_count). Worker count is small, so the OFFSET scans dominate the runtime. row_count can be an optimizer estimate for large tables, so the partitions are approximate. An overestimate walks off the end of the index and stops with fewer boundaries, resulting in some workers receiving less or no work. An underestimate leaves a larger final partition for the last worker, however both still correctly partition the table. Returns None if the primary key column type is not supported or the table is too small to split.

fetch_column_collation 🔒

Character set and collation of column in table, or None if the column has no collation (numeric/temporal types sort independently of collation) or is absent.

is_decimal_literal 🔒

is_plain_ident 🔒

A plain SQL identifier: non-empty, only ASCII alphanumerics and underscores. Used to gate charset/collation names before interpolating them (they can’t be parameters).

lock_and_prepare_snapshot 🔒

Leader-only snapshot setup: sample PK bounds, lock the tables READ, and read the snapshot GTID frontier. All fallible work happens here so the caller can always broadcast a result. A dropped snapshot_cap_set with no broadcast deadlocks the other workers waiting on the feedback loop.

lock_tables_and_read_gtid_set 🔒

plan_worker_reads 🔒

Returns the set of full tables/sections of tables to read.

publish_snapshot_size 🔒

Publish the snapshot size to each table’s statistics gauges, using the counts computed once during PK sampling. Called leader-only so the summed worker-local gauges reflect the upstream total without double-counting.

render 🔒

Renders the snapshot dataflow. See the module documentation for more information.

sample_pk_bounds 🔒

For every table, read the row count (exact only for small tables) and, for a supported single-column primary key, compute the PK-range split boundaries, concurrently over at most worker_count connections. None bounds means single-worker fallback for that table. The counts are reused for both the sampling stride and the snapshot size gauge. Snapshot size gauge is a metric for the snapshot size used to report how many rows we need to process. “Sampling stride” refers to the number of rows we use to page through the table to find roughly evenly spaced primary keys to use as partition boundaries.

set_wait_timeout 🔒

Sets the session wait_timeout so a lowered global value cannot reap the connection while it sits idle during snapshot setup.

try_extract_single_column_pk 🔒

The raw (unquoted) name and scalar type of a table’s primary key, when it is a single column. Callers quote the name for SQL predicates and use the raw name for information_schema lookups.

verify_output_schemas 🔒

verify_pk_bounds_monotonic 🔒

worker_pk_range 🔒

This worker’s PK range, or None if it owns no partition. Rotating the partition by the table’s owner keeps the open-ended ranges from always landing on the same workers.