10418752118

Committed 16 Aug 2024 10:51AM UTC coverage: 85.908% (-0.02%) from 85.926%

Build # 10418752118

Build Type

push

github

Committed by

sergepetrenko

Commit Message

engine: introduce stubs for checkpoint FETCH_SNAPSHOT

This commit introduces engine stubs that enable a new method
of fetching snapshots for anonymous replicas. Instead of using
the traditional read-view join approach, this update allows
file snapshot fetching. Note that file snapshot fetching
is only available in Tarantool EE.

Checkpoint fetching is done via IPROTO_IS_CHECKPOINT_JOIN,
IPROTO_CHECKPOINT_VCLOCK and IPROTO_CHECKPOINT_LSN fields.

If IPROTO_CHECKPOINT_JOIN is set to true, join will be done from
files: .snap for memtx, .run for vinyl, if false - from read view.

Checkpoint join allows to continue from the place, where client
stopped in case of snapshot fetching error. This allows to avoid
rebootstrap of an anonymous client. This can be done by specifying
CHECKPOINT_VCLOCK, which says from which file server should continue
join, client gets vclock at the beginning of the join. Specifying
CHECKPOINT_LSN allows to continue from some position in checkpoint.
Server sends all data >= CHECKPOINT_LSN.

If CHECKPOINT_VCLOCK is not specified, fetching is done from the latest
available checkpoint. If CHECKPOINT_LSN is not specified - start from
the beginning of the snap. So, specifying only IS_CHECKPOINT_JOIN
triggers fetching the latest checkpoint from files.

Needed for tarantool/tarantool-ee#741

NO_DOC=ee
NO_TEST=ee
NO_CHANGELOG=ee

(cherry picked from commit 2fca5c133)

Run Details

62977 of 114211 branches covered (55.14%)

17 of 63 new or added lines in 6 files covered. (26.98%)

459 existing lines in 17 files now uncovered.

93867 of 109264 relevant lines covered (85.91%)

2444495.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.18

/src/box/vy_run.c

/*
 * Copyright 2010-2017, Tarantool AUTHORS, please see AUTHORS file.
 *
 * Redistribution and use in source and binary forms, with or
 * without modification, are permitted provided that the following
 * conditions are met:
 *
 * 1. Redistributions of source code must retain the above
 *    copyright notice, this list of conditions and the
 *    following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above
 *    copyright notice, this list of conditions and the following
 *    disclaimer in the documentation and/or other materials
 *    provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY AUTHORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
 * AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
 * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
 * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
#include "vy_run.h"

#include <zstd.h>

#include "fiber.h"
#include "fiber_cond.h"
#include "fio.h"
#include "cbus.h"
#include "memory.h"
#include "coio_file.h"

#include "replication.h"
#include "tuple_bloom.h"
#include "xlog.h"
#include "xrow.h"
#include "vy_history.h"

static const uint64_t vy_page_info_key_map = (1 << VY_PAGE_INFO_OFFSET) |
                                             (1 << VY_PAGE_INFO_SIZE) |
                                             (1 << VY_PAGE_INFO_UNPACKED_SIZE) |
                                             (1 << VY_PAGE_INFO_ROW_COUNT) |
                                             (1 << VY_PAGE_INFO_MIN_KEY) |
                                             (1 << VY_PAGE_INFO_ROW_INDEX_OFFSET);

static const uint64_t vy_run_info_key_map = (1 << VY_RUN_INFO_MIN_KEY) |
                                            (1 << VY_RUN_INFO_MAX_KEY) |
                                            (1 << VY_RUN_INFO_MIN_LSN) |
                                            (1 << VY_RUN_INFO_MAX_LSN) |
                                            (1 << VY_RUN_INFO_PAGE_COUNT);

/** xlog meta type for .run files */
#define XLOG_META_TYPE_RUN "RUN"

/** xlog meta type for .index files */
#define XLOG_META_TYPE_INDEX "INDEX"

const char *vy_file_suffix[] = {
        "index",                        /* VY_FILE_INDEX */
        "index" inprogress_suffix,         /* VY_FILE_INDEX_INPROGRESS */
        "run",                                /* VY_FILE_RUN */
        "run" inprogress_suffix,         /* VY_FILE_RUN_INPROGRESS */
};

/* sync run and index files very 16 MB */
#define VY_RUN_SYNC_INTERVAL (1 << 24)

/**
 * We read runs in background threads so as not to stall tx.
 * This structure represents such a thread.
 */
struct vy_run_reader {
        /** Thread that processes read requests. */
        struct cord cord;
        /** Pipe from tx to the reader thread. */
        struct cpipe reader_pipe;
        /** Pipe from the reader thread to tx. */
        struct cpipe tx_pipe;
};

/** Cbus task for vinyl page read. */
struct vy_page_read_task {
        /** parent */
        struct cbus_call_msg base;
        /** vinyl page metadata */
        struct vy_page_info *page_info;
        /** vy_run with fd - ref. counted */
        struct vy_run *run;
        /** key to lookup within the page */
        struct vy_entry key;
        /** iterator type (needed for for key lookup) */
        enum iterator_type iterator_type;
        /** key definition (needed for key lookup) */
        struct key_def *cmp_def;
        /** disk format (needed for key lookup) */
        struct tuple_format *format;
        /** [out] position of the key in the page */
        uint32_t pos_in_page;
        /** [out] true if key was found in the page */
        bool equal_found;
        /** [out] resulting vinyl page */
        struct vy_page *page;
};

/** Destructor for env->zdctx_key thread-local variable */
static void
vy_free_zdctx(void *arg)
{
        assert(arg != NULL);
        ZSTD_freeDStream(arg);
}

/** Run reader thread function. */
static int
vy_run_reader_f(va_list ap)
{
        struct vy_run_reader *reader = va_arg(ap, struct vy_run_reader *);
        struct cbus_endpoint endpoint;

        cpipe_create(&reader->tx_pipe, "tx_prio");
        cbus_endpoint_create(&endpoint, cord_name(cord()),
                             fiber_schedule_cb, fiber());
        cbus_loop(&endpoint);
        cbus_endpoint_destroy(&endpoint, cbus_process);
        cpipe_destroy(&reader->tx_pipe);
        return 0;
}

/** Start run reader threads. */
static void
vy_run_env_start_readers(struct vy_run_env *env)
{
        assert(env->reader_pool == NULL);
        assert(env->reader_pool_size > 0);

        env->reader_pool = calloc(env->reader_pool_size,
                                  sizeof(*env->reader_pool));
        if (env->reader_pool == NULL)
                panic("failed to allocate vinyl reader thread pool");

        for (int i = 0; i < env->reader_pool_size; i++) {
                struct vy_run_reader *reader = &env->reader_pool[i];
                char name[FIBER_NAME_MAX];

                snprintf(name, sizeof(name), "vinyl.reader.%d", i);
                if (cord_costart(&reader->cord, name,
                                 vy_run_reader_f, reader) != 0)
                        panic("failed to start vinyl reader thread");
                cpipe_create(&reader->reader_pipe, name);
        }
        env->next_reader = 0;
}

/** Join run reader threads. */
static void
vy_run_env_stop_readers(struct vy_run_env *env)
{
        for (int i = 0; i < env->reader_pool_size; i++) {
                struct vy_run_reader *reader = &env->reader_pool[i];
                cord_cancel_and_join(&reader->cord);
        }
        free(env->reader_pool);
}

/**
 * Initialize vinyl run environment
 */
void
vy_run_env_create(struct vy_run_env *env, int read_threads)
{
        memset(env, 0, sizeof(*env));
        env->reader_pool_size = read_threads;
        tt_pthread_key_create(&env->zdctx_key, vy_free_zdctx);
        mempool_create(&env->read_task_pool, cord_slab_cache(),
                       sizeof(struct vy_page_read_task));
        env->initial_join = false;
}

/**
 * Destroy vinyl run environment
 */
void
vy_run_env_destroy(struct vy_run_env *env)
{
        if (env->reader_pool != NULL)
                vy_run_env_stop_readers(env);
        mempool_destroy(&env->read_task_pool);
        tt_pthread_key_delete(env->zdctx_key);
}

/**
 * Enable coio reads for a vinyl run environment.
 */
void
vy_run_env_enable_coio(struct vy_run_env *env)
{
        if (env->reader_pool != NULL)
                return; /* already enabled */
        vy_run_env_start_readers(env);
}

/**
 * Execute a task on behalf of a reader thread.
 */
static int
vy_run_env_coio_call(struct vy_run_env *env, struct cbus_call_msg *msg,
                     cbus_call_f func)
{
        /* Optimization: use blocking I/O during WAL recovery. */
        if (env->reader_pool == NULL)
                return func(msg);

        /* Pick a reader thread. */
        struct vy_run_reader *reader;
        reader = &env->reader_pool[env->next_reader++];
        env->next_reader %= env->reader_pool_size;

        /* Post the task to the reader thread. */
        if (cbus_call(&reader->reader_pipe, &reader->tx_pipe, msg, func) != 0)
                return -1;

        if (fiber_is_cancelled()) {
                diag_set(FiberIsCancelled);
                return -1;
        }
        return 0;
}

/**
 * Initialize page info struct
 *
 * @retval 0 for Success
 * @retval -1 for error
 */
static int
vy_page_info_create(struct vy_page_info *page_info, uint64_t offset,
                    const char *min_key, struct key_def *cmp_def)
{
        memset(page_info, 0, sizeof(*page_info));
        page_info->offset = offset;
        page_info->unpacked_size = 0;
        page_info->min_key = vy_key_dup(min_key);
        if (page_info->min_key == NULL)
                return -1;
        uint32_t part_count = mp_decode_array(&min_key);
        page_info->min_key_hint = key_hint(min_key, part_count, cmp_def);
        return page_info->min_key == NULL ? -1 : 0;
}

/**
 * Destroy page info struct
 */
static void
vy_page_info_destroy(struct vy_page_info *page_info)
{
        if (page_info->min_key != NULL)
                free(page_info->min_key);
}

struct vy_run *
vy_run_new(struct vy_run_env *env, int64_t id)
{
        struct vy_run *run = calloc(1, sizeof(struct vy_run));
        if (unlikely(run == NULL)) {
                diag_set(OutOfMemory, sizeof(struct vy_run), "malloc",
                         "struct vy_run");
                return NULL;
        }
        run->env = env;
        run->id = id;
        run->dump_lsn = -1;
        run->fd = -1;
        run->refs = 1;
        rlist_create(&run->in_lsm);
        rlist_create(&run->in_unused);
        return run;
}

static void
vy_run_clear(struct vy_run *run)
{
        if (run->page_info != NULL) {
                uint32_t page_no;
                for (page_no = 0; page_no < run->info.page_count; ++page_no)
                        vy_page_info_destroy(run->page_info + page_no);
                free(run->page_info);
        }
        run->page_info = NULL;
        run->page_index_size = 0;
        run->info.page_count = 0;
        if (run->info.bloom != NULL) {
                tuple_bloom_delete(run->info.bloom);
                run->info.bloom = NULL;
        }
        free(run->info.min_key);
        run->info.min_key = NULL;
        free(run->info.max_key);
        run->info.max_key = NULL;
}

void
vy_run_delete(struct vy_run *run)
{
        assert(run->refs == 0);
        if (run->fd >= 0 && close(run->fd) < 0)
                say_syserror("close failed");
        vy_run_clear(run);
        TRASH(run);
        free(run);
}

size_t
vy_run_bloom_size(struct vy_run *run)
{
        return run->info.bloom == NULL ? 0 : tuple_bloom_size(run->info.bloom);
}

/**
 * Find a page from which the iteration of a given key must be started.
 * LE and LT: the found page definitely contains the position
 *  for iteration start.
 * GE, GT, EQ: Since page search uses only min_key of pages,
 *  it may happen that the found page doesn't contain the position
 *  for iteration start. In this case it is certain that the iteration
 *  must be started from the beginning of the next page.
 *
 * @param run - run
 * @param key - key to find
 * @param key_def - key_def for comparison
 * @param itype - iterator type (see above)
 * @param equal_key: *equal_key is set to true if there is a page
 *  with min_key equal to the given key.
 * @return offset of the page in page index OR run->info.page_count if
 *  there no pages fulfilling the conditions.
 */
static uint32_t
vy_page_index_find_page(struct vy_run *run, struct vy_entry key,
                        struct key_def *cmp_def, enum iterator_type itype,
                        bool *equal_key)
{
        if (itype == ITER_EQ)
                itype = ITER_GE; /* One day it'll become obsolete */
        assert(itype == ITER_GE || itype == ITER_GT ||
               itype == ITER_LE || itype == ITER_LT);
        int dir = iterator_direction(itype);
        *equal_key = false;

        /**
         * Binary search in page index. Depends on given iterator_type:
         *  ITER_GE: lowest page with min_key >= given key.
         *  ITER_GT: lowest page with min_key > given key.
         *  ITER_LE: highest page with min_key <= given key.
         *  ITER_LT: highest page with min_key < given key.
         *
         * Example: we are searching for a value 2 in the run of 10 pages:
         * min_key:         [1   1   2   2   2   2   2   3   3   3]
         * we want to find: [    LT  GE              LE  GT       ]
         * For LT and GE it's a classical lower_bound search.
         * Let's set up a range with left page's min_key < key and
         *  right page's min >= key; binary cut the range until it
         *  becomes of length 1 and then LT pos = left bound of the range
         *  and GE pos = right bound of the range.
         * For LE and GT it's a classical upper_bound search.
         * Let's set up a range with left page's min_key <= key and
         *  right page's min > key; binary cut the range until it
         *  becomes of length 1 and then LE pos = left bound of the range
         *  and GT pos = right bound of the range.
         */
        bool is_lower_bound = itype == ITER_LT || itype == ITER_GE;

        assert(run->info.page_count > 0);
        /* Initially the range is set with virtual positions */
        int32_t range[2] = { -1, run->info.page_count };
        assert(run->info.page_count > 0);
        do {
                int32_t mid = range[0] + (range[1] - range[0]) / 2;
                struct vy_page_info *info = vy_run_page_info(run, mid);
                int cmp = vy_entry_compare_with_raw_key(key, info->min_key,
                                                        info->min_key_hint,
                                                        cmp_def);
                if (is_lower_bound)
                        range[cmp <= 0] = mid;
                else
                        range[cmp < 0] = mid;
                *equal_key = *equal_key || cmp == 0;
        } while (range[1] - range[0] > 1);
        if (range[0] < 0)
                range[0] = run->info.page_count;
        uint32_t page = range[dir > 0];

        /**
         * Since page search uses only min_key of pages,
         *  for GE, GT and EQ the previous page can contain
         *  the point where iteration must be started.
         */
        if (page > 0 && dir > 0)
                return page - 1;
        return page;
}

struct vy_slice *
vy_slice_new(int64_t id, struct vy_run *run, struct vy_entry begin,
             struct vy_entry end, struct key_def *cmp_def)
{
        struct vy_slice *slice = malloc(sizeof(*slice));
        if (slice == NULL) {
                diag_set(OutOfMemory, sizeof(*slice),
                         "malloc", "struct vy_slice");
                return NULL;
        }
        memset(slice, 0, sizeof(*slice));
        slice->id = id;
        slice->run = run;
        slice->seed = rand();
        vy_run_ref(run);
        run->slice_count++;
        if (begin.stmt != NULL)
                tuple_ref(begin.stmt);
        slice->begin = begin;
        if (end.stmt != NULL)
                tuple_ref(end.stmt);
        slice->end = end;
        rlist_create(&slice->in_range);
        fiber_cond_create(&slice->pin_cond);
        if (run->info.page_count == 0) {
                /* The run is empty hence the slice is empty too. */
                return slice;
        }
        /** Lookup the first and the last pages spanned by the slice. */
        bool unused;
        if (slice->begin.stmt == NULL) {
                slice->first_page_no = 0;
        } else {
                slice->first_page_no =
                        vy_page_index_find_page(run, slice->begin, cmp_def,
                                                ITER_GE, &unused);
                assert(slice->first_page_no < run->info.page_count);
        }
        if (slice->end.stmt == NULL) {
                slice->last_page_no = run->info.page_count - 1;
        } else {
                slice->last_page_no =
                        vy_page_index_find_page(run, slice->end, cmp_def,
                                                ITER_LT, &unused);
                if (slice->last_page_no == run->info.page_count) {
                        /* It's an empty slice */
                        slice->first_page_no = 0;
                        slice->last_page_no = 0;
                        return slice;
                }
        }
        assert(slice->last_page_no >= slice->first_page_no);
        /** Estimate the number of statements in the slice. */
        uint32_t run_pages = run->info.page_count;
        uint32_t slice_pages = slice->last_page_no - slice->first_page_no + 1;
        slice->count.pages = slice_pages;
        slice->count.rows = DIV_ROUND_UP(run->count.rows *
                                         slice_pages, run_pages);
        slice->count.bytes = DIV_ROUND_UP(run->count.bytes *
                                          slice_pages, run_pages);
        slice->count.bytes_compressed = DIV_ROUND_UP(
                run->count.bytes_compressed * slice_pages, run_pages);
        return slice;
}

void
vy_slice_delete(struct vy_slice *slice)
{
        assert(slice->pin_count == 0);
        assert(slice->run->slice_count > 0);
        slice->run->slice_count--;
        vy_run_unref(slice->run);
        if (slice->begin.stmt != NULL)
                tuple_unref(slice->begin.stmt);
        if (slice->end.stmt != NULL)
                tuple_unref(slice->end.stmt);
        fiber_cond_destroy(&slice->pin_cond);
        TRASH(slice);
        free(slice);
}

int
vy_slice_cut(struct vy_slice *slice, int64_t id, struct vy_entry begin,
             struct vy_entry end, struct key_def *cmp_def,
             struct vy_slice **result)
{
        *result = NULL;

        if (begin.stmt != NULL && slice->end.stmt != NULL &&
            vy_entry_compare(begin, slice->end, cmp_def) >= 0)
                return 0; /* no intersection: begin >= slice->end */

        if (end.stmt != NULL && slice->begin.stmt != NULL &&
            vy_entry_compare(end, slice->begin, cmp_def) <= 0)
                return 0; /* no intersection: end <= slice->end */

        /* begin = MAX(begin, slice->begin) */
        if (slice->begin.stmt != NULL &&
            (begin.stmt == NULL || vy_entry_compare(begin, slice->begin,
                                                    cmp_def) < 0))
                begin = slice->begin;

        /* end = MIN(end, slice->end) */
        if (slice->end.stmt != NULL &&
            (end.stmt == NULL || vy_entry_compare(end, slice->end,
                                                  cmp_def) > 0))
                end = slice->end;

        *result = vy_slice_new(id, slice->run, begin, end, cmp_def);
        if (*result == NULL)
                return -1; /* OOM */

        return 0;
}

/**
 * Decode page information from xrow.
 *
 * @param[out] page Page information.
 * @param xrow      Xrow to decode.
 * @param cmp_def   Definition of keys stored in the page.
 * @param filename  Filename for error reporting.
 *
 * @retval  0 Success.
 * @retval -1 Error.
 */
static int
vy_page_info_decode(struct vy_page_info *page, const struct xrow_header *xrow,
                    struct key_def *cmp_def, const char *filename)
{
        assert(xrow->type == VY_INDEX_PAGE_INFO);
        const char *pos = xrow->body->iov_base;
        memset(page, 0, sizeof(*page));
        uint64_t key_map = vy_page_info_key_map;
        uint32_t map_size = mp_decode_map(&pos);
        uint32_t map_item;
        const char *key_beg;
        uint32_t part_count;
        for (map_item = 0; map_item < map_size; ++map_item) {
                uint32_t key = mp_decode_uint(&pos);
                key_map &= ~(1ULL << key);
                switch (key) {
                case VY_PAGE_INFO_OFFSET:
                        page->offset = mp_decode_uint(&pos);
                        break;
                case VY_PAGE_INFO_SIZE:
                        page->size = mp_decode_uint(&pos);
                        break;
                case VY_PAGE_INFO_ROW_COUNT:
                        page->row_count = mp_decode_uint(&pos);
                        break;
                case VY_PAGE_INFO_MIN_KEY:
                        key_beg = pos;
                        mp_next(&pos);
                        page->min_key = vy_key_dup(key_beg);
                        if (page->min_key == NULL)
                                return -1;
                        part_count = mp_decode_array(&key_beg);
                        page->min_key_hint = key_hint(key_beg, part_count,
                                                      cmp_def);
                        break;
                case VY_PAGE_INFO_UNPACKED_SIZE:
                        page->unpacked_size = mp_decode_uint(&pos);
                        break;
                case VY_PAGE_INFO_ROW_INDEX_OFFSET:
                        page->row_index_offset = mp_decode_uint(&pos);
                        break;
                default:
                        mp_next(&pos); /* unknown key, ignore */
                        break;
                }
        }
        if (key_map) {
                enum vy_page_info_key key = bit_ctz_u64(key_map);
                diag_set(ClientError, ER_INVALID_INDEX_FILE, filename,
                         tt_sprintf("Can't decode page info: "
                                    "missing mandatory key %s",
                                    vy_page_info_key_name(key)));
                return -1;
        }

        return 0;
}

/** Decode statement statistics from @data and advance @data. */
static void
vy_stmt_stat_decode(struct vy_stmt_stat *stat, const char **data)
{
        uint32_t size = mp_decode_map(data);
        for (uint32_t i = 0; i < size; i++) {
                uint64_t key = mp_decode_uint(data);
                uint64_t value = mp_decode_uint(data);
                switch (key) {
                case IPROTO_INSERT:
                        stat->inserts = value;
                        break;
                case IPROTO_REPLACE:
                        stat->replaces = value;
                        break;
                case IPROTO_DELETE:
                        stat->deletes = value;
                        break;
                case IPROTO_UPSERT:
                        stat->upserts = value;
                        break;
                default:
                        break;
                }
        }
}

/**
 * Decode the run metadata from xrow.
 *
 * @param xrow xrow to decode
 * @param[out] run_info the run information
 * @param filename File name for error reporting.
 *
 * @retval  0 success
 * @retval -1 error (check diag)
 */
int
vy_run_info_decode(struct vy_run_info *run_info,
                   const struct xrow_header *xrow,
                   const char *filename)
{
        assert(xrow->type == VY_INDEX_RUN_INFO);
        /* decode run */
        const char *pos = xrow->body->iov_base;
        memset(run_info, 0, sizeof(*run_info));
        uint64_t key_map = vy_run_info_key_map;
        uint32_t map_size = mp_decode_map(&pos);
        uint32_t map_item;
        const char *tmp;
        /* decode run values */
        for (map_item = 0; map_item < map_size; ++map_item) {
                uint32_t key = mp_decode_uint(&pos);
                key_map &= ~(1ULL << key);
                switch (key) {
                case VY_RUN_INFO_MIN_KEY:
                        tmp = pos;
                        mp_next(&pos);
                        run_info->min_key = vy_key_dup(tmp);
                        if (run_info->min_key == NULL)
                                return -1;
                        break;
                case VY_RUN_INFO_MAX_KEY:
                        tmp = pos;
                        mp_next(&pos);
                        run_info->max_key = vy_key_dup(tmp);
                        if (run_info->max_key == NULL)
                                return -1;
                        break;
                case VY_RUN_INFO_MIN_LSN:
                        run_info->min_lsn = mp_decode_uint(&pos);
                        break;
                case VY_RUN_INFO_MAX_LSN:
                        run_info->max_lsn = mp_decode_uint(&pos);
                        break;
                case VY_RUN_INFO_PAGE_COUNT:
                        run_info->page_count = mp_decode_uint(&pos);
                        break;
                case VY_RUN_INFO_BLOOM_FILTER_LEGACY:
                        run_info->bloom = tuple_bloom_decode_legacy(&pos);
                        if (run_info->bloom == NULL)
                                return -1;
                        break;
                case VY_RUN_INFO_BLOOM_FILTER:
                        run_info->bloom = tuple_bloom_decode(&pos);
                        if (run_info->bloom == NULL)
                                return -1;
                        break;
                case VY_RUN_INFO_STMT_STAT:
                        vy_stmt_stat_decode(&run_info->stmt_stat, &pos);
                        break;
                default:
                        mp_next(&pos); /* unknown key, ignore */
                        break;
                }
        }
        if (key_map) {
                enum vy_run_info_key key = bit_ctz_u64(key_map);
                diag_set(ClientError, ER_INVALID_INDEX_FILE, filename,
                         tt_sprintf("Can't decode run info: "
                                    "missing mandatory key %s",
                                    vy_run_info_key_name(key)));
                return -1;
        }
        return 0;
}

static struct vy_page *
vy_page_new(const struct vy_page_info *page_info)
{
        struct vy_page *page = malloc(sizeof(*page));
        if (page == NULL) {
                diag_set(OutOfMemory, sizeof(*page),
                         "load_page", "page cache");
                return NULL;
        }
        page->unpacked_size = page_info->unpacked_size;
        page->row_count = page_info->row_count;
        page->row_index = calloc(page_info->row_count, sizeof(uint32_t));
        if (page->row_index == NULL) {
                diag_set(OutOfMemory, page_info->row_count * sizeof(uint32_t),
                         "malloc", "page->row_index");
                free(page);
                return NULL;
        }

        page->data = (char *)malloc(page_info->unpacked_size);
        if (page->data == NULL) {
                diag_set(OutOfMemory, page_info->unpacked_size,
                         "malloc", "page->data");
                free(page->row_index);
                free(page);
                return NULL;
        }
        return page;
}

static void
vy_page_delete(struct vy_page *page)
{
        uint32_t *row_index = page->row_index;
        char *data = page->data;
#if !defined(NDEBUG)
        memset(row_index, '#', sizeof(uint32_t) * page->row_count);
        memset(data, '#', page->unpacked_size);
        memset(page, '#', sizeof(*page));
#endif /* !defined(NDEBUG) */
        free(row_index);
        free(data);
        free(page);
}

static int
vy_page_xrow(struct vy_page *page, uint32_t stmt_no,
             struct xrow_header *xrow)
{
        assert(stmt_no < page->row_count);
        const char *data = page->data + page->row_index[stmt_no];
        const char *data_end = stmt_no + 1 < page->row_count ?
                               page->data + page->row_index[stmt_no + 1] :
                               page->data + page->unpacked_size;
        return xrow_header_decode(xrow, &data, data_end, false);
}

/* {{{ vy_run_iterator vy_run_iterator support functions */

/**
 * Read raw stmt data from the page
 * @param page          Page.
 * @param stmt_no       Statement position in the page.
 * @param cmp_def       Definition of keys stored in the page.
 * @param format        Format for REPLACE/DELETE tuples.
 *
 * @retval not NULL Statement read from page.
 * @retval     NULL Memory error.
 */
static struct vy_entry
vy_page_stmt(struct vy_page *page, uint32_t stmt_no,
             struct key_def *cmp_def, struct tuple_format *format)
{
        struct xrow_header xrow;
        if (vy_page_xrow(page, stmt_no, &xrow) != 0)
                return vy_entry_none();
        struct vy_entry entry;
        entry.stmt = vy_stmt_decode(&xrow, format);
        if (entry.stmt == NULL)
                return vy_entry_none();
        entry.hint = vy_stmt_hint(entry.stmt, cmp_def);
        return entry;
}

/**
 * Binary search in page
 * In terms of STL, makes lower_bound for EQ,GE,LT and upper_bound for GT,LE
 * Additionally *equal_key argument is set to true if the found value is
 * equal to given key (set to false otherwise).
 * @retval position in the page
 */
static uint32_t
vy_page_find_key(struct vy_page *page, struct vy_entry key,
                 struct key_def *cmp_def, struct tuple_format *format,
                 enum iterator_type iterator_type, bool *equal_key)
{
        uint32_t beg = 0;
        uint32_t end = page->row_count;
        *equal_key = false;
        /* for upper bound we change zero comparison result to -1 */
        int zero_cmp = (iterator_type == ITER_GT ||
                        iterator_type == ITER_LE ? -1 : 0);
        while (beg != end) {
                uint32_t mid = beg + (end - beg) / 2;
                struct vy_entry fnd_key = vy_page_stmt(page, mid, cmp_def,
                                                       format);
                if (fnd_key.stmt == NULL)
                        return end;
                int cmp = vy_entry_compare(fnd_key, key, cmp_def);
                cmp = cmp ? cmp : zero_cmp;
                *equal_key = *equal_key || cmp == 0;
                if (cmp < 0)
                        beg = mid + 1;
                else
                        end = mid;
                tuple_unref(fnd_key.stmt);
        }
        return end;
}

/**
 * End iteration and free cached data.
 */
static void
vy_run_iterator_stop(struct vy_run_iterator *itr)
{
        if (itr->curr.stmt != NULL) {
                tuple_unref(itr->curr.stmt);
                itr->curr = vy_entry_none();
        }
        if (itr->curr_page != NULL) {
                vy_page_delete(itr->curr_page);
                if (itr->prev_page != NULL)
                        vy_page_delete(itr->prev_page);
                itr->curr_page = itr->prev_page = NULL;
        }
}

static int
vy_row_index_decode(uint32_t *row_index, uint32_t row_count,
                    struct xrow_header *xrow)
{
        assert(xrow->type == VY_RUN_ROW_INDEX);
        const char *pos = xrow->body->iov_base;
        uint32_t map_size = mp_decode_map(&pos);
        uint32_t map_item;
        uint32_t size = 0;
        for (map_item = 0; map_item < map_size; ++map_item) {
                uint32_t key = mp_decode_uint(&pos);
                switch (key) {
                case VY_ROW_INDEX_DATA:
                        size = mp_decode_binl(&pos);
                        break;
                }
        }
        if (size != sizeof(uint32_t) * row_count) {
                diag_set(ClientError, ER_INVALID_RUN_FILE,
                         tt_sprintf("Wrong row index size "
                                    "(expected %zu, got %u",
                                    sizeof(uint32_t) * row_count,
                                    (unsigned)size));
                return -1;
        }
        for (uint32_t i = 0; i < row_count; ++i) {
                row_index[i] = mp_load_u32(&pos);
        }
        assert(pos == xrow->body->iov_base + xrow->body->iov_len);
        return 0;
}

/** Return the name of a run data file. */
static inline const char *
vy_run_filename(struct vy_run *run)
{
        char *buf = tt_static_buf();
        vy_run_snprint_filename(buf, TT_STATIC_BUF_LEN, run->id, VY_FILE_RUN);
        return buf;
}

/**
 * Read a page requests from vinyl xlog data file.
 *
 * @retval 0 on success
 * @retval -1 on error, check diag
 */
static int
vy_page_read(struct vy_page *page, const struct vy_page_info *page_info,
             struct vy_run *run, ZSTD_DStream *zdctx)
{
        /* read xlog tx from xlog file */
        size_t region_svp = region_used(&fiber()->gc);
        char *data = (char *)region_alloc(&fiber()->gc, page_info->size);
        if (data == NULL) {
                diag_set(OutOfMemory, page_info->size, "region gc", "page");
                return -1;
        }
        ssize_t readen = fio_pread(run->fd, data, page_info->size,
                                   page_info->offset);
        ERROR_INJECT(ERRINJ_VYRUN_DATA_READ, {
                readen = -1;
                errno = EIO;});
        if (readen < 0) {
                diag_set(SystemError, "failed to read from file");
                goto error;
        }
        if (readen != (ssize_t)page_info->size) {
                diag_set(ClientError, ER_INVALID_RUN_FILE,
                         "Unexpected end of file");
                goto error;
        }

        struct errinj *inj = errinj(ERRINJ_VY_READ_PAGE_TIMEOUT, ERRINJ_DOUBLE);
        if (inj != NULL && inj->dparam > 0)
                thread_sleep(inj->dparam);

        ERROR_INJECT_SLEEP(ERRINJ_VY_READ_PAGE_DELAY);

        /* decode xlog tx */
        const char *data_pos = data;
        const char *data_end = data + readen;
        char *rows = page->data;
        char *rows_end = rows + page_info->unpacked_size;
        if (xlog_tx_decode(data, data_end, rows, rows_end, zdctx) != 0)
                goto error;

        struct xrow_header xrow;
        data_pos = page->data + page_info->row_index_offset;
        data_end = page->data + page_info->unpacked_size;
        if (xrow_header_decode(&xrow, &data_pos, data_end, true) == -1)
                goto error;
        if (xrow.type != VY_RUN_ROW_INDEX) {
                diag_set(ClientError, ER_INVALID_RUN_FILE,
                         tt_sprintf("Wrong row index type "
                                    "(expected %d, got %u)",
                                    VY_RUN_ROW_INDEX, (unsigned)xrow.type));
                goto error;
        }
        if (vy_row_index_decode(page->row_index, page->row_count, &xrow) != 0)
                goto error;
        region_truncate(&fiber()->gc, region_svp);
        ERROR_INJECT(ERRINJ_VY_READ_PAGE, {
                diag_set(ClientError, ER_INJECTION, "vinyl page read");
                return -1;});
        return 0;
error:
        region_truncate(&fiber()->gc, region_svp);
        diag_log();
        say_error("error reading %s@%llu:%u", vy_run_filename(run),
                  (unsigned long long)page_info->offset,
                  (unsigned)page_info->size);
        return -1;
}

/**
 * Get thread local zstd decompression context
 */
static ZSTD_DStream *
vy_env_get_zdctx(struct vy_run_env *env)
{
        ZSTD_DStream *zdctx = tt_pthread_getspecific(env->zdctx_key);
        if (zdctx == NULL) {
                zdctx = ZSTD_createDStream();
                if (zdctx == NULL) {
                        diag_set(OutOfMemory, sizeof(zdctx), "malloc",
                                 "zstd context");
                        return NULL;
                }
                tt_pthread_setspecific(env->zdctx_key, zdctx);
        }
        return zdctx;
}

/**
 * vinyl read task callback
 */
static int
vy_page_read_cb(struct cbus_call_msg *base)
{
        struct vy_page_read_task *task = (struct vy_page_read_task *)base;
        ZSTD_DStream *zdctx = vy_env_get_zdctx(task->run->env);
        if (zdctx == NULL)
                return -1;
        if (vy_page_read(task->page, task->page_info, task->run, zdctx) != 0)
                return -1;
        if (task->key.stmt != NULL) {
                task->pos_in_page = vy_page_find_key(task->page, task->key,
                                                     task->cmp_def, task->format,
                                                     task->iterator_type,
                                                     &task->equal_found);
        }
        return 0;
}

/**
 * Read a page from disk given its number.
 * The function caches two most recently read pages.
 *
 * @retval 0 success
 * @retval -1 critical error
 */
static NODISCARD int
vy_run_iterator_load_page(struct vy_run_iterator *itr, uint32_t page_no,
                          struct vy_entry key, enum iterator_type iterator_type,
                          struct vy_page **result, uint32_t *pos_in_page,
                          bool *equal_found)
{
        struct vy_slice *slice = itr->slice;
        struct vy_run_env *env = slice->run->env;

        /* Check cache */
        struct vy_page *page = NULL;
        if (itr->curr_page != NULL &&
            itr->curr_page->page_no == page_no) {
                page = itr->curr_page;
        } else if (itr->prev_page != NULL &&
                   itr->prev_page->page_no == page_no) {
                SWAP(itr->prev_page, itr->curr_page);
                page = itr->curr_page;
        }
        if (page != NULL) {
                if (key.stmt != NULL)
                        *pos_in_page = vy_page_find_key(page, key, itr->cmp_def,
                                                        itr->format, iterator_type,
                                                        equal_found);
                *result = page;
                return 0;
        }

        /* Allocate buffers */
        struct vy_page_info *page_info = vy_run_page_info(slice->run, page_no);
        page = vy_page_new(page_info);
        if (page == NULL)
                return -1;

        /* Read page data from the disk */
        struct vy_page_read_task *task = mempool_alloc(&env->read_task_pool);
        if (task == NULL) {
                diag_set(OutOfMemory, sizeof(*task),
                         "mempool", "vy_page_read_task");
                vy_page_delete(page);
                return -1;
        }
        task->run = slice->run;
        task->page_info = page_info;
        task->page = page;
        task->key = key;
        task->iterator_type = iterator_type;
        task->cmp_def = itr->cmp_def;
        task->format = itr->format;
        task->pos_in_page = 0;
        task->equal_found = false;

        int rc = vy_run_env_coio_call(env, &task->base, vy_page_read_cb);

        *pos_in_page = task->pos_in_page;
        *equal_found = task->equal_found;

        mempool_free(&env->read_task_pool, task);
        if (rc != 0) {
                vy_page_delete(page);
                return -1;
        }

        /* Update cache */
        if (itr->prev_page != NULL)
                vy_page_delete(itr->prev_page);
        itr->prev_page = itr->curr_page;
        itr->curr_page = page;
        page->page_no = page_no;

        /* Update read statistics. */
        itr->stat->read.rows += page_info->row_count;
        itr->stat->read.bytes += page_info->unpacked_size;
        itr->stat->read.bytes_compressed += page_info->size;
        itr->stat->read.pages++;

        *result = page;
        return 0;
}

/**
 * Read key and lsn by a given wide position.
 * For the first record in a page reads the result from the page
 * index instead of fetching it from disk.
 *
 * @retval 0 success
 * @retval -1 read error or out of memory.
 */
static NODISCARD int
vy_run_iterator_read(struct vy_run_iterator *itr,
                     struct vy_run_iterator_pos pos,
                     struct vy_entry *ret)
{
        struct vy_page *page;
        bool equal_found;
        uint32_t pos_in_page;
        int rc = vy_run_iterator_load_page(itr, pos.page_no, vy_entry_none(),
                                           ITER_GE, &page, &pos_in_page,
                                           &equal_found);
        if (rc != 0)
                return rc;
        *ret = vy_page_stmt(page, pos.pos_in_page, itr->cmp_def, itr->format);
        if (ret->stmt == NULL)
                return -1;
        return 0;
}

/**
 * Binary search in a run for the given key.
 * In terms of STL, makes lower_bound for EQ,GE,LT and upper_bound for GT,LE
 * Resulting wide position is stored it *pos argument
 * Additionally *equal_key argument is set to true if the found value is
 * equal to given key (untouched otherwise)
 *
 * @retval 0 success
 * @retval 1 EOF
 * @retval -1 read or memory error
 */
static NODISCARD int
vy_run_iterator_search(struct vy_run_iterator *itr,
                       enum iterator_type iterator_type, struct vy_entry key,
                       struct vy_run_iterator_pos *pos, bool *equal_key)
{
        pos->page_no = vy_page_index_find_page(itr->slice->run, key,
                                               itr->cmp_def, iterator_type,
                                               equal_key);
        if (pos->page_no == itr->slice->run->info.page_count)
                return 1;
        bool equal_in_page;
        struct vy_page *page;
        int rc = vy_run_iterator_load_page(itr, pos->page_no, key,
                                           iterator_type, &page,
                                           &pos->pos_in_page, &equal_in_page);
        if (rc != 0)
                return rc;
        if (pos->pos_in_page == page->row_count) {
                pos->page_no++;
                pos->pos_in_page = 0;
        } else {
                *equal_key = equal_in_page;
        }
        return 0;
}

/**
 * Increment (or decrement, depending on the order) the current
 * wide position.
 * @retval 0 success, set *pos to new value
 * @retval 1 EOF
 * Affects: curr_loaded_page
 */
static NODISCARD int
vy_run_iterator_next_pos(struct vy_run_iterator *itr,
                         enum iterator_type iterator_type,
                         struct vy_run_iterator_pos *pos)
{
        struct vy_run *run = itr->slice->run;
        *pos = itr->curr_pos;
        if (iterator_type == ITER_LE || iterator_type == ITER_LT) {
                assert(pos->page_no <= run->info.page_count);
                if (pos->pos_in_page > 0) {
                        pos->pos_in_page--;
                } else {
                        if (pos->page_no == 0)
                                return 1;
                        pos->page_no--;
                        struct vy_page_info *page_info =
                                vy_run_page_info(run, pos->page_no);
                        assert(page_info->row_count > 0);
                        pos->pos_in_page = page_info->row_count - 1;
                }
        } else {
                assert(iterator_type == ITER_GE || iterator_type == ITER_GT ||
                       iterator_type == ITER_EQ);
                assert(pos->page_no < run->info.page_count);
                struct vy_page_info *page_info =
                        vy_run_page_info(run, pos->page_no);
                assert(page_info->row_count > 0);
                pos->pos_in_page++;
                if (pos->pos_in_page >= page_info->row_count) {
                        pos->page_no++;
                        pos->pos_in_page = 0;
                        if (pos->page_no == run->info.page_count)
                                return 1;
                }
        }
        return 0;
}

/**
 * Find the next record with lsn <= itr->lsn record.
 * The current position must be at the beginning of a series of
 * records with the same key it terms of direction of iterator
 * (i.e. left for GE, right for LE).
 * @retval 0 success or EOF (*ret == NULL)
 * @retval -1 read or memory error
 * Affects: curr_loaded_page, curr_pos
 */
static NODISCARD int
vy_run_iterator_find_lsn(struct vy_run_iterator *itr, struct vy_entry *ret)
{
        struct vy_slice *slice = itr->slice;
        struct key_def *cmp_def = itr->cmp_def;

        *ret = vy_entry_none();

        assert(itr->search_started);
        assert(itr->curr.stmt != NULL);
        assert(itr->curr_pos.page_no < slice->run->info.page_count);

        while (vy_stmt_lsn(itr->curr.stmt) > (**itr->read_view).vlsn ||
               vy_stmt_flags(itr->curr.stmt) & VY_STMT_SKIP_READ) {
                if (vy_run_iterator_next_pos(itr, itr->iterator_type,
                                             &itr->curr_pos) != 0) {
                        vy_run_iterator_stop(itr);
                        return 0;
                }
                tuple_unref(itr->curr.stmt);
                itr->curr = vy_entry_none();
                if (vy_run_iterator_read(itr, itr->curr_pos, &itr->curr) != 0)
                        return -1;
                if (itr->iterator_type == ITER_EQ &&
                    vy_entry_compare(itr->curr, itr->key, cmp_def) != 0) {
                        vy_run_iterator_stop(itr);
                        return 0;
                }
        }
        if (itr->iterator_type == ITER_LE || itr->iterator_type == ITER_LT) {
                struct vy_run_iterator_pos test_pos;
                while (vy_run_iterator_next_pos(itr, itr->iterator_type,
                                                &test_pos) == 0) {
                        struct vy_entry test;
                        if (vy_run_iterator_read(itr, test_pos, &test) != 0)
                                return -1;
                        if (vy_stmt_lsn(test.stmt) > (**itr->read_view).vlsn ||
                            vy_stmt_flags(test.stmt) & VY_STMT_SKIP_READ ||
                            vy_entry_compare(itr->curr, test, cmp_def) != 0) {
                                tuple_unref(test.stmt);
                                break;
                        }
                        tuple_unref(itr->curr.stmt);
                        itr->curr = test;
                        itr->curr_pos = test_pos;
                }
        }
        /* Check if the result is within the slice boundaries. */
        if (itr->iterator_type == ITER_LE || itr->iterator_type == ITER_LT) {
                if (slice->begin.stmt != NULL &&
                    vy_entry_compare(itr->curr, slice->begin, cmp_def) < 0) {
                        vy_run_iterator_stop(itr);
                        return 0;
                }
        } else {
                assert(itr->iterator_type == ITER_GE ||
                       itr->iterator_type == ITER_GT ||
                       itr->iterator_type == ITER_EQ);
                if (slice->end.stmt != NULL &&
                    vy_entry_compare(itr->curr, slice->end, cmp_def) >= 0) {
                        vy_run_iterator_stop(itr);
                        return 0;
                }
        }
        vy_stmt_counter_acct_tuple(&itr->stat->get, itr->curr.stmt);
        *ret = itr->curr;
        return 0;
}

/**
 * Helper function for vy_run_iterator_seek().
 *
 * Positions the iterator to the beginning (i.e. leftmost for GE,
 * rightmost for LE) of a series of statements matching the given
 * search criteria.
 *
 * Updates itr->curr_pos. Doesn't affect itr->curr.
 *
 * @retval 0 success
 * @retval 1 EOF
 * @retval -1 read or memory error
 */
static NODISCARD int
vy_run_iterator_do_seek(struct vy_run_iterator *itr,
                        enum iterator_type iterator_type, struct vy_entry key)
{
        struct vy_run *run = itr->slice->run;
        struct vy_run_iterator_pos end_pos = {run->info.page_count, 0};
        bool equal_found = false;
        if (!vy_stmt_is_empty_key(key.stmt)) {
                int rc = vy_run_iterator_search(itr, iterator_type, key,
                                                &itr->curr_pos, &equal_found);
                if (rc != 0)
                        return rc;
        } else if (iterator_type == ITER_LE) {
                itr->curr_pos = end_pos;
        } else {
                assert(iterator_type == ITER_GE);
                itr->curr_pos.page_no = 0;
                itr->curr_pos.pos_in_page = 0;
        }
        if (iterator_type == ITER_EQ && !equal_found)
                return 1;
        if ((iterator_type == ITER_GE || iterator_type == ITER_GT) &&
            itr->curr_pos.page_no == end_pos.page_no)
                return 1;
        if (iterator_type == ITER_LT || iterator_type == ITER_LE) {
                /**
                 * 1) in case of ITER_LT we now positioned on the value >= than
                 * given, so we need to make a step on previous key
                 * 2) in case if ITER_LE we now positioned on the value > than
                 * given (special branch of code in vy_run_iterator_search),
                 * so we need to make a step on previous key
                 */
                return vy_run_iterator_next_pos(itr, iterator_type,
                                                &itr->curr_pos);
        } else {
                assert(iterator_type == ITER_GE || iterator_type == ITER_GT ||
                       iterator_type == ITER_EQ);
                /**
                 * 1) in case of ITER_GT we now positioned on the value > than
                 * given (special branch of code in vy_run_iterator_search),
                 * so we need just to find proper lsn
                 * 2) in case if ITER_GE or ITER_EQ we now positioned on the
                 * value >= given, so we need just to find proper lsn
                 */
                return 0;
        }
}

/**
 * Position the iterator to the first statement satisfying
 * the iterator search criteria and following the given key
 * (pass NULL to start iteration).
 */
static NODISCARD int
vy_run_iterator_seek(struct vy_run_iterator *itr, struct vy_entry last,
                     struct vy_entry *ret)
{
        struct key_def *cmp_def = itr->cmp_def;
        struct vy_slice *slice = itr->slice;
        struct tuple_bloom *bloom = slice->run->info.bloom;
        struct vy_entry key = itr->key;
        enum iterator_type iterator_type = itr->iterator_type;

        *ret = vy_entry_none();
        assert(itr->search_started);

        /* Check the bloom filter on the first iteration. */
        bool check_bloom = (itr->iterator_type == ITER_EQ &&
                            itr->curr.stmt == NULL && bloom != NULL);
        if (check_bloom && !vy_bloom_maybe_has(bloom, itr->key, itr->key_def)) {
                vy_run_iterator_stop(itr);
                itr->stat->bloom_hit++;
                return 0;
        }

        /*
         * vy_run_iterator_do_seek() implements its own EQ check.
         * We only need to check EQ here if iterator type and key
         * passed to it differ from the original.
         */
        bool check_eq = false;

        /*
         * Modify iterator type and key so as to position it to
         * the first statement following the given key.
         */
        if (last.stmt != NULL) {
                if (iterator_type == ITER_EQ)
                        check_eq = true;
                iterator_type = iterator_direction(iterator_type) > 0 ?
                                ITER_GT : ITER_LT;
                key = last;
        }

        /* Take slice boundaries into account. */
        if (slice->begin.stmt != NULL &&
            (iterator_type == ITER_GT || iterator_type == ITER_GE ||
             iterator_type == ITER_EQ)) {
                /*
                 *    original   |     start
                 * --------------+-------+-----+
                 *   KEY   | DIR |  KEY  | DIR |
                 * --------+-----+-------+-----+
                 * > begin | *   | key   | *   |
                 * = begin | gt  | key   | gt  |
                 *         | ge  | begin | ge  |
                 *         | eq  | begin | ge  |
                 * < begin | gt  | begin | ge  |
                 *         | ge  | begin | ge  |
                 *         | eq  |    stop     |
                 */
                int cmp = vy_entry_compare(key, slice->begin, cmp_def);
                if (cmp < 0 && iterator_type == ITER_EQ) {
                        vy_run_iterator_stop(itr);
                        return 0;
                }
                if (cmp < 0 || (cmp == 0 && iterator_type != ITER_GT)) {
                        if (iterator_type == ITER_EQ)
                                check_eq = true;
                        iterator_type = ITER_GE;
                        key = slice->begin;
                }
        }
        if (slice->end.stmt != NULL &&
            (iterator_type == ITER_LT || iterator_type == ITER_LE)) {
                /*
                 *    original   |     start
                 * --------------+-------+-----+
                 *   KEY   | DIR |  KEY  | DIR |
                 * --------+-----+-------+-----+
                 * < end   | *   | key   | *   |
                 * = end   | lt  | key   | lt  |
                 *         | le  | end   | lt  |
                 * > end   | lt  | end   | lt  |
                 *         | le  | end   | lt  |
                 */
                int cmp = vy_entry_compare(key, slice->end, cmp_def);
                if (cmp > 0 || (cmp == 0 && iterator_type != ITER_LT)) {
                        iterator_type = ITER_LT;
                        key = slice->end;
                }
        }

        /* Perform a lookup in the run. */
        itr->stat->lookup++;
        int rc = vy_run_iterator_do_seek(itr, iterator_type, key);
        if (rc < 0)
                return -1;
        if (rc > 0)
                goto not_found;

        /* Load the found statement. */
        if (itr->curr.stmt != NULL) {
                tuple_unref(itr->curr.stmt);
                itr->curr = vy_entry_none();
        }
        if (vy_run_iterator_read(itr, itr->curr_pos, &itr->curr) != 0)
                return -1;

        /* Check EQ constraint if necessary. */
        if (check_eq && vy_entry_compare(itr->curr, itr->key,
                                         itr->cmp_def) != 0)
                goto not_found;

        /* Skip statements invisible from the iterator read view. */
        return vy_run_iterator_find_lsn(itr, ret);

not_found:
        if (check_bloom)
                itr->stat->bloom_miss++;
        vy_run_iterator_stop(itr);
        return 0;
}

/* }}} vy_run_iterator vy_run_iterator support functions */

/* {{{ vy_run_iterator API implementation */

void
vy_run_iterator_open(struct vy_run_iterator *itr,
                     struct vy_run_iterator_stat *stat,
                     struct vy_slice *slice, enum iterator_type iterator_type,
                     struct vy_entry key, const struct vy_read_view **rv,
                     struct key_def *cmp_def, struct key_def *key_def,
                     struct tuple_format *format)
{
        itr->stat = stat;
        itr->cmp_def = cmp_def;
        itr->key_def = key_def;
        itr->format = format;
        itr->slice = slice;

        itr->iterator_type = iterator_type;
        itr->key = key;
        itr->read_view = rv;

        itr->curr = vy_entry_none();
        itr->curr_pos.page_no = slice->run->info.page_count;
        itr->curr_page = NULL;
        itr->prev_page = NULL;
        itr->search_started = false;

        /*
         * Make sure the format we use to create tuples won't
         * go away if DDL is called while the iterator is used.
         *
         * XXX: Please remove this kludge when proper DDL locking
         * is implemented on transaction management level or multi
         * version data dictionary is in place.
         */
        tuple_format_ref(format);
}

/**
 * Advance a run iterator to the newest statement for the next key.
 * The statement is returned in @ret (NULL if EOF).
 * Returns 0 on success, -1 on memory allocation or IO error.
 */
static NODISCARD int
vy_run_iterator_next_key(struct vy_run_iterator *itr, struct vy_entry *ret)
{
        *ret = vy_entry_none();

        if (!itr->search_started) {
                itr->search_started = true;
                return vy_run_iterator_seek(itr, vy_entry_none(), ret);
        }
        if (itr->curr.stmt == NULL)
                return 0;

        assert(itr->curr_pos.page_no < itr->slice->run->info.page_count);

        struct vy_entry next = vy_entry_none();
        do {
                if (next.stmt != NULL)
                        tuple_unref(next.stmt);
                if (vy_run_iterator_next_pos(itr, itr->iterator_type,
                                             &itr->curr_pos) != 0) {
                        vy_run_iterator_stop(itr);
                        return 0;
                }

                if (vy_run_iterator_read(itr, itr->curr_pos, &next) != 0)
                        return -1;
        } while (vy_entry_compare(itr->curr, next, itr->cmp_def) == 0);

        tuple_unref(itr->curr.stmt);
        itr->curr = next;

        if (itr->iterator_type == ITER_EQ &&
            vy_entry_compare(next, itr->key, itr->cmp_def) != 0) {
                vy_run_iterator_stop(itr);
                return 0;
        }
        return vy_run_iterator_find_lsn(itr, ret);
}

/**
 * Advance a run iterator to the next (older) statement for the
 * current key. The statement is returned in @ret (NULL if EOF).
 * Returns 0 on success, -1 on memory allocation or IO error.
 */
static NODISCARD int
vy_run_iterator_next_lsn(struct vy_run_iterator *itr, struct vy_entry *ret)
{
        *ret = vy_entry_none();

        assert(itr->search_started);
        assert(itr->curr.stmt != NULL);
        assert(itr->curr_pos.page_no < itr->slice->run->info.page_count);

        struct vy_run_iterator_pos next_pos;
next:
        if (vy_run_iterator_next_pos(itr, ITER_GE, &next_pos) != 0)
                return 0;

        struct vy_entry next;
        if (vy_run_iterator_read(itr, next_pos, &next) != 0)
                return -1;

        if (vy_entry_compare(itr->curr, next, itr->cmp_def) != 0) {
                tuple_unref(next.stmt);
                return 0;
        }

        tuple_unref(itr->curr.stmt);
        itr->curr = next;
        itr->curr_pos = next_pos;
        if (vy_stmt_flags(itr->curr.stmt) & VY_STMT_SKIP_READ)
                goto next;

        vy_stmt_counter_acct_tuple(&itr->stat->get, itr->curr.stmt);
        *ret = itr->curr;
        return 0;
}

NODISCARD int
vy_run_iterator_next(struct vy_run_iterator *itr,
                     struct vy_history *history)
{
        vy_history_cleanup(history);
        struct vy_entry entry;
        if (vy_run_iterator_next_key(itr, &entry) != 0)
                return -1;
        while (entry.stmt != NULL) {
                if (vy_history_append_stmt(history, entry) != 0)
                        return -1;
                if (vy_history_is_terminal(history))
                        break;
                if (vy_run_iterator_next_lsn(itr, &entry) != 0)
                        return -1;
        }
        return 0;
}

NODISCARD int
vy_run_iterator_skip(struct vy_run_iterator *itr, struct vy_entry last,
                     struct vy_history *history)
{
        /*
         * Check if the iterator is already positioned
         * at the statement following last.
         */
        if (itr->search_started &&
            (itr->curr.stmt == NULL || last.stmt == NULL ||
             iterator_direction(itr->iterator_type) *
             vy_entry_compare(itr->curr, last, itr->cmp_def) > 0))
                return 0;

        vy_history_cleanup(history);

        itr->search_started = true;
        struct vy_entry entry;
        if (vy_run_iterator_seek(itr, last, &entry) != 0)
                return -1;

        while (entry.stmt != NULL) {
                if (vy_history_append_stmt(history, entry) != 0)
                        return -1;
                if (vy_history_is_terminal(history))
                        break;
                if (vy_run_iterator_next_lsn(itr, &entry) != 0)
                        return -1;
        }
        return 0;
}

void
vy_run_iterator_close(struct vy_run_iterator *itr)
{
        vy_run_iterator_stop(itr);
        tuple_format_unref(itr->format);
        TRASH(itr);
}

/* }}} vy_run_iterator API implementation */

/** Account a page to run statistics. */
static void
vy_run_acct_page(struct vy_run *run, struct vy_page_info *page)
{
        const char *min_key_end = page->min_key;
        mp_next(&min_key_end);
        run->page_index_size += sizeof(struct vy_page_info);
        run->page_index_size += min_key_end - page->min_key;
        run->count.rows += page->row_count;
        run->count.bytes += page->unpacked_size;
        run->count.bytes_compressed += page->size;
        run->count.pages++;
}

int
vy_run_recover(struct vy_run *run, const char *dir,
               uint32_t space_id, uint32_t iid, struct key_def *cmp_def)
{
        char path[PATH_MAX];
        vy_run_snprint_path(path, sizeof(path), dir,
                            space_id, iid, run->id, VY_FILE_INDEX);

        struct xlog_cursor cursor;
        ERROR_INJECT_COUNTDOWN(ERRINJ_VY_RUN_OPEN, {
                diag_set(SystemError, "failed to open '%s' file", path);
                goto fail;
        });
        if (xlog_cursor_open(&cursor, path))
                goto fail;

        struct xlog_meta *meta = &cursor.meta;
        if (strcmp(meta->filetype, XLOG_META_TYPE_INDEX) != 0) {
                diag_set(ClientError, ER_INVALID_XLOG_TYPE,
                         XLOG_META_TYPE_INDEX, meta->filetype);
                goto fail_close;
        }

        /* Read run header. */
        struct xrow_header xrow;
        /* all rows should be in one tx */
        int rc = xlog_cursor_next_tx(&cursor);

        if (rc != 0) {
                if (rc > 0)
                        diag_set(ClientError, ER_INVALID_INDEX_FILE,
                                 path, "Unexpected end of file");
                goto fail_close;
        }
        rc = xlog_cursor_next_row(&cursor, &xrow);
        if (rc != 0) {
                if (rc > 0)
                        diag_set(ClientError, ER_INVALID_INDEX_FILE,
                                 path, "Unexpected end of file");
                goto fail_close;
        }

        if (xrow.type != VY_INDEX_RUN_INFO) {
                diag_set(ClientError, ER_INVALID_INDEX_FILE, path,
                         tt_sprintf("Wrong xrow type (expected %d, got %u)",
                                    VY_INDEX_RUN_INFO, (unsigned)xrow.type));
                goto fail_close;
        }

        if (vy_run_info_decode(&run->info, &xrow, path) != 0)
                goto fail_close;

        /* Allocate buffer for page info. */
        run->page_info = calloc(run->info.page_count,
                                      sizeof(struct vy_page_info));
        if (run->page_info == NULL) {
                diag_set(OutOfMemory,
                         run->info.page_count * sizeof(struct vy_page_info),
                         "malloc", "struct vy_page_info");
                goto fail_close;
        }

        for (uint32_t page_no = 0; page_no < run->info.page_count; page_no++) {
                int rc = xlog_cursor_next_row(&cursor, &xrow);
                if (rc != 0) {
                        if (rc > 0) {
                                /** To few pages in file */
                                diag_set(ClientError, ER_INVALID_INDEX_FILE,
                                         path, "Unexpected end of file");
                        }
                        /*
                         * Limit the count of pages to
                         * successfully created pages.
                         */
                        run->info.page_count = page_no;
                        goto fail_close;
                }
                if (xrow.type != VY_INDEX_PAGE_INFO) {
                        diag_set(ClientError, ER_INVALID_INDEX_FILE,
                                 tt_sprintf("Wrong xrow type "
                                            "(expected %d, got %u)",
                                            VY_INDEX_PAGE_INFO,
                                            (unsigned)xrow.type));
                        goto fail_close;
                }
                struct vy_page_info *page = run->page_info + page_no;
                if (vy_page_info_decode(page, &xrow, cmp_def, path) < 0) {
                        /**
                         * Limit the count of pages to successfully
                         * created pages
                         */
                        run->info.page_count = page_no;
                        goto fail_close;
                }
                vy_run_acct_page(run, page);
        }

        /* We don't need to keep metadata file open any longer. */
        xlog_cursor_close(&cursor, false);

        /* Prepare data file for reading. */
        vy_run_snprint_path(path, sizeof(path), dir,
                            space_id, iid, run->id, VY_FILE_RUN);
        if (xlog_cursor_open(&cursor, path))
                goto fail;
        meta = &cursor.meta;
        if (strcmp(meta->filetype, XLOG_META_TYPE_RUN) != 0) {
                diag_set(ClientError, ER_INVALID_XLOG_TYPE,
                         XLOG_META_TYPE_RUN, meta->filetype);
                goto fail_close;
        }
        run->fd = cursor.fd;
        xlog_cursor_close(&cursor, true);
        return 0;

fail_close:
        xlog_cursor_close(&cursor, false);
fail:
        vy_run_clear(run);
        diag_log();
        say_error("failed to load `%s'", path);
        return -1;
}

/* dump statement to the run page buffers (stmt header and data) */
static int
vy_run_dump_stmt(struct vy_entry entry, struct xlog *data_xlog,
                 struct vy_page_info *info, struct key_def *key_def,
                 bool is_primary)
{
        struct xrow_header xrow;
        int rc = (is_primary ?
                  vy_stmt_encode_primary(entry.stmt, key_def, 0, &xrow) :
                  vy_stmt_encode_secondary(entry.stmt, key_def,
                                           vy_entry_multikey_idx(entry, key_def),
                                           &xrow));
        if (rc != 0)
                return -1;

        ssize_t row_size;
        if ((row_size = xlog_write_row(data_xlog, &xrow)) < 0)
                return -1;

        info->unpacked_size += row_size;
        info->row_count++;
        return 0;
}

/**
 * Encode uint32_t array of row offsets (row index) as xrow
 *
 * @param row_index row index
 * @param row_count size of row index
 * @param[out] xrow xrow to fill.
 * @retval 0 for success
 * @retval -1 for error
 */
static int
vy_row_index_encode(const uint32_t *row_index, uint32_t row_count,
                    struct xrow_header *xrow)
{
        memset(xrow, 0, sizeof(*xrow));
        xrow->type = VY_RUN_ROW_INDEX;

        size_t size = mp_sizeof_map(1) +
                      mp_sizeof_uint(VY_ROW_INDEX_DATA) +
                      mp_sizeof_bin(sizeof(uint32_t) * row_count);
        char *pos = region_alloc(&fiber()->gc, size);
        if (pos == NULL) {
                diag_set(OutOfMemory, size, "region", "row index");
                return -1;
        }
        xrow->body->iov_base = pos;
        pos = mp_encode_map(pos, 1);
        pos = mp_encode_uint(pos, VY_ROW_INDEX_DATA);
        pos = mp_encode_binl(pos, sizeof(uint32_t) * row_count);
        for (uint32_t i = 0; i < row_count; ++i)
                pos = mp_store_u32(pos, row_index[i]);
        xrow->body->iov_len = (void *)pos - xrow->body->iov_base;
        assert(xrow->body->iov_len == size);
        xrow->bodycnt = 1;
        return 0;
}

/**
 * Helper to extend run page info array
 */
static inline int
vy_run_alloc_page_info(struct vy_run *run, uint32_t *page_info_capacity)
{
        uint32_t cap = *page_info_capacity > 0 ?
                       *page_info_capacity * 2 : 16;
        struct vy_page_info *page_info = realloc(run->page_info,
                                        cap * sizeof(*page_info));
        if (page_info == NULL) {
                diag_set(OutOfMemory, cap * sizeof(*page_info),
                         "realloc", "struct vy_page_info");
                return -1;
        }
        run->page_info = page_info;
        *page_info_capacity = cap;
        return 0;
}

/** {{{ vy_page_info */

/**
 * Encode vy_page_info as xrow.
 * Allocates using region_alloc.
 *
 * @param page_info page information to encode
 * @param[out] xrow xrow to fill
 *
 * @retval  0 success
 * @retval -1 error, check diag
 */
static int
vy_page_info_encode(const struct vy_page_info *page_info,
                    struct xrow_header *xrow)
{
        struct region *region = &fiber()->gc;

        uint32_t min_key_size;
        const char *tmp = page_info->min_key;
        assert(mp_typeof(*tmp) == MP_ARRAY);
        mp_next(&tmp);
        min_key_size = tmp - page_info->min_key;

        /* calc tuple size */
        uint32_t size;
        /* 3 items: page offset, size, and map */
        size = mp_sizeof_map(6) +
               mp_sizeof_uint(VY_PAGE_INFO_OFFSET) +
               mp_sizeof_uint(page_info->offset) +
               mp_sizeof_uint(VY_PAGE_INFO_SIZE) +
               mp_sizeof_uint(page_info->size) +
               mp_sizeof_uint(VY_PAGE_INFO_ROW_COUNT) +
               mp_sizeof_uint(page_info->row_count) +
               mp_sizeof_uint(VY_PAGE_INFO_MIN_KEY) +
               min_key_size +
               mp_sizeof_uint(VY_PAGE_INFO_UNPACKED_SIZE) +
               mp_sizeof_uint(page_info->unpacked_size) +
               mp_sizeof_uint(VY_PAGE_INFO_ROW_INDEX_OFFSET) +
               mp_sizeof_uint(page_info->row_index_offset);

        char *pos = region_alloc(region, size);
        if (pos == NULL) {
                diag_set(OutOfMemory, size, "region", "page encode");
                return -1;
        }

        memset(xrow, 0, sizeof(*xrow));
        /* encode page */
        xrow->body->iov_base = pos;
        pos = mp_encode_map(pos, 6);
        pos = mp_encode_uint(pos, VY_PAGE_INFO_OFFSET);
        pos = mp_encode_uint(pos, page_info->offset);
        pos = mp_encode_uint(pos, VY_PAGE_INFO_SIZE);
        pos = mp_encode_uint(pos, page_info->size);
        pos = mp_encode_uint(pos, VY_PAGE_INFO_ROW_COUNT);
        pos = mp_encode_uint(pos, page_info->row_count);
        pos = mp_encode_uint(pos, VY_PAGE_INFO_MIN_KEY);
        memcpy(pos, page_info->min_key, min_key_size);
        pos += min_key_size;
        pos = mp_encode_uint(pos, VY_PAGE_INFO_UNPACKED_SIZE);
        pos = mp_encode_uint(pos, page_info->unpacked_size);
        pos = mp_encode_uint(pos, VY_PAGE_INFO_ROW_INDEX_OFFSET);
        pos = mp_encode_uint(pos, page_info->row_index_offset);
        xrow->body->iov_len = (void *)pos - xrow->body->iov_base;
        xrow->bodycnt = 1;

        xrow->type = VY_INDEX_PAGE_INFO;
        return 0;
}

/** vy_page_info }}} */

/** {{{ vy_run_info */

/** Return the size of encoded statement statistics. */
static size_t
vy_stmt_stat_sizeof(const struct vy_stmt_stat *stat)
{
        return mp_sizeof_map(4) +
                mp_sizeof_uint(IPROTO_INSERT) +
                mp_sizeof_uint(IPROTO_REPLACE) +
                mp_sizeof_uint(IPROTO_DELETE) +
                mp_sizeof_uint(IPROTO_UPSERT) +
                mp_sizeof_uint(stat->inserts) +
                mp_sizeof_uint(stat->replaces) +
                mp_sizeof_uint(stat->deletes) +
                mp_sizeof_uint(stat->upserts);
}

/** Encode statement statistics to @buf and return advanced @buf. */
static char *
vy_stmt_stat_encode(const struct vy_stmt_stat *stat, char *buf)
{
        buf = mp_encode_map(buf, 4);
        buf = mp_encode_uint(buf, IPROTO_INSERT);
        buf = mp_encode_uint(buf, stat->inserts);
        buf = mp_encode_uint(buf, IPROTO_REPLACE);
        buf = mp_encode_uint(buf, stat->replaces);
        buf = mp_encode_uint(buf, IPROTO_DELETE);
        buf = mp_encode_uint(buf, stat->deletes);
        buf = mp_encode_uint(buf, IPROTO_UPSERT);
        buf = mp_encode_uint(buf, stat->upserts);
        return buf;
}

/**
 * Encode vy_run_info as xrow
 * Allocates using region alloc
 *
 * @param run_info the run information
 * @param xrow xrow to fill.
 *
 * @retval  0 success
 * @retval -1 on error, check diag
 */
static int
vy_run_info_encode(const struct vy_run_info *run_info,
                   struct xrow_header *xrow)
{
        const char *tmp;
        tmp = run_info->min_key;
        mp_next(&tmp);
        size_t min_key_size = tmp - run_info->min_key;
        tmp = run_info->max_key;
        mp_next(&tmp);
        size_t max_key_size = tmp - run_info->max_key;

        uint32_t key_count = 6;
        if (run_info->bloom != NULL)
                key_count++;

        size_t size = mp_sizeof_map(key_count);
        size += mp_sizeof_uint(VY_RUN_INFO_MIN_KEY) + min_key_size;
        size += mp_sizeof_uint(VY_RUN_INFO_MAX_KEY) + max_key_size;
        size += mp_sizeof_uint(VY_RUN_INFO_MIN_LSN) +
                mp_sizeof_uint(run_info->min_lsn);
        size += mp_sizeof_uint(VY_RUN_INFO_MAX_LSN) +
                mp_sizeof_uint(run_info->max_lsn);
        size += mp_sizeof_uint(VY_RUN_INFO_PAGE_COUNT) +
                mp_sizeof_uint(run_info->page_count);
        if (run_info->bloom != NULL)
                size += mp_sizeof_uint(VY_RUN_INFO_BLOOM_FILTER) +
                        tuple_bloom_size(run_info->bloom);
        size += mp_sizeof_uint(VY_RUN_INFO_STMT_STAT) +
                vy_stmt_stat_sizeof(&run_info->stmt_stat);

        char *pos = region_alloc(&fiber()->gc, size);
        if (pos == NULL) {
                diag_set(OutOfMemory, size, "region", "run encode");
                return -1;
        }
        memset(xrow, 0, sizeof(*xrow));
        xrow->body->iov_base = pos;
        /* encode values */
        pos = mp_encode_map(pos, key_count);
        pos = mp_encode_uint(pos, VY_RUN_INFO_MIN_KEY);
        memcpy(pos, run_info->min_key, min_key_size);
        pos += min_key_size;
        pos = mp_encode_uint(pos, VY_RUN_INFO_MAX_KEY);
        memcpy(pos, run_info->max_key, max_key_size);
        pos += max_key_size;
        pos = mp_encode_uint(pos, VY_RUN_INFO_MIN_LSN);
        pos = mp_encode_uint(pos, run_info->min_lsn);
        pos = mp_encode_uint(pos, VY_RUN_INFO_MAX_LSN);
        pos = mp_encode_uint(pos, run_info->max_lsn);
        pos = mp_encode_uint(pos, VY_RUN_INFO_PAGE_COUNT);
        pos = mp_encode_uint(pos, run_info->page_count);
        if (run_info->bloom != NULL) {
                pos = mp_encode_uint(pos, VY_RUN_INFO_BLOOM_FILTER);
                pos = tuple_bloom_encode(run_info->bloom, pos);
        }
        pos = mp_encode_uint(pos, VY_RUN_INFO_STMT_STAT);
        pos = vy_stmt_stat_encode(&run_info->stmt_stat, pos);
        xrow->body->iov_len = (void *)pos - xrow->body->iov_base;
        xrow->bodycnt = 1;
        xrow->type = VY_INDEX_RUN_INFO;
        return 0;
}

/* vy_run_info }}} */

/**
 * Write run index to file.
 */
static int
vy_run_write_index(struct vy_run *run, const char *dirpath,
                   uint32_t space_id, uint32_t iid)
{
        char path[PATH_MAX];
        vy_run_snprint_path(path, sizeof(path), dirpath,
                            space_id, iid, run->id, VY_FILE_INDEX);

        say_info("writing `%s'", path);

        struct xlog index_xlog;
        struct xlog_meta meta;
        xlog_meta_create(&meta, XLOG_META_TYPE_INDEX, &INSTANCE_UUID,
                         NULL, NULL);
        struct xlog_opts opts = xlog_opts_default;
        opts.rate_limit = run->env->snap_io_rate_limit;
        opts.sync_interval = VY_RUN_SYNC_INTERVAL;
        if (xlog_create(&index_xlog, path, 0, &meta, &opts) < 0)
                return -1;

        xlog_tx_begin(&index_xlog);
        struct region *region = &fiber()->gc;
        size_t mem_used = region_used(region);

        struct xrow_header xrow;
        if (vy_run_info_encode(&run->info, &xrow) != 0 ||
            xlog_write_row(&index_xlog, &xrow) < 0)
                goto fail_rollback;

        for (uint32_t page_no = 0; page_no < run->info.page_count; ++page_no) {
                struct vy_page_info *page_info = vy_run_page_info(run, page_no);
                if (vy_page_info_encode(page_info, &xrow) < 0) {
                        goto fail_rollback;
                }
                if (xlog_write_row(&index_xlog, &xrow) < 0)
                        goto fail_rollback;
        }

        region_truncate(region, mem_used);
        if (xlog_tx_commit(&index_xlog) < 0)
                goto fail;

        ERROR_INJECT(ERRINJ_VY_INDEX_FILE_RENAME, {
                diag_set(ClientError, ER_INJECTION, "vinyl index file rename");
                xlog_close(&index_xlog, false);
                return -1;
        });

        if (xlog_flush(&index_xlog) < 0 ||
            xlog_rename(&index_xlog) < 0)
                goto fail;

        xlog_close(&index_xlog, false);
        return 0;

fail_rollback:
        region_truncate(region, mem_used);
        xlog_tx_rollback(&index_xlog);
fail:
        xlog_close(&index_xlog, false);
        unlink(path);
        return -1;
}

int
vy_run_writer_create(struct vy_run_writer *writer, struct vy_run *run,
                     const char *dirpath, uint32_t space_id, uint32_t iid,
                     struct key_def *cmp_def, struct key_def *key_def,
                     uint64_t page_size, double bloom_fpr, bool no_compression)
{
        memset(writer, 0, sizeof(*writer));
        writer->run = run;
        writer->dirpath = dirpath;
        writer->space_id = space_id;
        writer->iid = iid;
        writer->cmp_def = cmp_def;
        writer->key_def = key_def;
        writer->page_size = page_size;
        writer->bloom_fpr = bloom_fpr;
        writer->no_compression = no_compression;
        if (bloom_fpr < 1) {
                writer->bloom = tuple_bloom_builder_new(key_def->part_count);
                if (writer->bloom == NULL)
                        return -1;
        }
        xlog_clear(&writer->data_xlog);
        ibuf_create(&writer->row_index_buf, &cord()->slabc,
                    4096 * sizeof(uint32_t));
        run->info.min_lsn = INT64_MAX;
        run->info.max_lsn = -1;
        assert(run->page_info == NULL);
        return 0;
}

/**
 * Create an xlog to write run.
 * @param writer Run writer.
 * @retval -1 Memory or IO error.
 * @retval  0 Success.
 */
static int
vy_run_writer_create_xlog(struct vy_run_writer *writer)
{
        assert(!xlog_is_open(&writer->data_xlog));
        char path[PATH_MAX];
        vy_run_snprint_path(path, sizeof(path), writer->dirpath,
                            writer->space_id, writer->iid, writer->run->id,
                            VY_FILE_RUN);
        say_info("writing `%s'", path);
        struct xlog_meta meta;
        xlog_meta_create(&meta, XLOG_META_TYPE_RUN, &INSTANCE_UUID,
                         NULL, NULL);
        struct xlog_opts opts = xlog_opts_default;
        opts.rate_limit = writer->run->env->snap_io_rate_limit;
        opts.sync_interval = VY_RUN_SYNC_INTERVAL;
        opts.no_compression = writer->no_compression;
        if (xlog_create(&writer->data_xlog, path, 0, &meta, &opts) != 0)
                return -1;
        return 0;
}

/**
 * Start a new page with a min_key stored in @a first_entry.
 * @param writer Run writer.
 * @param first_entry First statement of a page.
 *
 * @retval -1 Memory error.
 * @retval  0 Success.
 */
static int
vy_run_writer_start_page(struct vy_run_writer *writer,
                         struct vy_entry first_entry)
{
        struct vy_run *run = writer->run;
        if (run->info.page_count >= writer->page_info_capacity &&
            vy_run_alloc_page_info(run, &writer->page_info_capacity) != 0)
                return -1;
        const char *key = vy_stmt_is_key(first_entry.stmt) ?
                          tuple_data(first_entry.stmt) :
                          tuple_extract_key(first_entry.stmt, writer->cmp_def,
                                            vy_entry_multikey_idx(first_entry,
                                                                  writer->cmp_def),
                                            NULL);
        if (key == NULL)
                return -1;
        if (run->info.page_count == 0) {
                assert(run->info.min_key == NULL);
                run->info.min_key = vy_key_dup(key);
                if (run->info.min_key == NULL)
                        return -1;
        }
        struct vy_page_info *page = run->page_info + run->info.page_count;
        if (vy_page_info_create(page, writer->data_xlog.offset,
                                key, writer->cmp_def) != 0)
                return -1;
        xlog_tx_begin(&writer->data_xlog);
        return 0;
}

/**
 * Write @a stmt into a current page.
 * @param writer Run writer.
 * @param entry Statement to write.
 *
 * @retval -1 Memory or IO error.
 * @retval  0 Success.
 */
static int
vy_run_writer_write_to_page(struct vy_run_writer *writer, struct vy_entry entry)
{
        if (writer->bloom != NULL &&
            vy_bloom_builder_add(writer->bloom, entry, writer->key_def) != 0)
                return -1;
        if (writer->last.stmt != NULL)
                vy_stmt_unref_if_possible(writer->last.stmt);
        writer->last = entry;
        vy_stmt_ref_if_possible(entry.stmt);
        struct vy_run *run = writer->run;
        struct vy_page_info *page = run->page_info + run->info.page_count;
        uint32_t *offset = (uint32_t *)ibuf_alloc(&writer->row_index_buf,
                                                  sizeof(uint32_t));
        if (offset == NULL) {
                diag_set(OutOfMemory, sizeof(uint32_t), "ibuf", "row index");
                return -1;
        }
        *offset = page->unpacked_size;
        if (vy_run_dump_stmt(entry, &writer->data_xlog, page,
                             writer->cmp_def, writer->iid == 0) != 0)
                return -1;
        int64_t lsn = vy_stmt_lsn(entry.stmt);
        run->info.min_lsn = MIN(run->info.min_lsn, lsn);
        run->info.max_lsn = MAX(run->info.max_lsn, lsn);
        vy_stmt_stat_acct(&run->info.stmt_stat, vy_stmt_type(entry.stmt));
        return 0;
}

/**
 * Finish a current page.
 * @param writer Run writer.
 * @retval -1 Memory or IO error.
 * @retval  0 Success.
 */
static int
vy_run_writer_end_page(struct vy_run_writer *writer)
{
        struct vy_run *run = writer->run;
        struct vy_page_info *page = run->page_info + run->info.page_count;

        assert(page->row_count > 0);
        assert(ibuf_used(&writer->row_index_buf) ==
               sizeof(uint32_t) * page->row_count);

        struct xrow_header xrow;
        uint32_t *row_index = (uint32_t *)writer->row_index_buf.rpos;
        if (vy_row_index_encode(row_index, page->row_count, &xrow) < 0)
                return -1;
        ssize_t written = xlog_write_row(&writer->data_xlog, &xrow);
        if (written < 0)
                return -1;
        page->row_index_offset = page->unpacked_size;
        page->unpacked_size += written;

        written = xlog_tx_commit(&writer->data_xlog);
        if (written == 0)
                written = xlog_flush(&writer->data_xlog);
        if (written < 0)
                return -1;
        page->size = written;
        run->info.page_count++;
        vy_run_acct_page(run, page);
        ibuf_reset(&writer->row_index_buf);
        return 0;
}

int
vy_run_writer_append_stmt(struct vy_run_writer *writer, struct vy_entry entry)
{
        int rc = -1;
        size_t region_svp = region_used(&fiber()->gc);
        if (!xlog_is_open(&writer->data_xlog) &&
            vy_run_writer_create_xlog(writer) != 0)
                goto out;
        if (ibuf_used(&writer->row_index_buf) == 0 &&
            vy_run_writer_start_page(writer, entry) != 0)
                goto out;
        if (vy_run_writer_write_to_page(writer, entry) != 0)
                goto out;
        if (obuf_size(&writer->data_xlog.obuf) >= writer->page_size &&
            vy_run_writer_end_page(writer) != 0)
                goto out;
        rc = 0;
out:
        region_truncate(&fiber()->gc, region_svp);
        return rc;
}

/**
 * Destroy a run writer.
 * @param writer Writer to destroy.
 * @param reuse_fd True in a case of success run write. And else
 *        false.
 */
static void
vy_run_writer_destroy(struct vy_run_writer *writer, bool reuse_fd)
{
        if (writer->last.stmt != NULL)
                vy_stmt_unref_if_possible(writer->last.stmt);
        if (xlog_is_open(&writer->data_xlog))
                xlog_close(&writer->data_xlog, reuse_fd);
        if (writer->bloom != NULL)
                tuple_bloom_builder_delete(writer->bloom);
        ibuf_destroy(&writer->row_index_buf);
}

int
vy_run_writer_commit(struct vy_run_writer *writer)
{
        int rc = -1;
        size_t region_svp = region_used(&fiber()->gc);

        if (ibuf_used(&writer->row_index_buf) != 0 &&
            vy_run_writer_end_page(writer) != 0)
                goto out;

        struct vy_run *run = writer->run;
        if (vy_run_is_empty(run)) {
                vy_run_writer_destroy(writer, false);
                rc = 0;
                goto out;
        }

        assert(writer->last.stmt != NULL);
        const char *key = vy_stmt_is_key(writer->last.stmt) ?
                          tuple_data(writer->last.stmt) :
                          tuple_extract_key(writer->last.stmt, writer->cmp_def,
                                            vy_entry_multikey_idx(writer->last,
                                                                  writer->cmp_def),
                                            NULL);
        if (key == NULL)
                goto out;

        assert(run->info.max_key == NULL);
        run->info.max_key = vy_key_dup(key);
        if (run->info.max_key == NULL)
                goto out;

        ERROR_INJECT(ERRINJ_VY_RUN_FILE_RENAME, {
                diag_set(ClientError, ER_INJECTION, "vinyl run file rename");
                goto out;
        });

        /* Sync data and link the file to the final name. */
        if (xlog_sync(&writer->data_xlog) < 0 ||
            xlog_rename(&writer->data_xlog) < 0)
                goto out;

        if (writer->bloom != NULL) {
                run->info.bloom = tuple_bloom_new(writer->bloom,
                                                  writer->bloom_fpr);
                if (run->info.bloom == NULL)
                        goto out;
        }
        if (vy_run_write_index(run, writer->dirpath,
                               writer->space_id, writer->iid) != 0)
                goto out;

        run->fd = writer->data_xlog.fd;
        vy_run_writer_destroy(writer, true);
        rc = 0;
out:
        region_truncate(&fiber()->gc, region_svp);
        return rc;
}

void
vy_run_writer_abort(struct vy_run_writer *writer)
{
        vy_run_writer_destroy(writer, false);
}

int
vy_run_rebuild_index(struct vy_run *run, const char *dir,
                     uint32_t space_id, uint32_t iid,
                     struct key_def *cmp_def, struct key_def *key_def,
                     struct tuple_format *format, const struct index_opts *opts)
{
        assert(run->info.bloom == NULL);
        assert(run->page_info == NULL);
        struct region *region = &fiber()->gc;
        size_t mem_used = region_used(region);

        struct xlog_cursor cursor;
        char path[PATH_MAX];
        vy_run_snprint_path(path, sizeof(path), dir,
                            space_id, iid, run->id, VY_FILE_RUN);

        say_info("rebuilding index for `%s'", path);
        if (xlog_cursor_open(&cursor, path))
                return -1;

        int rc = 0;
        uint32_t page_info_capacity = 0;

        const char *key = NULL;
        int64_t max_lsn = 0;
        int64_t min_lsn = INT64_MAX;
        struct tuple *prev_tuple = NULL;
        char *page_min_key = NULL;

        struct tuple_bloom_builder *bloom_builder = NULL;
        if (opts->bloom_fpr < 1) {
                bloom_builder = tuple_bloom_builder_new(key_def->part_count);
                if (bloom_builder == NULL)
                        goto close_err;
        }

        off_t page_offset, next_page_offset = xlog_cursor_pos(&cursor);
        while ((rc = xlog_cursor_next_tx(&cursor)) == 0) {
                region_truncate(region, mem_used);
                page_offset = next_page_offset;
                next_page_offset = xlog_cursor_pos(&cursor);

                if (run->info.page_count == page_info_capacity &&
                    vy_run_alloc_page_info(run, &page_info_capacity) != 0)
                        goto close_err;
                uint32_t page_row_count = 0;
                uint64_t page_row_index_offset = 0;
                uint64_t row_offset = xlog_cursor_tx_pos(&cursor);

                struct xrow_header xrow;
                while ((rc = xlog_cursor_next_row(&cursor, &xrow)) == 0) {
                        if (xrow.type == VY_RUN_ROW_INDEX) {
                                page_row_index_offset = row_offset;
                                row_offset = xlog_cursor_tx_pos(&cursor);
                                continue;
                        }
                        ++page_row_count;
                        struct tuple *tuple = vy_stmt_decode(&xrow, format);
                        if (tuple == NULL)
                                goto close_err;
                        if (bloom_builder != NULL) {
                                struct vy_entry entry = {tuple, HINT_NONE};
                                if (vy_bloom_builder_add(bloom_builder, entry,
                                                         key_def) != 0) {
                                        tuple_unref(tuple);
                                        goto close_err;
                                }
                        }
                        key = vy_stmt_is_key(tuple) ? tuple_data(tuple) :
                              tuple_extract_key(tuple, cmp_def,
                                                MULTIKEY_NONE, NULL);
                        if (prev_tuple != NULL)
                                tuple_unref(prev_tuple);
                        prev_tuple = tuple;
                        if (key == NULL)
                                goto close_err;
                        if (run->info.min_key == NULL) {
                                run->info.min_key = vy_key_dup(key);
                                if (run->info.min_key == NULL)
                                        goto close_err;
                        }
                        if (page_min_key == NULL) {
                                page_min_key = vy_key_dup(key);
                                if (page_min_key == NULL)
                                        goto close_err;
                        }
                        if (xrow.lsn > max_lsn)
                                max_lsn = xrow.lsn;
                        if (xrow.lsn < min_lsn)
                                min_lsn = xrow.lsn;
                        row_offset = xlog_cursor_tx_pos(&cursor);
                }
                struct vy_page_info *info;
                info = run->page_info + run->info.page_count;
                if (vy_page_info_create(info, page_offset,
                                        page_min_key, cmp_def) != 0)
                        goto close_err;
                info->row_count = page_row_count;
                info->size = next_page_offset - page_offset;
                info->unpacked_size = xlog_cursor_tx_pos(&cursor);
                info->row_index_offset = page_row_index_offset;
                ++run->info.page_count;
                vy_run_acct_page(run, info);

                free(page_min_key);
                page_min_key = NULL;
        }

        if (key != NULL) {
                run->info.max_key = vy_key_dup(key);
                if (run->info.max_key == NULL)
                        goto close_err;
        }
        run->info.max_lsn = max_lsn;
        run->info.min_lsn = min_lsn;

        if (prev_tuple != NULL) {
                tuple_unref(prev_tuple);
                prev_tuple = NULL;
        }
        region_truncate(region, mem_used);
        run->fd = cursor.fd;
        xlog_cursor_close(&cursor, true);

        if (bloom_builder != NULL) {
                run->info.bloom = tuple_bloom_new(bloom_builder,
                                                  opts->bloom_fpr);
                if (run->info.bloom == NULL)
                        goto close_err;
                tuple_bloom_builder_delete(bloom_builder);
                bloom_builder = NULL;
        }

        /* New run index is ready for write, unlink old file if exists */
        vy_run_snprint_path(path, sizeof(path), dir,
                            space_id, iid, run->id, VY_FILE_INDEX);
        if (unlink(path) < 0 && errno != ENOENT) {
                diag_set(SystemError, "failed to unlink file '%s'",
                         path);
                goto close_err;
        }
        if (vy_run_write_index(run, dir, space_id, iid) != 0)
                goto close_err;
        return 0;
close_err:
        vy_run_clear(run);
        region_truncate(region, mem_used);
        if (prev_tuple != NULL)
                tuple_unref(prev_tuple);
        if (page_min_key != NULL)
                free(page_min_key);
        if (bloom_builder != NULL)
                tuple_bloom_builder_delete(bloom_builder);
        if (xlog_cursor_is_open(&cursor))
                xlog_cursor_close(&cursor, false);
        return -1;
}

/**
 * Try to remove an empty directory.
 *
 * @param path Path to the directory.
 * @return int 0 on success, -1 on error.
 */
static int
try_rmdir(const char *path)
{
        int rc = 0;
        if (coio_rmdir(path) < 0) {
                if (errno != ENOENT) {
                        if (errno != ENOTEMPTY)
                                say_syserror("error while removing %s", path);
                        rc = -1;
                }
        } else {
                say_info("removed %s", path);
        }
        return rc;
}

int
vy_run_remove_files(const char *dir, uint32_t space_id,
                    uint32_t iid, int64_t run_id)
{
        ERROR_INJECT(ERRINJ_VY_GC,
                     {say_error("error injection: vinyl run %lld not deleted",
                                (long long)run_id); return -1;});
        int ret = 0;
        char path[PATH_MAX];
        for (int type = 0; type < vy_file_MAX; type++) {
                vy_run_snprint_path(path, sizeof(path), dir,
                                    space_id, iid, run_id, type);
                if (coio_unlink(path) < 0) {
                        if (errno != ENOENT) {
                                say_syserror("error while removing %s", path);
                                ret = -1;
                        }
                } else
                        say_info("removed %s", path);
        }
        /* Remove the root directory if it's empty. */
        vy_lsm_snprint_path(path, sizeof(path), dir, space_id, iid);
        if (try_rmdir(path) < 0)
                return ret;
        vy_space_snprint_path(path, sizeof(path), dir, space_id);
        try_rmdir(path);
        return ret;
}

/**
 * Read a page with stream->page_no from the run and save it in stream->page.
 * Support function of slice stream.
 * @param stream - the stream.
 * @return 0 on success, -1 of memory or read error (diag is set).
 */
static NODISCARD int
vy_slice_stream_read_page(struct vy_slice_stream *stream)
{
        struct vy_run *run = stream->slice->run;

        assert(stream->page == NULL);
        ZSTD_DStream *zdctx = vy_env_get_zdctx(run->env);
        if (zdctx == NULL)
                return -1;

        struct vy_page_info *page_info = vy_run_page_info(run, stream->page_no);
        stream->page = vy_page_new(page_info);
        if (stream->page == NULL)
                return -1;

        if (vy_page_read(stream->page, page_info, run, zdctx) != 0) {
                vy_page_delete(stream->page);
                stream->page = NULL;
                return -1;
        }
        return 0;
}

/**
 * Binary search in a run for the given key. Find the first position with
 * a tuple greater or equal to slice
 * @retval 0 success
 * @retval -1 read or memory error
 */
static NODISCARD int
vy_slice_stream_search(struct vy_stmt_stream *virt_stream)
{
        assert(virt_stream->iface->start == vy_slice_stream_search);
        struct vy_slice_stream *stream = (struct vy_slice_stream *)virt_stream;
        assert(stream->page == NULL);
        if (stream->slice->begin.stmt == NULL) {
                /* Already at the beginning */
                assert(stream->page_no == 0);
                assert(stream->pos_in_page == 0);
                return 0;
        }

        if (vy_slice_stream_read_page(stream) != 0)
                return -1;

        bool unused;
        stream->pos_in_page = vy_page_find_key(stream->page,
                                               stream->slice->begin,
                                               stream->cmp_def,
                                               stream->format,
                                               ITER_GE, &unused);

        if (stream->pos_in_page == stream->page->row_count) {
                /* The first tuple is in the beginning of the next page */
                vy_page_delete(stream->page);
                stream->page = NULL;
                stream->page_no++;
                stream->pos_in_page = 0;
        }
        return 0;
}

/**
 * Get the value from the stream and move to the next position.
 * Set *ret to the value or NULL if EOF.
 * @param virt_stream - virtual stream.
 * @param ret - pointer to the result.
 * @return 0 on success, -1 on memory or read error.
 */
static NODISCARD int
vy_slice_stream_next(struct vy_stmt_stream *virt_stream, struct vy_entry *ret)
{
        assert(virt_stream->iface->next == vy_slice_stream_next);
        struct vy_slice_stream *stream = (struct vy_slice_stream *)virt_stream;
        *ret = vy_entry_none();

        /* If the slice is ended, return EOF */
        if (stream->page_no > stream->slice->last_page_no)
                return 0;

        /* If current page is not already read, read it */
        if (stream->page == NULL && vy_slice_stream_read_page(stream) != 0)
                return -1;

        /* Read current tuple from the page */
        struct vy_entry entry = vy_page_stmt(stream->page, stream->pos_in_page,
                                             stream->cmp_def, stream->format);
        if (entry.stmt == NULL) /* Read or memory error */
                return -1;

        /* Check that the tuple is not out of slice bounds = */
        if (stream->slice->end.stmt != NULL &&
            stream->page_no >= stream->slice->last_page_no &&
            vy_entry_compare(entry, stream->slice->end, stream->cmp_def) >= 0) {
                tuple_unref(entry.stmt);
                return 0;
        }

        /* We definitely has the next non-null tuple. Save it in stream */
        if (stream->entry.stmt != NULL)
                tuple_unref(stream->entry.stmt);
        stream->entry = entry;
        *ret = entry;

        /* Increment position */
        stream->pos_in_page++;

        /* Check whether the position is out of page */
        struct vy_page_info *page_info = vy_run_page_info(stream->slice->run,
                                                          stream->page_no);
        if (stream->pos_in_page >= page_info->row_count) {
                /**
                 * Out of page. Free page, move the position to the next page
                 * and * nullify page pointer to read it on the next iteration.
                 */
                vy_page_delete(stream->page);
                stream->page = NULL;
                stream->page_no++;
                stream->pos_in_page = 0;
        }

        return 0;
}

/**
 * Free resources.
 */
static void
vy_slice_stream_stop(struct vy_stmt_stream *virt_stream)
{
        assert(virt_stream->iface->stop == vy_slice_stream_stop);
        struct vy_slice_stream *stream = (struct vy_slice_stream *)virt_stream;
        if (stream->page != NULL) {
                vy_page_delete(stream->page);
                stream->page = NULL;
        }
        if (stream->entry.stmt != NULL) {
                tuple_unref(stream->entry.stmt);
                stream->entry = vy_entry_none();
        }
}

static void
vy_slice_stream_close(struct vy_stmt_stream *virt_stream)
{
        assert(virt_stream->iface->close == vy_slice_stream_close);
        struct vy_slice_stream *stream = (struct vy_slice_stream *)virt_stream;
        tuple_format_unref(stream->format);
}

static const struct vy_stmt_stream_iface vy_slice_stream_iface = {
        .start = vy_slice_stream_search,
        .next = vy_slice_stream_next,
        .stop = vy_slice_stream_stop,
        .close = vy_slice_stream_close,
};

void
vy_slice_stream_open(struct vy_slice_stream *stream, struct vy_slice *slice,
                     struct key_def *cmp_def, struct tuple_format *format)
{
        stream->base.iface = &vy_slice_stream_iface;

        stream->page_no = slice->first_page_no;
        stream->pos_in_page = 0; /* We'll find it later */
        stream->page = NULL;
        stream->entry = vy_entry_none();

        stream->slice = slice;
        stream->cmp_def = cmp_def;
        stream->format = format;
        tuple_format_ref(format);
}

tarantool / tarantool / 10418752118

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous