database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2026-10-23 18:49:29 +03:00
Code Activity

MDEV-40005 Parallel Query: allow parallel scans on secondary indexes in InnoDB

Browse Source 
This commit extends parallel scanning to InnoDB secondary indexes, both
    whole-index scans and scans of ranges of the key. It complements the
    previously introduced partitioning of the clustered index.

    Both unique and non-unique secondary indexes are supported, as well as
    covering and non-covering scans.
This commit is contained in:
Oleg Smirnov
2026-08-27 02:12:54 +07:00
committed by Rex Johnston
parent 7f81dc8119
commit 66eefaeb95
7 changed files with 254 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
sql/handler.cc
View File

@@ -4103,25 +4103,46 @@ int handler::ha_parallel_get_next_row(Parallel_worker_ctx *ctx)
int result;
DBUG_ENTER("handler::ha_parallel_get_next_row");
TABLE_IO_WAIT(tracker, PSI_TABLE_FETCH_ROW, MAX_KEY, result,
/*
A parallel scan reads either the clustered index (inited == RND, like a
table scan) or one named index (inited == INDEX). All the bookkeeping below
has to match whichever serial reader this plan replaced - ha_rnd_next() or
ha_index_next() - otherwise Handler_read_*, the index statistics and the
sync points move depending on whether the optimizer parallelized.
*/
const bool index_scan= inited == INDEX;
TABLE_IO_WAIT(tracker, PSI_TABLE_FETCH_ROW,
index_scan ? active_index : MAX_KEY, result,
{ result = parallel_get_next_row(ctx); });
/*
increment_statistics() also runs check_limit_rows_examined(),
which is what enforces LIMIT ROWS EXAMINED.
*/
increment_statistics(index_scan ? &SSV::ha_read_next_count
: &SSV::ha_read_rnd_next_count);
if (result == 0)
{
update_rows_read();
if (index_scan)
update_index_statistics();
else
update_rows_read();
if (table->vfield)
table->update_virtual_fields(this, VCOL_UPDATE_FOR_READ);
table->status= 0;
}
else if (result != HA_ERR_END_OF_FILE)
{
table->status= STATUS_NOT_FOUND;
}
else
{
table->status= STATUS_NOT_FOUND; // EOF also marks not-found
}
status_var_increment(table->in_use->status_var.ha_read_rnd_next_count);
/* Same sync point the serial reader would have fired */
if (index_scan)
DEBUG_SYNC(ha_thd(), "handler_ha_index_next_end");
else
DEBUG_SYNC(ha_thd(), "handler_rnd_next_end");
DBUG_RETURN(result);
}

3
sql/handler.h
View File

@@ -3788,13 +3788,14 @@ public:
To be called from the master thread to initiate the parallel scanning.
@param n_workers number of workers to partition the scan for
@param keynr MAX_KEY for full table scan / index number for index scan
@param ranges key intervals to scan, in key order, in the same
representation the MRR interface uses: an endpoint whose
keypart_map is 0 is unbounded, and start_key.flag /
end_key.flag say whether the endpoint itself belongs to
the interval. Empty means scan the whole table.
*/
virtual int parallel_init_coordinator(size_t n_workers,
virtual int parallel_init_coordinator(size_t n_workers, uint keynr,
const Dynamic_array<KEY_MULTI_RANGE> &ranges)
__attribute__ ((warn_unused_result))
{

169
storage/innobase/handler/ha_innodb.cc
View File

@@ -14742,7 +14742,7 @@ func_exit:
}
int ha_innobase::parallel_init_coordinator(size_t n_threads,
int ha_innobase::parallel_init_coordinator(size_t n_threads, uint keynr,
const Dynamic_array<KEY_MULTI_RANGE> &ranges)
{
/* Reset any state left by a prior execution (correlateds subquery
@@ -14763,15 +14763,10 @@ int ha_innobase::parallel_init_coordinator(size_t n_threads,
return HA_ERR_UNSUPPORTED;
}
/* The partitioner and the end-of-chunk check (cmp_dtuple_rec(...) <= 0)
assume ascending physical key order. A descending key column in the
clustered index reverses B+Tree traversal, breaking range boundaries
and dropping rows. Decline; fall back to the serial scan. */
const dict_index_t *clust = m_prebuilt->table->indexes.start;
for (unsigned i = dict_index_get_n_unique_in_tree(clust); i--; ) {
if (clust->fields[i].descending) {
return HA_ERR_UNSUPPORTED;
}
dict_index_t *index = pscan_resolve_index(keynr);
if (index == NULL) {
return HA_ERR_UNSUPPORTED;
}
/* Refuse the scan if the tablespace is discarded or unreadable. */
@@ -14802,11 +14797,8 @@ int ha_innobase::parallel_init_coordinator(size_t n_threads,
m_parallel_coordinator.initialize(n_threads);
// Get the clustered index which is always first in the list
dict_index_t *index = m_prebuilt->table->indexes.start;
ut_ad(index->is_clust());
dberr_t err = DB_SUCCESS;
m_pscan_keynr = keynr;
if (ranges.size() == 0) {
/* Already cleared by the parallel_end_coordinator() above;
@@ -14864,9 +14856,49 @@ int ha_innobase::parallel_init_coordinator(size_t n_threads,
return convert_error_code_to_mysql(err, m_prebuilt->table->flags,
m_user_thd);
}
return 0;
}
dict_index_t* ha_innobase::pscan_resolve_index(uint keynr)
{
/* keynr indexes table->key_info[].
MAX_KEY asks for the clustered index, generated or not. */
dict_index_t *index = innobase_get_index(keynr);
if (index == NULL)
return NULL;
if (index->is_corrupted()
|| !row_merge_is_index_usable(m_prebuilt->trx, index)) {
/* Being built by online DDL, or flagged corrupt. Unlike the
clustered index of an open table, a secondary index can be in
this state on its own. */
return NULL;
}
/* Index kinds the partitioner cannot reason about:
- spatial: non-leaf records hold MBRs;
- FTS: not a regular B+Tree of table rows;
- virtual columns: records hold computed values which have to be
validated against the clustered row before use. */
if (index->is_spatial() || (index->type & DICT_FTS)
|| index->has_virtual()) {
return NULL;
}
/* The partitioner and the end-of-chunk check (cmp_dtuple_rec(...) <= 0)
assume ascending physical key order. A descending key column reverses
B+Tree traversal, breaking range boundaries, so decline. */
for (unsigned i = dict_index_get_n_unique_in_tree(index); i--; ) {
if (index->fields[i].descending) {
return NULL;
}
}
return index;
}
const key_range* ha_innobase::pscan_get_start_key(size_t scan_id) const
{
/* Scan ids and m_pscan_ranges positions are the same sequence:
@@ -14894,14 +14926,12 @@ void ha_innobase::pscan_begin_chunk(Parallel_coordinator::Worker_ctx *wctx)
if (start_key) {
/* pscan_before_range_start() compares table->record[0] against
the key through range_key_part. A key range is only ever handed
to us for a scan of the primary key, so active_index is a real
key number here - unlike a full scan of a table whose clustered
index was auto-generated, where ha_rnd_init() leaves it
MAX_KEY. */
ut_ad(active_index == table->s->primary_key);
ut_ad(active_index < MAX_KEY);
range_key_part = table->key_info[active_index].key_part;
the key through range_key_part. Intervals are only ever handed
to us naming a real key, so m_pscan_keynr is not MAX_KEY here -
unlike a full scan of a table whose clustered index was
auto-generated, where there is no MySQL key to point at. */
ut_ad(m_pscan_keynr < MAX_KEY);
range_key_part = table->key_info[m_pscan_keynr].key_part;
}
}
@@ -14928,6 +14958,53 @@ bool ha_innobase::pscan_before_range_start(
return false;
}
const dtuple_t *ha_innobase::pscan_exclusive_start(
const Parallel_coordinator::Exec_ctx &exec_ctx) const
{
/* A chunk is opened at its own first record, and that record belongs to
the chunk, so the open has to be PAGE_CUR_GE. That cannot express an
exclusive lower bound: for "a > 5" the partitioner's first chunk begins
at the first record with a = 5, and pscan_before_range_start() then
throws away every record sharing that value - the bound's whole
multiplicity, which on a low-cardinality column is most of a chunk.
A serial scan never reads them: convert_search_mode_to_innobase() turns
HA_READ_AFTER_KEY into PAGE_CUR_G and the descent lands past them.
So do the same here, for as long as the chunk still holds the bound. */
const key_range *start_key = pscan_get_start_key(exec_ctx.scan_id());
if (start_key == nullptr || start_key->flag != HA_READ_AFTER_KEY) {
return nullptr;
}
// Start bound as requested by the caller, not the actual chunk start:
const dtuple_t *start_bound = exec_ctx.scan_start();
if (!start_bound)
return nullptr;
const dtuple_t *chunk_start = exec_ctx.m_range.first->m_tuple;
const ulint n_cmp = dtuple_get_n_fields_cmp(start_bound);
ut_ad(n_cmp > 0);
ut_ad(n_cmp <= dtuple_get_n_fields(chunk_start));
for (ulint i = 0; i < n_cmp; i++) {
const int cmp = cmp_dfield_dfield(
dtuple_get_nth_field(start_bound, i),
dtuple_get_nth_field(chunk_start, i));
if (cmp != 0) {
/* cmp < 0: the chunk begins above the bound already. */
return cmp < 0 ? nullptr : start_bound;
}
}
/* Equal on every named field: this chunk begins inside the start bound's
key value, so every record from here up to the first one above the bound
is out of range and has to be skipped. Opening on the bound is what
skips them. */
return start_bound;
}
dtuple_t* ha_innobase::pscan_convert_key(const key_range *kr,
const dict_index_t *index,
mem_heap_t *heap)
@@ -14943,9 +15020,9 @@ dtuple_t* ha_innobase::pscan_convert_key(const key_range *kr,
byte* buf = static_cast<byte*>(
mem_heap_alloc(heap, m_prebuilt->srch_key_val_len));
ut_ad(table->s->primary_key < MAX_KEY);
ut_ad(m_pscan_keynr < MAX_KEY);
const uint n_key_fields =
table->key_info[table->s->primary_key].ext_key_parts;
table->key_info[m_pscan_keynr].ext_key_parts;
dtuple_t* tuple = dtuple_create(heap, n_key_fields);
dict_index_copy_types(tuple, index, n_key_fields);
@@ -14984,7 +15061,12 @@ int ha_innobase::parallel_init_worker(Parallel_worker_ctx *wctx)
the coordinator releases the chunks. */
m_pscan_saved_search_tuple= m_prebuilt->search_tuple;
if (int err= ha_rnd_init(/*scan*/ true))
/* MAX_KEY means the clustered index, which may be auto-generated and so
have no table->key_info[] entry for ha_index_init() to name. Any other
value names a real key, clustered or not. */
if (int err= m_pscan_keynr == MAX_KEY
? ha_rnd_init(/*scan*/ true)
: ha_index_init(m_pscan_keynr, /*sorted*/ false))
return err; // preserve HA_ERR_* (e.g. HA_ERR_TABLE_DEF_CHANGED)
worker_ctx->m_exec_ctx= exec_ctx;
@@ -15015,19 +15097,28 @@ int ha_innobase::parallel_get_next_row(Parallel_worker_ctx *wctx)
chunk.m_range.second->m_tuple,
chunk.m_end_inclusive);
if (chunk.m_range.first->m_tuple) {
// Position at first record >= start, AND load it.
m_prebuilt->search_tuple = const_cast<dtuple_t *>(
chunk.m_range.first->m_tuple);
err= row_search_mvcc(table->record[0], PAGE_CUR_GE,
m_prebuilt, 0, 0 /*opening*/);
} else {
/* -infinity: empty (0-field) tuple + PAGE_CUR_G =
first user record. */
m_prebuilt->search_tuple = dtuple_create(m_prebuilt->heap, 0);
err = row_search_mvcc(table->record[0], PAGE_CUR_G,
m_prebuilt, 0, 0 /*opening*/);
const dtuple_t *open_tuple =
chunk.m_range.first->m_tuple;
page_cur_mode_t open_mode = PAGE_CUR_GE;
if (open_tuple == NULL) {
/* -infinity:
empty (0-field) tuple + PAGE_CUR_G = first user record. */
open_tuple = dtuple_create(m_prebuilt->heap, 0);
open_mode = PAGE_CUR_G;
} else if (const dtuple_t *start_bound =
pscan_exclusive_start(chunk)) {
/* Skip the start bound's key value in the descent rather
than reading it row by row. */
open_tuple = start_bound;
open_mode = PAGE_CUR_G;
}
// Position at the first record to read, AND load it.
m_prebuilt->search_tuple =
const_cast<dtuple_t *>(open_tuple);
err = row_search_mvcc(table->record[0], open_mode,
m_prebuilt, 0, 0 /*opening*/);
}
else {
// Continuation: advance from stored position.
@@ -15070,8 +15161,7 @@ int ha_innobase::parallel_get_next_row(Parallel_worker_ctx *wctx)
int ha_innobase::parallel_end_worker()
{
if (inited == RND)
ha_rnd_end();
ha_index_or_rnd_end();
if (m_pscan_saved_search_tuple)
{
m_prebuilt->search_tuple = m_pscan_saved_search_tuple;
@@ -15090,6 +15180,7 @@ int ha_innobase::parallel_end_coordinator()
}
m_pscan_ranges = nullptr;
m_pscan_n_ranges = 0;
m_pscan_keynr = MAX_KEY;
return 0;
}

21
storage/innobase/handler/ha_innodb.h
View File

@@ -521,7 +521,7 @@ protected:
}
/** Implementation of handler API for parallel scans */
int parallel_init_coordinator(size_t n_threads,
int parallel_init_coordinator(size_t n_threads, uint keynr,
const Dynamic_array<KEY_MULTI_RANGE> &ranges) override;
/** Convert one MySQL-format key endpoint into an InnoDB tuple.
@@ -599,6 +599,20 @@ protected:
const KEY_MULTI_RANGE *m_pscan_ranges{};
uint m_pscan_n_ranges{};
/** table->key_info[] number of the index being scanned in parallel, as
passed to parallel_init_coordinator(). MAX_KEY means the clustered
index; the range helpers below only run when there are intervals, which
always name a real key. */
uint m_pscan_keynr{MAX_KEY};
/** Resolve the index a parallel scan was asked for and check that the
partitioner can handle it.
@param keynr table->key_info[] number, or MAX_KEY for the clustered
index
@return the index, or NULL if it does not exist or is not supported -
the caller declines the scan and the serial path runs instead */
dict_index_t* pscan_resolve_index(uint keynr);
/** Lower bound of interval 'scan_id', or NULL if it is unbounded or
this is a full table scan. The upper bound needs no handler-side state:
it is enforced by the chunk clamp inside row_search_mvcc(). */
@@ -611,6 +625,11 @@ protected:
of the interval wctx is currently reading. Clears wctx->m_check_start
once a row clears the bound. */
bool pscan_before_range_start(Parallel_coordinator::Worker_ctx *wctx);
/** The bound to open 'chunk' on so that an exclusive lower bound costs
no wasted reads, or NULL to open on the chunk's own first record. */
const dtuple_t *pscan_exclusive_start(
const Parallel_coordinator::Exec_ctx &chunk) const;
};

8
storage/innobase/include/row0pcoord.h
View File

@@ -526,6 +526,14 @@ class Parallel_coordinator::Exec_ctx {
return m_scan_ctx->m_config.m_index;
}
/** @return the scan's own lower bound, nullptr for -infinity.
This is the caller's bound, not this chunk's start: the two coincide only
for the chunk the bound falls in. */
[[nodiscard]] const dtuple_t *scan_start() const {
return m_scan_ctx->m_config.m_scan_range.m_start;
}
/** @return the partition id of the index.
@note this is std::numeric_limits<size_t>::max() if the index does not
belong to a partition. */

69
storage/innobase/row/row0pcoord.cc
View File

@@ -288,7 +288,7 @@ page_no_t Parallel_coordinator::Scan_ctx::search(buf_block_t *block,
if (key != nullptr)
{
auto ps_err = pread_page_cur_search(block, index, key,
PAGE_CUR_LE, &page_cursor);
PAGE_CUR_L, &page_cursor);
if (ps_err != DB_SUCCESS)
{
*err = ps_err;
@@ -369,6 +369,7 @@ page_cur_t Parallel_coordinator::Scan_ctx::start_range(
}
page_cur_t page_cursor;
page_cursor.index = index;
if (key != nullptr) {
auto ps_err = pread_page_cur_search(block, index, key, PAGE_CUR_GE, &page_cursor);
@@ -393,6 +394,47 @@ page_cur_t Parallel_coordinator::Scan_ctx::start_range(
}
}
/* search() descends with the strict PAGE_CUR_L, which can stop one
sub-tree short of `key`: it lands on the last leaf whose lowest key is
below `key`, and every record on that leaf may still be below `key`. The
record we want is then the first one on the following page - the leaf after
it starts at or above `key` by construction. */
while (page_cur_is_after_last(&page_cursor))
{
const auto next_no = btr_page_get_next(buf_block_get_frame(block));
/* End of the level: nothing at or after `key`. The caller sees the
supremum and creates no range. */
if (next_no == FIL_NULL)
break;
/* Drop the page we are leaving: the record we want is further right */
ut_ad(!savepoints.empty() && savepoints.back().second == block);
mtr->release(*block);
savepoints.pop_back();
savepoint = mtr->get_savepoint();
page_id.set_page_no(next_no);
block = block_get_s_latched(page_id, mtr, __LINE__);
if (!block)
{
*err = DB_CORRUPTION;
return page_cur_t{};
}
savepoints.push_back({savepoint, block});
page_cur_set_before_first(block, &page_cursor);
if (!page_cur_move_to_next(&page_cursor))
{
*err = DB_CORRUPTION;
return page_cur_t{};
}
}
return (page_cursor);
}
}
@@ -462,21 +504,13 @@ Parallel_coordinator::Scan_ctx::create_ranges(const Scan_range &scan_range,
if (start != nullptr)
{
/* A node pointer key is the *lowest* key of its child sub-tree, so the
child that may hold `start` is the last node pointer <= start, not the
first one >= start. PAGE_CUR_GE would descend into the child after it
and lose every row from `start` up to that child's first key. The
leftmost node pointer additionally carries REC_INFO_MIN_REC_FLAG and
compares less than any tuple (see cmp_dtuple_rec_with_match_low()), so
GE always skips the leftmost sub-tree. Use the same mode Scan_ctx::
search() uses to descend, and keep PAGE_CUR_GE once we are on a leaf,
where we do want the first record >= start.
Only Exec_ctx::split() reaches this with a non-NULL start - it
re-partitions an existing chunk beginning at that chunk's first key -
which is why this only shows up on trees deep enough to be re-split. */
/* On a leaf, `start` is where the scan begins, so we want the first record
>= start: PAGE_CUR_GE. One level up the same bound asks something else - a
node pointer key is the *lowest* key of its sub-tree, so we need the child
at or before `start`, and PAGE_CUR_L rather than LE because `start` may be
a prefix (see Scan_ctx::search()). */
const page_cur_mode_t mode= page_is_leaf(buf_block_get_frame(block))
? PAGE_CUR_GE : PAGE_CUR_LE;
? PAGE_CUR_GE : PAGE_CUR_L;
auto err = pread_page_cur_search(block, index, start,
mode, &page_cursor);
@@ -485,6 +519,9 @@ Parallel_coordinator::Scan_ctx::create_ranges(const Scan_range &scan_range,
if (page_cur_is_after_last(&page_cursor))
{
/* Only the leaf mode (GE) can land here; L stops at the infimum at
worst. `start` is past every record on this leaf, so this sub-tree
contributes no range */
return (DB_SUCCESS);
}
else if (page_cur_is_before_first((&page_cursor)))
@@ -602,7 +639,7 @@ Parallel_coordinator::Scan_ctx::create_ranges(const Scan_range &scan_range,
create_range(ranges, level_page_cursor);
}
/* We've created the persistent cursor, safe to release S latches on
/* We've created the range, safe to release S latches on
the blocks that are in this range (sub-tree). */
for (auto &savepoint : savepoints)
mtr->release(*savepoint.second);

30
storage/innobase/row/row0sel.cc
View File

@@ -4406,7 +4406,7 @@ struct InnoDBPolicy
}
};
/** Parallel scan: check whether a qualifying record has reached the
/** Parallel scan: check whether the record the cursor is on has reached the
exclusive upper bound of the chunk that is being scanned.
The straightforward check is one cmp_dtuple_rec() per fetched record, which
@@ -4425,7 +4425,9 @@ not bump modify_clock), and an old version built in the heap is not covered
by the verdict of any page.
@param prebuilt row fetch struct with a non-NULL m_pscan_end_tuple
@param rec the record that is about to be returned to the SQL layer
@param rec the record the cursor is positioned on. Must belong to the
index being scanned - m_pscan_end_tuple is a key of that
index, and cmp_dtuple_rec() pairs fields by ordinal.
@param index the index that rec belongs to
@param offsets rec_get_offsets(rec, index)
@return whether rec is past the upper bound of the chunk. The bound itself
@@ -5272,6 +5274,15 @@ wrong_offs:
}
}
/* Parallel scan: stop at the upper bound of this chunk. Checked here,
on the record the cursor is on, because the boundary is a position in
`index`. */
if (UNIV_UNLIKELY(prebuilt->m_pscan_end_tuple != NULL)
&& row_pscan_reached_chunk_end(prebuilt, rec, index, offsets)) {
err = DB_RECORD_NOT_FOUND;
goto idx_cond_failed;
}
/* Note that we cannot trust the up_match value in the cursor at this
place because we can arrive here after moving the cursor! Thus
we have to recompare rec and search_tuple to determine if they
@@ -5852,21 +5863,6 @@ use_covering_index:
offsets));
ut_ad(!rec_get_deleted_flag(result_rec, comp));
/* Parallel scan: stop at the exclusive upper bound of the chunk.
This runs before the record is returned or prefetched, so the
fetch cache never reaches past the chunk boundary. We treat the
boundary exactly like end-of-range (DB_RECORD_NOT_FOUND): any
rows already buffered in this call are still in range and will be
returned; the caller switches to the next chunk on the terminal
code. */
if (UNIV_UNLIKELY(prebuilt->m_pscan_end_tuple != NULL)
&& row_pscan_reached_chunk_end(
prebuilt, result_rec,
result_rec != rec ? clust_index : index, offsets)) {
err = DB_RECORD_NOT_FOUND;
goto idx_cond_failed;
}
/* Decide whether to prefetch extra rows.
At this point, the clustered index record is protected
by a page latch that was acquired when pcur was positioned.