12397458527

Committed 18 Dec 2024 04:44PM UTC coverage: 87.367%. Remained the same

Build # 12397458527

Build Type

push

github

Committed by

Totktonada

Commit Message

config: fix on_event roles callback config argument

The config object was passed to the `on_event` callback instead of the
corresponding roles config before. Lets fix it, implementing the
intended behavior.

NO_DOC=bugfix

Closes #10934

(cherry picked from commit d5d2291ae)

Run Details

69673 of 123524 branches covered (56.4%)

102617 of 117455 relevant lines covered (87.37%)

2871961.47 hits per line

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

87.53

/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_task.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];
                cbus_stop_loop(&reader->reader_pipe);
                cpipe_destroy(&reader->reader_pipe);
        }
        for (int i = 0; i < env->reader_pool_size; i++) {
                struct vy_run_reader *reader = &env->reader_pool[i];
                if (cord_join(&reader->cord) != 0)
                        panic_syserror("failed to join vinyl reader thread");
        }
        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_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).
 */
static int
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, uint32_t *pos,
                 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 -1;
                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);
        }
        *pos = end;
        return 0;
}

/**
 * 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_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 &&
            vy_page_find_key(task->page, task->key, task->cmp_def,
                             task->format, task->iterator_type,
                             &task->pos_in_page, &task->equal_found) != 0)
                return -1;
        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 &&
                    vy_page_find_key(page, key, itr->cmp_def,
                                     itr->format, iterator_type,
                                     pos_in_page, equal_found) != 0)
                        return -1;
                *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_RECOVER_COUNTDOWN, {
                diag_set(ClientError, ER_INJECTION, "vinyl run recover");
                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");
                goto fail;
        });

        if (xlog_close(&index_xlog) != 0 ||
            xlog_materialize(&index_xlog) != 0)
                goto fail;

        return 0;

fail_rollback:
        region_truncate(region, mem_used);
        xlog_tx_rollback(&index_xlog);
fail:
        xlog_discard(&index_xlog);
        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;
        run->info.page_count++;
        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 - 1;
        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;
        assert(run->info.page_count > 0);
        struct vy_page_info *page = run->page_info + run->info.page_count - 1;

        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;
        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.
 */
static void
vy_run_writer_destroy(struct vy_run_writer *writer)
{
        if (writer->last.stmt != NULL)
                vy_stmt_unref_if_possible(writer->last.stmt);
        if (xlog_is_open(&writer->data_xlog))
                xlog_discard(&writer->data_xlog);
        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);
                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_close_reuse_fd(&writer->data_xlog, &run->fd) != 0 ||
            xlog_materialize(&writer->data_xlog) != 0) {
                xlog_discard(&writer->data_xlog);
                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;

        vy_run_writer_destroy(writer);
        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);
}

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);
        xlog_remove_file(path, 0);
        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 (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;
}

static ssize_t
vy_run_remove_files_f(va_list ap)
{
        const char *dir = va_arg(ap, typeof(dir));
        uint32_t space_id = va_arg(ap, typeof(space_id));
        uint32_t iid = va_arg(ap, typeof(iid));
        int64_t run_id = va_arg(ap, typeof(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 (!xlog_remove_file(path, XLOG_RM_VERBOSE))
                        ret = -1;
        }
        /* 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;
}

int
vy_run_remove_files(const char *dir, uint32_t space_id,
                    uint32_t iid, int64_t run_id)
{
        return coio_call(vy_run_remove_files_f, dir, space_id, iid, run_id);
}

/**
 * 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;
        if (vy_page_find_key(stream->page, stream->slice->begin,
                             stream->cmp_def, stream->format, ITER_GE,
                             &stream->pos_in_page, &unused) != 0)
                return -1;

        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;

        /* If the slice is ended, return EOF */
        if (stream->page_no > stream->slice->last_page_no) {
                *ret = vy_entry_none();
                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);
                *ret = vy_entry_none();
                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 / 12397458527

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