Split Discovery and Why a Parallel Read Is Not a Consistent Read

A single reader on a large table is bounded by one connection’s throughput. Adding readers multiplies throughput only if the table can be divided into ranges that can be read independently, and the division has to be computed before any reading starts. That computation, split discovery, is where parallel extraction either works or quietly produces a wrong result.

The difficulty is that a split boundary is a claim about the data. “Rows with id in [1000, 2000)” is only a well-defined unit of work if the key space is dense enough that each range has work, and stable enough that a row does not move between ranges during the read. Neither property is guaranteed.

The shapes a split can take

Splits are defined by predicates, and the predicate’s form determines what can go wrong.

Key range. The table is divided by a numeric or lexicographic key: id >= 0 AND id < 1000, then id >= 1000 AND id < 2000. This is the common case and the one that behaves best, because a range on an indexed column is a range scan whose cost is proportional to the rows returned.

Hash partition. The key is hashed into $N$ buckets and each reader takes one bucket: MOD(id, 8) = 3. This distributes evenly regardless of key distribution, at the cost of a predicate the source cannot use an index for unless the hash is a stored column. Each reader therefore scans the whole table and discards most of it, which multiplies the source’s I/O by $N$.

Time range. The table is divided by a date column, and this is the only form that maps naturally onto a source that is already partitioned by time. Its problem is skew: recent partitions are large and old ones are empty, so equal-width time ranges produce wildly unequal work.

Block or page ranges. The connector reads the source’s physical layout and assigns page ranges to readers. This gives perfect balance and is entirely dependent on the source’s internal format, so it is only available to a connector written for one engine and one version.

The choice determines both the parallelism achieved and the failure mode. Key ranges fail on skew. Hash partitions fail on cost. Time ranges fail on both.

Estimating the boundaries

A key-range split needs to know where to place the boundaries so that each range has roughly equal rows. Two approaches are available.

Sampling. Read a sample of the key column, taking the first and last values plus a sample of rows, and choose boundaries at the sample’s quantiles. The accuracy depends on the sample size and on the key’s distribution; a table whose keys are dense in one region and sparse in another gives boundaries that are correct for the sample and wrong for the table. The sample is also a second query against the source, which costs I/O and, under read-committed, sees a different snapshot than the subsequent reads.

Catalog statistics. Use the source’s own histogram, if it maintains one. A query planner’s statistics include a most-common-values list and a histogram of the rest, from which quantiles can be approximated. The advantage is that the statistics were gathered by the source for its own purposes and cost nothing extra to read. The disadvantage is that they are stale by design: they reflect the table as of the last statistics collection, and a table that grew by a factor since then has boundaries that no longer split evenly.

Boundary selection from a sample is a few lines, and the open-ended final range is the part that prevents omissions:

def boundaries(sample_keys, splits):
    keys = sorted(sample_keys)
    n = len(keys)
    bounds = [keys[i * n // splits] for i in range(1, splits)]
    return bounds + [None]        # last range has no upper bound

# 4 splits over keys sampled from the source: [k1, k2, k3, None]

The honest framing is that split discovery produces a good guess, and the connector should treat an empty or oversized split as normal rather than as an error. A split with zero rows costs one query and finishes immediately; a split with ten times the median costs ten times as long, and the overall read finishes when the slowest split does. With $N$ splits and a skewed distribution, the wall-clock time is set by the largest split, not the average, which is why a modest skew can erase most of the parallelism’s benefit.

The consistency problem the split introduces

Reading splits in parallel means reading at different times. Under read-committed, each split’s query sees a different snapshot, so the union of the splits is not a snapshot of the table at any instant. A row inserted into split 2 after split 2 was read but before split 7 was read is absent from the result entirely; a row moved from split 3 to split 5 by an update is present in neither if the update happened after split 3’s read and before split 5’s.

The first case is the dangerous one: a row that exists in the source at both the start and the end of the read, and is absent from the extraction. No error surfaces, and the reconciliation that compares counts will find a difference, but only after the load is complete.

The mechanisms that prevent it are the same ones the snapshot-to-stream handoff uses. Either every split reads the same exported snapshot, which requires each reader to attach to the snapshot and hold it open for the duration, or the splits are read independently and the stream supplies the changes that occurred during the read. The second is what makes a parallel initial load compatible with a live feed: the load does not need to be consistent, because the feed repairs it.

Repair by feed requires the feed’s coverage to start before the load, and requires the sink to accept changes for rows it does not yet have. The load’s split boundaries then become a performance decision rather than a correctness one, which is the right division.

The boundary predicate and the row that straddles it

A key-range split’s boundary is defined by a comparison. If the boundary is placed at id = 1000, one split reads id < 1000 and the next reads id >= 1000. The two predicates partition the key space exactly, with no overlap and no gap, as long as the comparison is total and the key is unique.

Neither holds automatically. A non-unique key, say a range on customer_id when a customer has many rows, means a boundary between two customers splits cleanly, but a boundary in the middle of a customer’s rows requires a tiebreak on a second column, exactly as the composite cursor does. A connector that splits on a non-unique key without a tiebreak reads the boundary customer’s rows in both splits or in neither, depending on which comparison is used.

Floating-point keys are worse. A split boundary expressed as a float literal may not be exactly representable, so id < 1000.1 and id >= 1000.1 partition correctly in the connector’s arithmetic and differently in the source’s, and rows near the boundary appear twice or not at all. The remedy is to split only on integer or string keys, and to derive boundaries from values read from the source rather than from computed literals.

String keys introduce collation. A boundary on a text column compares under the source’s collation, which may order uppercase before lowercase, ignore accents, or treat certain sequences as equal. A connector that computes boundaries by sorting strings in its own runtime, with a different collation, produces boundaries that do not partition the source’s ordering. The result is the same straddle: rows on one side of the connector’s boundary fall on the other side of the source’s.

Parallel readers and the source’s resources

Each reader is a connection with its own memory, its own transaction, and its own share of the source’s I/O. A source sized for a handful of application connections may have a connection limit far below the reader count a parallel load wants, and exceeding it fails connections rather than degrading gracefully.

Beyond connections, the parallel scan’s aggregate I/O is the sum of the splits’ I/O, which for a full table read is the table’s size regardless of parallelism. Parallelism reduces wall-clock time by using more of the device’s queue depth, and it does not reduce the total bytes read. On a source whose storage is already saturated, adding readers makes every reader slower and the aggregate no faster.

The buffer cache is the shared resource that parallel reads damage most. A full scan of a table larger than memory evicts the pages the live workload depends on, so the live workload’s latency rises during the load. This is the same effect described in backfilling beside a live stream, and it applies whether the read is a backfill or an initial load.

Skew that appears after discovery

The most interesting failure is a split plan computed against one distribution and executed against another. A table that receives inserts concentrated at high keys, the common pattern for an append-mostly table with an increasing primary key, has a distribution that moves continuously. Boundaries computed from yesterday’s statistics put most of today’s rows in the last split, and the load finishes when that split finishes.

The mitigation is to compute boundaries from the current maximum rather than from a histogram, so the last split covers the recent range and the earlier splits cover the bulk. A plan of id < 1000, 1000 <= id < 2000, …, with the final split open-ended above the maximum observed at plan time, gives an unbalanced plan for a table with a growing key but ensures that no row is outside every split.

The open-ended final split is also the safe default for a table whose key space is unknown, since a plan with a closed upper bound silently omits every row above it. That omission is the parallel-load analogue of the tie-group loss in a single-column cursor: a boundary chosen for convenience excludes rows, and nothing reports it.

What the splits cannot divide

A table with no suitable key cannot be split by predicate at all. A table whose key is a random UUID has no locality, so range splits are balanced by construction only if the UUIDs are uniformly distributed, and hash splits are balanced but require a full scan per reader. A table with no key beyond a surrogate has splits that are meaningless to a consumer, so the parallel read is a performance device with no semantic content.

For those tables, the split is the physical page range, and the connector needs engine-specific knowledge to compute it. Where that is unavailable, the load is single-threaded, and the honest design accepts it rather than inventing a key to split on, because an invented key is one more boundary whose partition property is assumed rather than established.