COP 606: Coprocessor Patterns
COP 503 introduced the high-level coprocessor path: TiDB divides a query into Region-scoped tasks, TiKV reads each Region from a snapshot, and TiDB combines the partial results.
This chapter looks at what those tasks actually read.
SQL talks about tables and indexes, while TiKV stores ordered key-value pairs. TiDB bridges the two by encoding rows and index entries as keys.
From Tables to TiKV Keys
Consider a small orders table with a non-unique index on status and a unique index on order_number:
orders
id user_id status amount order_number
1001 201 paid 80 A-1001
1002 202 pending 35 A-1002
1003 201 paid 52 A-1003
1004 202 paid 70 A-1004
Assume that TiDB assigned this table the ID 42, the status index the ID 3, and the order_number index the ID 4.
Each stored row needs a row handle, the value TiKV uses to identify that row.
With a clustered primary key, the SQL primary key itself is the row handle. With a nonclustered primary key, TiDB assigns the row a hidden internal ID, _tidb_rowid, as its handle.
For this example, assume that id is the row handle. TiDB calls this a clustered primary key, and the row keys use id directly:
TiKV does not store a table object. It stores entries shaped roughly like these:
row data
t42_r1001 -> { user_id: 201, status: "paid", amount: 80, order_number: "A-1001" }
t42_r1002 -> { user_id: 202, status: "pending", amount: 35, order_number: "A-1002" }
t42_r1003 -> { user_id: 201, status: "paid", amount: 52, order_number: "A-1003" }
t42_r1004 -> { user_id: 202, status: "paid", amount: 70, order_number: "A-1004" }
status index (non-unique; value omitted)
t42_i3_paid_1001
t42_i3_paid_1003
t42_i3_paid_1004
t42_i3_pending_1002
These are simplified readable forms, not the literal bytes stored by TiKV. The important pieces are:
t42identifies table 42.rmarks a row key; the following value is the row's handle,idin this clustered example.i3marks index 3; it is followed by the indexed value and, for a non-unique index, the row handle.
The row handle distinguishes rows with the same status. The index values are omitted above because the key already identifies both the indexed value and the row. Their exact layout is not needed for this access pattern.
A unique index does not need to put the handle in its key. It stores the handle in the index value instead:
t42_i4_A-1001 -> 1001
With a nonclustered primary key, the SQL primary key still enforces uniqueness, but it is stored as a unique index that translates id to the hidden handle:
row data
t42_r5001 -> { id: 1001, user_id: 201, status: "paid", amount: 80, order_number: "A-1001" }
nonclustered primary-key index
t42_i1_1001 -> 5001
non-unique status index (value omitted)
t42_i3_paid_5001
The exact byte encoding has more details, but this shape explains the access pattern. All row keys for orders occupy one interval of TiKV's key space. The entries for one index occupy another interval. TiDB turns a table or index access into one or more ranges within those intervals, then splits those ranges at Region boundaries.
Table Scans and Index Scans
Consider a query that reads a continuous range of row IDs:
SELECT *
FROM orders
WHERE id >= 1000 AND id < 1100;
The relevant row keys form one continuous interval:
[t42_r1000, t42_r1100)
Each cop task seeks to the beginning of its Region's portion, then moves forward with next() through the row keys. This is a table range scan.
Now consider a query whose requested columns are available from the status index:
SELECT id, status
FROM orders
WHERE status = 'paid';
If TiDB chooses that index, TiKV seeks to the paid part of the index interval and scans its matching entries:
t42_i3_paid_1001
t42_i3_paid_1003
t42_i3_paid_1004
The index contains status and the row handle, so this query can be answered directly from the index. This is an index scan.
Index Lookup
The same condition can need more work when the query asks for columns not stored in the index:
SELECT *
FROM orders
WHERE status = 'paid';
TiKV first scans the paid index entries to find row handles. It then reads the full rows using those handles:
index scan
t42_i3_paid_1001 -> row handle 1001
t42_i3_paid_1003 -> row handle 1003
t42_i3_paid_1004 -> row handle 1004
|
v
row reads
seek t42_r1001
seek t42_r1003
seek t42_r1004
This is an index lookup. The index portion is still a contiguous scan, but the returned row handles may be far apart. Fetching full rows can therefore require many separate seeks. Even when the index scan itself is efficient, the back-to-table row reads can make the query expensive.
The optimizer chooses between these patterns from the query, available indexes, and its estimate of how much data each option will read.
Seek and Scan
Once TiDB has chosen an access path, TiKV sees two basic access shapes:
continuous key interval -> seek once, then scan with next()
many unrelated row keys -> seek repeatedly
Joining Two Tables
So far, each query has read one table. A join reads two tables and finds rows that match a condition between them. Consider an orders table and a users table:
users
id name
201 Alice
202 Bob
203 Carol
204 David
SELECT orders.id, users.name
FROM orders
JOIN users ON orders.user_id = users.id
WHERE orders.status = 'paid';
With an index join, the plan calls the first input the outer table and the table looked up for each outer row the inner table. In this example, orders is the outer table and users is the inner table.
TiDB first reads the paid orders. It scans the status index for their row handles, then reads the matching order rows to obtain user_id. Each resulting order supplies a user_id. TiDB then uses that ID to look up the matching user:
outer: orders
paid order 1001 -> user_id 201
paid order 1003 -> user_id 201
paid order 1004 -> user_id 202
|
v
inner: users
seek the row or unique index entry for user 201
seek the row or unique index entry for user 201
seek the row or unique index entry for user 202
TiKV therefore sees an outer scan followed by inner row or index lookups. The inner keys may be unrelated across outer rows, so this pattern can be seek-heavy even when the outer scan is efficient.
With a hash join, TiDB chooses one input as the build side, scans it, and builds an in-memory hash table keyed by the join column. It then scans the other input, the probe side, and looks for each join key in that hash table. TiDB usually chooses the input expected to be smaller as the build side, so the hash table uses less memory.
For the same query, TiDB might use the paid orders as the build side. TiKV reads the paid orders, and TiDB builds this in-memory structure:
user_id -> order IDs
201 -> [1001, 1003]
202 -> [1004]
TiKV then scans the users input. When TiDB receives user_id = 201, it probes the hash table, finds orders 1001 and 1003, and produces the joined rows with Alice. For user_id = 202, it finds order 1004 and produces the row with Bob.
Hash-table matching is done in TiDB. TiKV performs the table or index scans chosen by TiDB and can apply pushed-down filters. The users input may be a full table scan here because the query has no condition that narrows it, but it could use an index or range scan for a different query.
A hash join avoids the index join's lookup into users for every paid order. Instead, it reads each join input once and matches rows in memory. Each input may use a table scan, index scan, or index lookup.
The optimizer chooses the join strategy from available indexes and its estimate of the input sizes. In either case, the join eventually becomes table ranges, index ranges, or individual row accesses in TiKV.
How Operators Pull Data
A scan is often only the first step of the pushed-down work. TiDB describes the work as a directed acyclic graph, or DAG, of operators. The graph shows which operator supplies data to another.
The common operators have simple jobs:
- A scan reads row or index keys from the RocksDB snapshot.
- A filter keeps only rows that satisfy a condition.
- An aggregate combines many rows into a value, such as
COUNTorSUM. WithGROUP BY, it produces one value for each group. - A TopN operator keeps the first N results under an ordering.
Consider a query that finds the three users with the largest total value of paid orders:
SELECT user_id, SUM(amount) AS total
FROM orders
WHERE status = 'paid'
GROUP BY user_id
ORDER BY total DESC
LIMIT 3;
Each TiKV task can scan its Region's rows, filter the paid orders, and calculate a partial sum for each user_id. TiDB receives those partial sums, combines groups that appeared in different Regions, and selects the final three results:
TiDB
TopN: largest 3 totals
|
v
final aggregate by user_id
|
partial sums from TiKV tasks
/ | \
v v v
TiKV, one task per Region
partial aggregate: SUM(amount)
|
v
filter: status = "paid"
|
v
table scan
For example, one task might produce user_id 201 -> 80, while another produces user_id 201 -> 52. TiDB must combine them into user_id 201 -> 132 before it can choose the global TopN results.
Inside a TiKV task, operators pull data from bottom to top. When the partial aggregate needs another row, it asks the filter. The filter asks the scan for rows until it finds one that passes status = 'paid'. The scan advances through the RocksDB snapshot. Rows are consumed incrementally rather than passed upward as one complete raw result set, although an aggregate keeps state such as user_id -> partial sum while it runs.
The scan reads through the snapshot established by Raftstore, so every operator in the task uses one consistent view of that Region.
COP 707: Coprocessor Execution examines when these tasks yield and how TiKV reports their waiting and execution.