宏btr_pcur_open_on_user_rec

#define btr_pcur_open_on_user_rec(i,t,md,l,c,m)                \
    btr_pcur_open_on_user_rec_func(i,t,md,l,c,__FILE__,__LINE__,m)

/**************************************************************//**
If mode is PAGE_CUR_G or PAGE_CUR_GE, opens a persistent cursor on the first
user record satisfying the search condition, in the case PAGE_CUR_L or
PAGE_CUR_LE, on the last user record. If no such user record exists, then
in the first case sets the cursor after last in tree, and in the latter case
before first in tree. The latching mode must be BTR_SEARCH_LEAF or
BTR_MODIFY_LEAF. */
UNIV_INTERN
void
btr_pcur_open_on_user_rec_func(
/*===========================*/
    dict_index_t*    index,        /*!< in: index */
    const dtuple_t*    tuple,        /*!< in: tuple on which search done */
    ulint        mode,        /*!< in: PAGE_CUR_L, ... */
    ulint        latch_mode,    /*!< in: BTR_SEARCH_LEAF or
                    BTR_MODIFY_LEAF */
    btr_pcur_t*    cursor,        /*!< in: memory buffer for persistent
                    cursor */
    const char*    file,        /*!< in: file name */
    ulint        line,        /*!< in: line where called */
    mtr_t*        mtr)        /*!< in: mtr */
{
    btr_pcur_open_func(index, tuple, mode, latch_mode, cursor,
               file, line, mtr);

    if ((mode == PAGE_CUR_GE) || (mode == PAGE_CUR_G)) {

        if (btr_pcur_is_after_last_on_page(cursor)) {

            btr_pcur_move_to_next_user_rec(cursor, mtr);
        }
    } else {
        ut_ad((mode == PAGE_CUR_LE) || (mode == PAGE_CUR_L));

        /* Not implemented yet */

        ut_error;
    }
}

/**************************************************************//**
Initializes and opens a persistent cursor to an index tree. It should be
closed with btr_pcur_close. */
UNIV_INLINE
void
btr_pcur_open_func(
/*===============*/
    dict_index_t*    index,    /*!< in: index */
    const dtuple_t*    tuple,    /*!< in: tuple on which search done */
    ulint        mode,    /*!< in: PAGE_CUR_L, ...;
                NOTE that if the search is made using a unique
                prefix of a record, mode should be
                PAGE_CUR_LE, not PAGE_CUR_GE, as the latter
                may end up on the previous page from the
                record! */
    ulint        latch_mode,/*!< in: BTR_SEARCH_LEAF, ... */
    btr_pcur_t*    cursor, /*!< in: memory buffer for persistent cursor */
    const char*    file,    /*!< in: file name */
    ulint        line,    /*!< in: line where called */
    mtr_t*        mtr)    /*!< in: mtr */
{
    btr_cur_t*    btr_cursor;

    /* Initialize the cursor */

    btr_pcur_init(cursor);

    cursor->latch_mode = latch_mode;
    cursor->search_mode = mode;

    /* Search with the tree cursor */

    btr_cursor = btr_pcur_get_btr_cur(cursor);

    btr_cur_search_to_nth_level(index, 0, tuple, mode, latch_mode,
                    btr_cursor, 0, file, line, mtr);
    cursor->pos_state = BTR_PCUR_IS_POSITIONED;

    cursor->trx_if_known = NULL;
}

/********************************************************************//**
Searches an index tree and positions a tree cursor on a given level.
NOTE: n_fields_cmp in tuple must be set so that it cannot be compared
to node pointer page number fields on the upper levels of the tree!
Note that if mode is PAGE_CUR_LE, which is used in inserts, then
cursor->up_match and cursor->low_match both will have sensible values.
If mode is PAGE_CUR_GE, then up_match will a have a sensible value.

If mode is PAGE_CUR_LE , cursor is left at the place where an insert of the
search tuple should be performed in the B-tree. InnoDB does an insert
immediately after the cursor. Thus, the cursor may end up on a user record,
or on a page infimum record. */
UNIV_INTERN
void
btr_cur_search_to_nth_level(
/*========================*/
    dict_index_t*    index,    /*!< in: index */
    ulint        level,    /*!< in: the tree level of search */
    const dtuple_t*    tuple,    /*!< in: data tuple; NOTE: n_fields_cmp in
                tuple must be set so that it cannot get
                compared to the node ptr page number field! */
    ulint        mode,    /*!< in: PAGE_CUR_L, ...;
                Inserts should always be made using
                PAGE_CUR_LE to search the position! */
    ulint        latch_mode, /*!< in: BTR_SEARCH_LEAF, ..., ORed with
                at most one of BTR_INSERT, BTR_DELETE_MARK,
                BTR_DELETE, or BTR_ESTIMATE;
                cursor->left_block is used to store a pointer
                to the left neighbor page, in the cases
                BTR_SEARCH_PREV and BTR_MODIFY_PREV;
                NOTE that if has_search_latch
                is != 0, we maybe do not have a latch set
                on the cursor page, we assume
                the caller uses his search latch
                to protect the record! */
    btr_cur_t*    cursor, /*!< in/out: tree cursor; the cursor page is
                s- or x-latched, but see also above! */
    ulint        has_search_latch,/*!< in: info on the latch mode the
                caller currently has on btr_search_latch:
                RW_S_LATCH, or 0 */
    const char*    file,    /*!< in: file name */
    ulint        line,    /*!< in: line where called */
    mtr_t*        mtr)    /*!< in: mtr */
{
    page_t*        page;
    buf_block_t*    block;
    ulint        space;
    buf_block_t*    guess;
    ulint        height;
    ulint        page_no;
    ulint        up_match;
    ulint        up_bytes;
    ulint        low_match;
    ulint        low_bytes;
    ulint        savepoint;
    ulint        rw_latch;
    ulint        page_mode;
    ulint        buf_mode;
    ulint        estimate;
    ulint        zip_size;
    page_cur_t*    page_cursor;
    btr_op_t    btr_op;
    ulint        root_height = 0; /* remove warning */

#ifdef BTR_CUR_ADAPT
    btr_search_t*    info;
#endif
    mem_heap_t*    heap        = NULL;
    ulint        offsets_[REC_OFFS_NORMAL_SIZE];
    ulint*        offsets        = offsets_;
    rec_offs_init(offsets_);
    /* Currently, PAGE_CUR_LE is the only search mode used for searches
    ending to upper levels */

    ut_ad(level == 0 || mode == PAGE_CUR_LE);
    ut_ad(dict_index_check_search_tuple(index, tuple));
    ut_ad(!dict_index_is_ibuf(index) || ibuf_inside(mtr));
    ut_ad(dtuple_check_typed(tuple));
    ut_ad(index->page != FIL_NULL);

    UNIV_MEM_INVALID(&cursor->up_match, sizeof cursor->up_match);
    UNIV_MEM_INVALID(&cursor->up_bytes, sizeof cursor->up_bytes);
    UNIV_MEM_INVALID(&cursor->low_match, sizeof cursor->low_match);
    UNIV_MEM_INVALID(&cursor->low_bytes, sizeof cursor->low_bytes);
#ifdef UNIV_DEBUG
    cursor->up_match = ULINT_UNDEFINED;
    cursor->low_match = ULINT_UNDEFINED;
#endif

    /* These flags are mutually exclusive, they are lumped together
    with the latch mode for historical reasons. It‘s possible for
    none of the flags to be set. */
    switch (UNIV_EXPECT(latch_mode
                & (BTR_INSERT | BTR_DELETE | BTR_DELETE_MARK),
                0)) {
    case 0:
        btr_op = BTR_NO_OP;
        break;
    case BTR_INSERT:
        btr_op = (latch_mode & BTR_IGNORE_SEC_UNIQUE)
            ? BTR_INSERT_IGNORE_UNIQUE_OP
            : BTR_INSERT_OP;
        break;
    case BTR_DELETE:
        btr_op = BTR_DELETE_OP;
        ut_a(cursor->purge_node);
        break;
    case BTR_DELETE_MARK:
        btr_op = BTR_DELMARK_OP;
        break;
    default:
        /* only one of BTR_INSERT, BTR_DELETE, BTR_DELETE_MARK
        should be specified at a time */
        ut_error;
    }

    /* Operations on the insert buffer tree cannot be buffered. */
    ut_ad(btr_op == BTR_NO_OP || !dict_index_is_ibuf(index));
    /* Operations on the clustered index cannot be buffered. */
    ut_ad(btr_op == BTR_NO_OP || !dict_index_is_clust(index));

    estimate = latch_mode & BTR_ESTIMATE;

    /* Turn the flags unrelated to the latch mode off. */
    latch_mode &= ~(BTR_INSERT
            | BTR_DELETE_MARK
            | BTR_DELETE
            | BTR_ESTIMATE
            | BTR_IGNORE_SEC_UNIQUE);

    cursor->flag = BTR_CUR_BINARY;
    cursor->index = index;

#ifndef BTR_CUR_ADAPT
    guess = NULL;
#else
    info = btr_search_get_info(index);

    guess = info->root_guess;

#ifdef BTR_CUR_HASH_ADAPT

#ifdef UNIV_SEARCH_PERF_STAT
    info->n_searches++;
#endif
    if (rw_lock_get_writer(&btr_search_latch) == RW_LOCK_NOT_LOCKED
        && latch_mode <= BTR_MODIFY_LEAF
        && info->last_hash_succ
        && !estimate
#ifdef PAGE_CUR_LE_OR_EXTENDS
        && mode != PAGE_CUR_LE_OR_EXTENDS
#endif /* PAGE_CUR_LE_OR_EXTENDS */
        /* If !has_search_latch, we do a dirty read of
        btr_search_enabled below, and btr_search_guess_on_hash()
        will have to check it again. */
        && UNIV_LIKELY(btr_search_enabled)
        && btr_search_guess_on_hash(index, info, tuple, mode,
                    latch_mode, cursor,
                    has_search_latch, mtr)) {

        /* Search using the hash index succeeded */

        ut_ad(cursor->up_match != ULINT_UNDEFINED
              || mode != PAGE_CUR_GE);
        ut_ad(cursor->up_match != ULINT_UNDEFINED
              || mode != PAGE_CUR_LE);
        ut_ad(cursor->low_match != ULINT_UNDEFINED
              || mode != PAGE_CUR_LE);
        btr_cur_n_sea++;

        return;
    }
#endif /* BTR_CUR_HASH_ADAPT */
#endif /* BTR_CUR_ADAPT */
    btr_cur_n_non_sea++;

    /* If the hash search did not succeed, do binary search down the
    tree */

    if (has_search_latch) {
        /* Release possible search latch to obey latching order */
        rw_lock_s_unlock(&btr_search_latch);
    }

    /* Store the position of the tree latch we push to mtr so that we
    know how to release it when we have latched leaf node(s) */

    savepoint = mtr_set_savepoint(mtr);

    if (latch_mode == BTR_MODIFY_TREE) {
        mtr_x_lock(dict_index_get_lock(index), mtr);

    } else if (latch_mode == BTR_CONT_MODIFY_TREE) {
        /* Do nothing */
        ut_ad(mtr_memo_contains(mtr, dict_index_get_lock(index),
                    MTR_MEMO_X_LOCK));
    } else {
        mtr_s_lock(dict_index_get_lock(index), mtr);
    }

    page_cursor = btr_cur_get_page_cur(cursor);

    space = dict_index_get_space(index);
    page_no = dict_index_get_page(index);

    up_match = 0;
    up_bytes = 0;
    low_match = 0;
    low_bytes = 0;

    height = ULINT_UNDEFINED;

    /* We use these modified search modes on non-leaf levels of the
    B-tree. These let us end up in the right B-tree leaf. In that leaf
    we use the original search mode. */

    switch (mode) {
    case PAGE_CUR_GE:
        page_mode = PAGE_CUR_L;
        break;
    case PAGE_CUR_G:
        page_mode = PAGE_CUR_LE;
        break;
    default:
#ifdef PAGE_CUR_LE_OR_EXTENDS
        ut_ad(mode == PAGE_CUR_L || mode == PAGE_CUR_LE
              || mode == PAGE_CUR_LE_OR_EXTENDS);
#else /* PAGE_CUR_LE_OR_EXTENDS */
        ut_ad(mode == PAGE_CUR_L || mode == PAGE_CUR_LE);
#endif /* PAGE_CUR_LE_OR_EXTENDS */
        page_mode = mode;
        break;
    }

    /* Loop and search until we arrive at the desired level */

search_loop:
    buf_mode = BUF_GET;
    rw_latch = RW_NO_LATCH;

    if (height != 0) {
        /* We are about to fetch the root or a non-leaf page. */
    } else if (latch_mode <= BTR_MODIFY_LEAF) {
        rw_latch = latch_mode;

        if (btr_op != BTR_NO_OP
            && ibuf_should_try(index, btr_op != BTR_INSERT_OP)) {

            /* Try to buffer the operation if the leaf
            page is not in the buffer pool. */

            buf_mode = btr_op == BTR_DELETE_OP
                ? BUF_GET_IF_IN_POOL_OR_WATCH
                : BUF_GET_IF_IN_POOL;
        }
    }

    zip_size = dict_table_zip_size(index->table);

retry_page_get:
    block = buf_page_get_gen(
        space, zip_size, page_no, rw_latch, guess, buf_mode,
        file, line, mtr);

    if (block == NULL) {
        /* This must be a search to perform an insert/delete
        mark/ delete; try using the insert/delete buffer */

        ut_ad(height == 0);
        ut_ad(cursor->thr);

        switch (btr_op) {
        case BTR_INSERT_OP:
        case BTR_INSERT_IGNORE_UNIQUE_OP:
            ut_ad(buf_mode == BUF_GET_IF_IN_POOL);

            if (ibuf_insert(IBUF_OP_INSERT, tuple, index,
                    space, zip_size, page_no,
                    cursor->thr)) {

                cursor->flag = BTR_CUR_INSERT_TO_IBUF;

                goto func_exit;
            }
            break;

        case BTR_DELMARK_OP:
            ut_ad(buf_mode == BUF_GET_IF_IN_POOL);

            if (ibuf_insert(IBUF_OP_DELETE_MARK, tuple,
                    index, space, zip_size,
                    page_no, cursor->thr)) {

                cursor->flag = BTR_CUR_DEL_MARK_IBUF;

                goto func_exit;
            }

            break;

        case BTR_DELETE_OP:
            ut_ad(buf_mode == BUF_GET_IF_IN_POOL_OR_WATCH);

            if (!row_purge_poss_sec(cursor->purge_node,
                        index, tuple)) {

                /* The record cannot be purged yet. */
                cursor->flag = BTR_CUR_DELETE_REF;
            } else if (ibuf_insert(IBUF_OP_DELETE, tuple,
                           index, space, zip_size,
                           page_no,
                           cursor->thr)) {

                /* The purge was buffered. */
                cursor->flag = BTR_CUR_DELETE_IBUF;
            } else {
                /* The purge could not be buffered. */
                buf_pool_watch_unset(space, page_no);
                break;
            }

            buf_pool_watch_unset(space, page_no);
            goto func_exit;

        default:
            ut_error;
        }

        /* Insert to the insert/delete buffer did not succeed, we
        must read the page from disk. */

        buf_mode = BUF_GET;

        goto retry_page_get;
    }

    block->check_index_page_at_flush = TRUE;
    page = buf_block_get_frame(block);

    if (rw_latch != RW_NO_LATCH) {
#ifdef UNIV_ZIP_DEBUG
        const page_zip_des_t*    page_zip
            = buf_block_get_page_zip(block);
        ut_a(!page_zip || page_zip_validate(page_zip, page, index));
#endif /* UNIV_ZIP_DEBUG */

        buf_block_dbg_add_level(
            block, dict_index_is_ibuf(index)
            ? SYNC_IBUF_TREE_NODE : SYNC_TREE_NODE);
    }

    ut_ad(index->id == btr_page_get_index_id(page));

    if (UNIV_UNLIKELY(height == ULINT_UNDEFINED)) {
        /* We are in the root node */

        height = btr_page_get_level(page, mtr);
        root_height = height;
        cursor->tree_height = root_height + 1;

#ifdef BTR_CUR_ADAPT
        if (block != guess) {
            info->root_guess = block;
        }
#endif
    }

    if (height == 0) {
        if (rw_latch == RW_NO_LATCH) {

            btr_cur_latch_leaves(
                page, space, zip_size, page_no, latch_mode,
                cursor, mtr);
        }

        if (latch_mode != BTR_MODIFY_TREE
            && latch_mode != BTR_CONT_MODIFY_TREE) {

            /* Release the tree s-latch */

            mtr_release_s_latch_at_savepoint(
                mtr, savepoint, dict_index_get_lock(index));
        }

        page_mode = mode;
    }

    page_cur_search_with_match(
        block, index, tuple, page_mode, &up_match, &up_bytes,
        &low_match, &low_bytes, page_cursor);

    if (estimate) {
        btr_cur_add_path_info(cursor, height, root_height);
    }

    /* If this is the desired level, leave the loop */

    ut_ad(height == btr_page_get_level(page_cur_get_page(page_cursor),
                       mtr));

    if (level != height) {

        const rec_t*    node_ptr;
        ut_ad(height > 0);

        height--;
        guess = NULL;

        node_ptr = page_cur_get_rec(page_cursor);

        offsets = rec_get_offsets(
            node_ptr, index, offsets, ULINT_UNDEFINED, &heap);

        /* Go to the child node */
        page_no = btr_node_ptr_get_child_page_no(node_ptr, offsets);

        if (UNIV_UNLIKELY(height == 0 && dict_index_is_ibuf(index))) {
            /* We‘re doing a search on an ibuf tree and we‘re one
            level above the leaf page. */

            ut_ad(level == 0);

            buf_mode = BUF_GET;
            rw_latch = RW_NO_LATCH;
            goto retry_page_get;
        }

        goto search_loop;
    }

    if (level != 0) {
        /* x-latch the page */
        buf_block_t*    child_block = btr_block_get(
            space, zip_size, page_no, RW_X_LATCH, index, mtr);

        page = buf_block_get_frame(child_block);
        btr_assert_not_corrupted(child_block, index);
    } else {
        cursor->low_match = low_match;
        cursor->low_bytes = low_bytes;
        cursor->up_match = up_match;
        cursor->up_bytes = up_bytes;

#ifdef BTR_CUR_ADAPT
        /* We do a dirty read of btr_search_enabled here.  We
        will properly check btr_search_enabled again in
        btr_search_build_page_hash_index() before building a
        page hash index, while holding btr_search_latch. */
        if (UNIV_LIKELY(btr_search_enabled)) {

            btr_search_info_update(index, cursor);
        }
#endif
        ut_ad(cursor->up_match != ULINT_UNDEFINED
              || mode != PAGE_CUR_GE);
        ut_ad(cursor->up_match != ULINT_UNDEFINED
              || mode != PAGE_CUR_LE);
        ut_ad(cursor->low_match != ULINT_UNDEFINED
              || mode != PAGE_CUR_LE);
    }

func_exit:

    if (UNIV_LIKELY_NULL(heap)) {
        mem_heap_free(heap);
    }

    if (has_search_latch) {

        rw_lock_s_lock(&btr_search_latch);
    }
}

/****************************************************************//**
Searches the right position for a page cursor. */
UNIV_INTERN
void
page_cur_search_with_match(
/*=======================*/
    const buf_block_t*    block,    /*!< in: buffer block */
    const dict_index_t*    index,    /*!< in: record descriptor */
    const dtuple_t*        tuple,    /*!< in: data tuple */
    ulint            mode,    /*!< in: PAGE_CUR_L,
                    PAGE_CUR_LE, PAGE_CUR_G, or
                    PAGE_CUR_GE */
    ulint*            iup_matched_fields,
                    /*!< in/out: already matched
                    fields in upper limit record */
    ulint*            iup_matched_bytes,
                    /*!< in/out: already matched
                    bytes in a field not yet
                    completely matched */
    ulint*            ilow_matched_fields,
                    /*!< in/out: already matched
                    fields in lower limit record */
    ulint*            ilow_matched_bytes,
                    /*!< in/out: already matched
                    bytes in a field not yet
                    completely matched */
    page_cur_t*        cursor)    /*!< out: page cursor */
{
    ulint        up;
    ulint        low;
    ulint        mid;
    const page_t*    page;
    const page_dir_slot_t* slot;
    const rec_t*    up_rec;
    const rec_t*    low_rec;
    const rec_t*    mid_rec;
    ulint        up_matched_fields;
    ulint        up_matched_bytes;
    ulint        low_matched_fields;
    ulint        low_matched_bytes;
    ulint        cur_matched_fields;
    ulint        cur_matched_bytes;
    int        cmp;
#ifdef UNIV_SEARCH_DEBUG
    int        dbg_cmp;
    ulint        dbg_matched_fields;
    ulint        dbg_matched_bytes;
#endif
#ifdef UNIV_ZIP_DEBUG
    const page_zip_des_t*    page_zip = buf_block_get_page_zip(block);
#endif /* UNIV_ZIP_DEBUG */
    mem_heap_t*    heap        = NULL;
    ulint        offsets_[REC_OFFS_NORMAL_SIZE];
    ulint*        offsets        = offsets_;
    rec_offs_init(offsets_);

    ut_ad(block && tuple && iup_matched_fields && iup_matched_bytes
          && ilow_matched_fields && ilow_matched_bytes && cursor);
    ut_ad(dtuple_validate(tuple));
#ifdef UNIV_DEBUG
# ifdef PAGE_CUR_DBG
    if (mode != PAGE_CUR_DBG)
# endif /* PAGE_CUR_DBG */
# ifdef PAGE_CUR_LE_OR_EXTENDS
        if (mode != PAGE_CUR_LE_OR_EXTENDS)
# endif /* PAGE_CUR_LE_OR_EXTENDS */
            ut_ad(mode == PAGE_CUR_L || mode == PAGE_CUR_LE
                  || mode == PAGE_CUR_G || mode == PAGE_CUR_GE);
#endif /* UNIV_DEBUG */
    page = buf_block_get_frame(block);
#ifdef UNIV_ZIP_DEBUG
    ut_a(!page_zip || page_zip_validate(page_zip, page, index));
#endif /* UNIV_ZIP_DEBUG */

    page_check_dir(page);

#ifdef PAGE_CUR_ADAPT
    if (page_is_leaf(page)
        && (mode == PAGE_CUR_LE)
        && (page_header_get_field(page, PAGE_N_DIRECTION) > 3)
        && (page_header_get_ptr(page, PAGE_LAST_INSERT))
        && (page_header_get_field(page, PAGE_DIRECTION) == PAGE_RIGHT)) {

        if (page_cur_try_search_shortcut(
                block, index, tuple,
                iup_matched_fields, iup_matched_bytes,
                ilow_matched_fields, ilow_matched_bytes,
                cursor)) {
            return;
        }
    }
# ifdef PAGE_CUR_DBG
    if (mode == PAGE_CUR_DBG) {
        mode = PAGE_CUR_LE;
    }
# endif
#endif

    /* The following flag does not work for non-latin1 char sets because
    cmp_full_field does not tell how many bytes matched */
#ifdef PAGE_CUR_LE_OR_EXTENDS
    ut_a(mode != PAGE_CUR_LE_OR_EXTENDS);
#endif /* PAGE_CUR_LE_OR_EXTENDS */

    /* If mode PAGE_CUR_G is specified, we are trying to position the
    cursor to answer a query of the form "tuple < X", where tuple is
    the input parameter, and X denotes an arbitrary physical record on
    the page. We want to position the cursor on the first X which
    satisfies the condition. */

    up_matched_fields  = *iup_matched_fields;
    up_matched_bytes   = *iup_matched_bytes;
    low_matched_fields = *ilow_matched_fields;
    low_matched_bytes  = *ilow_matched_bytes;

    /* Perform binary search. First the search is done through the page
    directory, after that as a linear search in the list of records
    owned by the upper limit directory slot. */

    low = 0;
    up = page_dir_get_n_slots(page) - 1;

    /* Perform binary search until the lower and upper limit directory
    slots come to the distance 1 of each other */

    while (up - low > 1) {
        mid = (low + up) / 2;
        slot = page_dir_get_nth_slot(page, mid);
        mid_rec = page_dir_slot_get_rec(slot);

        ut_pair_min(&cur_matched_fields, &cur_matched_bytes,
                low_matched_fields, low_matched_bytes,
                up_matched_fields, up_matched_bytes);

        offsets = rec_get_offsets(mid_rec, index, offsets,
                      dtuple_get_n_fields_cmp(tuple),
                      &heap);

        cmp = cmp_dtuple_rec_with_match(tuple, mid_rec, offsets,
                        &cur_matched_fields,
                        &cur_matched_bytes);
        if (UNIV_LIKELY(cmp > 0)) {
low_slot_match:
            low = mid;
            low_matched_fields = cur_matched_fields;
            low_matched_bytes = cur_matched_bytes;

        } else if (UNIV_EXPECT(cmp, -1)) {
#ifdef PAGE_CUR_LE_OR_EXTENDS
            if (mode == PAGE_CUR_LE_OR_EXTENDS
                && page_cur_rec_field_extends(
                    tuple, mid_rec, offsets,
                    cur_matched_fields)) {

                goto low_slot_match;
            }
#endif /* PAGE_CUR_LE_OR_EXTENDS */
up_slot_match:
            up = mid;
            up_matched_fields = cur_matched_fields;
            up_matched_bytes = cur_matched_bytes;

        } else if (mode == PAGE_CUR_G || mode == PAGE_CUR_LE
#ifdef PAGE_CUR_LE_OR_EXTENDS
               || mode == PAGE_CUR_LE_OR_EXTENDS
#endif /* PAGE_CUR_LE_OR_EXTENDS */
               ) {

            goto low_slot_match;
        } else {

            goto up_slot_match;
        }
    }

    slot = page_dir_get_nth_slot(page, low);
    low_rec = page_dir_slot_get_rec(slot);
    slot = page_dir_get_nth_slot(page, up);
    up_rec = page_dir_slot_get_rec(slot);

    /* Perform linear search until the upper and lower records come to
    distance 1 of each other. */

    while (page_rec_get_next_const(low_rec) != up_rec) {

        mid_rec = page_rec_get_next_const(low_rec);

        ut_pair_min(&cur_matched_fields, &cur_matched_bytes,
                low_matched_fields, low_matched_bytes,
                up_matched_fields, up_matched_bytes);

        offsets = rec_get_offsets(mid_rec, index, offsets,
                      dtuple_get_n_fields_cmp(tuple),
                      &heap);

        cmp = cmp_dtuple_rec_with_match(tuple, mid_rec, offsets,
                        &cur_matched_fields,
                        &cur_matched_bytes);
        if (UNIV_LIKELY(cmp > 0)) {
low_rec_match:
            low_rec = mid_rec;
            low_matched_fields = cur_matched_fields;
            low_matched_bytes = cur_matched_bytes;

        } else if (UNIV_EXPECT(cmp, -1)) {
#ifdef PAGE_CUR_LE_OR_EXTENDS
            if (mode == PAGE_CUR_LE_OR_EXTENDS
                && page_cur_rec_field_extends(
                    tuple, mid_rec, offsets,
                    cur_matched_fields)) {

                goto low_rec_match;
            }
#endif /* PAGE_CUR_LE_OR_EXTENDS */
up_rec_match:
            up_rec = mid_rec;
            up_matched_fields = cur_matched_fields;
            up_matched_bytes = cur_matched_bytes;
        } else if (mode == PAGE_CUR_G || mode == PAGE_CUR_LE
#ifdef PAGE_CUR_LE_OR_EXTENDS
               || mode == PAGE_CUR_LE_OR_EXTENDS
#endif /* PAGE_CUR_LE_OR_EXTENDS */
               ) {

            goto low_rec_match;
        } else {

            goto up_rec_match;
        }
    }

#ifdef UNIV_SEARCH_DEBUG

    /* Check that the lower and upper limit records have the
    right alphabetical order compared to tuple. */
    dbg_matched_fields = 0;
    dbg_matched_bytes = 0;

    offsets = rec_get_offsets(low_rec, index, offsets,
                  ULINT_UNDEFINED, &heap);
    dbg_cmp = page_cmp_dtuple_rec_with_match(tuple, low_rec, offsets,
                         &dbg_matched_fields,
                         &dbg_matched_bytes);
    if (mode == PAGE_CUR_G) {
        ut_a(dbg_cmp >= 0);
    } else if (mode == PAGE_CUR_GE) {
        ut_a(dbg_cmp == 1);
    } else if (mode == PAGE_CUR_L) {
        ut_a(dbg_cmp == 1);
    } else if (mode == PAGE_CUR_LE) {
        ut_a(dbg_cmp >= 0);
    }

    if (!page_rec_is_infimum(low_rec)) {

        ut_a(low_matched_fields == dbg_matched_fields);
        ut_a(low_matched_bytes == dbg_matched_bytes);
    }

    dbg_matched_fields = 0;
    dbg_matched_bytes = 0;

    offsets = rec_get_offsets(up_rec, index, offsets,
                  ULINT_UNDEFINED, &heap);
    dbg_cmp = page_cmp_dtuple_rec_with_match(tuple, up_rec, offsets,
                         &dbg_matched_fields,
                         &dbg_matched_bytes);
    if (mode == PAGE_CUR_G) {
        ut_a(dbg_cmp == -1);
    } else if (mode == PAGE_CUR_GE) {
        ut_a(dbg_cmp <= 0);
    } else if (mode == PAGE_CUR_L) {
        ut_a(dbg_cmp <= 0);
    } else if (mode == PAGE_CUR_LE) {
        ut_a(dbg_cmp == -1);
    }

    if (!page_rec_is_supremum(up_rec)) {

        ut_a(up_matched_fields == dbg_matched_fields);
        ut_a(up_matched_bytes == dbg_matched_bytes);
    }
#endif
    if (mode <= PAGE_CUR_GE) {
        page_cur_position(up_rec, block, cursor);
    } else {
        page_cur_position(low_rec, block, cursor);
    }

    *iup_matched_fields  = up_matched_fields;
    *iup_matched_bytes   = up_matched_bytes;
    *ilow_matched_fields = low_matched_fields;
    *ilow_matched_bytes  = low_matched_bytes;
    if (UNIV_LIKELY_NULL(heap)) {
        mem_heap_free(heap);
    }
}

/*************************************************************//**
This function is used to compare a data tuple to a physical record.
Only dtuple->n_fields_cmp first fields are taken into account for
the data tuple! If we denote by n = n_fields_cmp, then rec must
have either m >= n fields, or it must differ from dtuple in some of
the m fields rec has. If rec has an externally stored field we do not
compare it but return with value 0 if such a comparison should be
made.
@return 1, 0, -1, if dtuple is greater, equal, less than rec,
respectively, when only the common first fields are compared, or until
the first externally stored field in rec */
UNIV_INTERN
int
cmp_dtuple_rec_with_match(
/*======================*/
    const dtuple_t*    dtuple,    /*!< in: data tuple */
    const rec_t*    rec,    /*!< in: physical record which differs from
                dtuple in some of the common fields, or which
                has an equal number or more fields than
                dtuple */
    const ulint*    offsets,/*!< in: array returned by rec_get_offsets() */
    ulint*        matched_fields, /*!< in/out: number of already completely
                matched fields; when function returns,
                contains the value for current comparison */
    ulint*        matched_bytes) /*!< in/out: number of already matched
                bytes within the first field not completely
                matched; when function returns, contains the
                value for current comparison */
{
    const dfield_t*    dtuple_field;    /* current field in logical record */
    ulint        dtuple_f_len;    /* the length of the current field
                    in the logical record */
    const byte*    dtuple_b_ptr;    /* pointer to the current byte in
                    logical field data */
    ulint        dtuple_byte;    /* value of current byte to be compared
                    in dtuple*/
    ulint        rec_f_len;    /* length of current field in rec */
    const byte*    rec_b_ptr;    /* pointer to the current byte in
                    rec field */
    ulint        rec_byte;    /* value of current byte to be
                    compared in rec */
    ulint        cur_field;    /* current field number */
    ulint        cur_bytes;    /* number of already matched bytes
                    in current field */
    int        ret = 3333;    /* return value */

    ut_ad(dtuple && rec && matched_fields && matched_bytes);
    ut_ad(dtuple_check_typed(dtuple));
    ut_ad(rec_offs_validate(rec, NULL, offsets));

    cur_field = *matched_fields;
    cur_bytes = *matched_bytes;

    ut_ad(cur_field <= dtuple_get_n_fields_cmp(dtuple));
    ut_ad(cur_field <= rec_offs_n_fields(offsets));

    if (cur_bytes == 0 && cur_field == 0) {
        ulint    rec_info = rec_get_info_bits(rec,
                             rec_offs_comp(offsets));
        ulint    tup_info = dtuple_get_info_bits(dtuple);

        if (UNIV_UNLIKELY(rec_info & REC_INFO_MIN_REC_FLAG)) {
            ret = !(tup_info & REC_INFO_MIN_REC_FLAG);
            goto order_resolved;
        } else if (UNIV_UNLIKELY(tup_info & REC_INFO_MIN_REC_FLAG)) {
            ret = -1;
            goto order_resolved;
        }
    }

    /* Match fields in a loop; stop if we run out of fields in dtuple
    or find an externally stored field */

    while (cur_field < dtuple_get_n_fields_cmp(dtuple)) {

        ulint    mtype;
        ulint    prtype;

        dtuple_field = dtuple_get_nth_field(dtuple, cur_field);
        {
            const dtype_t*    type
                = dfield_get_type(dtuple_field);

            mtype = type->mtype;
            prtype = type->prtype;
        }

        dtuple_f_len = dfield_get_len(dtuple_field);

        rec_b_ptr = rec_get_nth_field(rec, offsets,
                          cur_field, &rec_f_len);

        /* If we have matched yet 0 bytes, it may be that one or
        both the fields are SQL null, or the record or dtuple may be
        the predefined minimum record, or the field is externally
        stored */

        if (UNIV_LIKELY(cur_bytes == 0)) {
            if (rec_offs_nth_extern(offsets, cur_field)) {
                /* We do not compare to an externally
                stored field */

                ret = 0;

                goto order_resolved;
            }

            if (dtuple_f_len == UNIV_SQL_NULL) {
                if (rec_f_len == UNIV_SQL_NULL) {

                    goto next_field;
                }

                ret = -1;
                goto order_resolved;
            } else if (rec_f_len == UNIV_SQL_NULL) {
                /* We define the SQL null to be the
                smallest possible value of a field
                in the alphabetical order */

                ret = 1;
                goto order_resolved;
            }
        }

        if (mtype >= DATA_FLOAT
            || (mtype == DATA_BLOB
            && 0 == (prtype & DATA_BINARY_TYPE)
            && dtype_get_charset_coll(prtype)
            != DATA_MYSQL_LATIN1_SWEDISH_CHARSET_COLL)) {

            ret = cmp_whole_field(mtype, prtype,
                          dfield_get_data(dtuple_field),
                          (unsigned) dtuple_f_len,
                          rec_b_ptr, (unsigned) rec_f_len);

            if (ret != 0) {
                cur_bytes = 0;

                goto order_resolved;
            } else {
                goto next_field;
            }
        }

        /* Set the pointers at the current byte */

        rec_b_ptr = rec_b_ptr + cur_bytes;
        dtuple_b_ptr = (byte*)dfield_get_data(dtuple_field)
            + cur_bytes;
        /* Compare then the fields */

        for (;;) {
            if (UNIV_UNLIKELY(rec_f_len <= cur_bytes)) {
                if (dtuple_f_len <= cur_bytes) {

                    goto next_field;
                }

                rec_byte = dtype_get_pad_char(mtype, prtype);

                if (rec_byte == ULINT_UNDEFINED) {
                    ret = 1;

                    goto order_resolved;
                }
            } else {
                rec_byte = *rec_b_ptr;
            }

            if (UNIV_UNLIKELY(dtuple_f_len <= cur_bytes)) {
                dtuple_byte = dtype_get_pad_char(mtype,
                                 prtype);

                if (dtuple_byte == ULINT_UNDEFINED) {
                    ret = -1;

                    goto order_resolved;
                }
            } else {
                dtuple_byte = *dtuple_b_ptr;
            }

            if (dtuple_byte == rec_byte) {
                /* If the bytes are equal, they will
                remain such even after the collation
                transformation below */

                goto next_byte;
            }

            if (mtype <= DATA_CHAR
                || (mtype == DATA_BLOB
                && !(prtype & DATA_BINARY_TYPE))) {

                rec_byte = cmp_collate(rec_byte);
                dtuple_byte = cmp_collate(dtuple_byte);
            }

            ret = (int) (dtuple_byte - rec_byte);
            if (UNIV_LIKELY(ret)) {
                if (ret < 0) {
                    ret = -1;
                    goto order_resolved;
                } else {
                    ret = 1;
                    goto order_resolved;
                }
            }
next_byte:
            /* Next byte */
            cur_bytes++;
            rec_b_ptr++;
            dtuple_b_ptr++;
        }

next_field:
        cur_field++;
        cur_bytes = 0;
    }

    ut_ad(cur_bytes == 0);

    ret = 0;    /* If we ran out of fields, dtuple was equal to rec
            up to the common fields */
order_resolved:
    ut_ad((ret >= - 1) && (ret <= 1));
    ut_ad(ret == cmp_debug_dtuple_rec_with_match(dtuple, rec, offsets,
                             matched_fields));
    ut_ad(*matched_fields == cur_field); /* In the debug version, the
                         above cmp_debug_... sets
                         *matched_fields to a value */
    *matched_fields = cur_field;
    *matched_bytes = cur_bytes;

    return(ret);
}