diff --git a/sql/handler.cc b/sql/handler.cc index 957f71e59b8..323b023299c 100644 --- a/sql/handler.cc +++ b/sql/handler.cc @@ -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); } diff --git a/sql/handler.h b/sql/handler.h index 4c52682e92a..d67e23476e8 100644 --- a/sql/handler.h +++ b/sql/handler.h @@ -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 &ranges) __attribute__ ((warn_unused_result)) { diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc index 6809182be35..1bc9ce5e6d9 100644 --- a/storage/innobase/handler/ha_innodb.cc +++ b/storage/innobase/handler/ha_innodb.cc @@ -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 &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( 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( - 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(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; } diff --git a/storage/innobase/handler/ha_innodb.h b/storage/innobase/handler/ha_innodb.h index a074d3f99ba..c5288bf41cb 100644 --- a/storage/innobase/handler/ha_innodb.h +++ b/storage/innobase/handler/ha_innodb.h @@ -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 &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; }; diff --git a/storage/innobase/include/row0pcoord.h b/storage/innobase/include/row0pcoord.h index 9a58241dfe4..2c934a75415 100644 --- a/storage/innobase/include/row0pcoord.h +++ b/storage/innobase/include/row0pcoord.h @@ -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::max() if the index does not belong to a partition. */ diff --git a/storage/innobase/row/row0pcoord.cc b/storage/innobase/row/row0pcoord.cc index 32058d255b6..d77cb35f626 100644 --- a/storage/innobase/row/row0pcoord.cc +++ b/storage/innobase/row/row0pcoord.cc @@ -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); diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc index dcaee860e69..0e3ba6947cd 100644 --- a/storage/innobase/row/row0sel.cc +++ b/storage/innobase/row/row0sel.cc @@ -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.