15365

pending completion

Build # 15365

Build Type

push

travis-ci

Committed by

locker

Commit Message

<a href="https://github.com/tarantool/tarantool/commit/<a class=hub.com/tarantool/tarantool/commit/96532cdd130371aa5fde82f330fffcf63fbe5529">96532cdd1<a href="https://github.com/tarantool/tarantool/commit/96532cdd130371aa5fde82f330fffcf63fbe5529">&quot;&gt;vinyl: randomize range compaction to avoid IO load spikes

Since all ranges constituting an LSM tree have the same configuration,
they tend to get compacted at approximately the same time. This entails
IO load spikes, which, in turn, lead to deviation of the LSM tree from
the target shape and hence increased read amplification. To prevent this
from happening, this patch implements compaction randomization: with 10%
probability we defer compaction at each LSM tree level, i.e. if the
number of runs at a level exceeds the configured run_count_per_level,
the level will be compacted with 90%-probability, but with 10%
probability it won&amp;#39;t - compaction will be deferred until another run
is added to the level.

Our simulations show that such a simple algorithm performs fairly well:
it randomizes compaction pace among ranges, spreading IO load evenly in
time, while the write amplification is increased by not more than 5-10%,
which seems to be a reasonable price for elimination of IO load spikes.

Closes #3944

(cherry picked from commit </a><a class="double-link" href="https://github.com/tarantool/tarantool/commit/<a class="double-link" href="https://github.com/tarantool/tarantool/commit/c9e7baedf4f478cd333087f08525c0a4ad29db82">c9e7baedf</a>">c9e7baedf</a><a href="https://github.com/tarantool/tarantool/commit/96532cdd130371aa5fde82f330fffcf63fbe5529">)

Run Details

5 of 5 new or added lines in 2 files covered. (100.0%)

41154 of 50055 relevant lines covered (82.22%)

836232.72 hits per line

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

91.42

/src/box/alter.cc

/*
 * Copyright 2010-2016, 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 <COPYRIGHT HOLDER> ``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
 * <COPYRIGHT HOLDER> 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 "alter.h"
#include "schema.h"
#include "user.h"
#include "space.h"
#include "index.h"
#include "func.h"
#include "coll_id_cache.h"
#include "coll_id_def.h"
#include "txn.h"
#include "tuple.h"
#include "fiber.h" /* for gc_pool */
#include "scoped_guard.h"
#include "third_party/base64.h"
#include <new> /* for placement new */
#include <stdio.h> /* snprintf() */
#include <ctype.h>
#include "replication.h" /* for replica_set_id() */
#include "session.h" /* to fetch the current user. */
#include "vclock.h" /* VCLOCK_MAX */
#include "xrow.h"
#include "iproto_constants.h"
#include "identifier.h"
#include "version.h"
#include "sequence.h"

/**
 * chap-sha1 of empty string, i.e.
 * base64_encode(sha1(sha1(""), 0)
 */
#define CHAP_SHA1_EMPTY_PASSWORD "vhvewKp0tNyweZQ+cFKAlsyphfg="

/* {{{ Auxiliary functions and methods. */

static void
access_check_ddl(const char *name, uint32_t object_id, uint32_t owner_uid,
                 enum schema_object_type type, enum priv_type priv_type,
                 bool is_17_compat_mode)
{
        struct credentials *cr = effective_user();
        user_access_t has_access = cr->universal_access;
        /*
         * XXX: pre 1.7.7 there was no specific 'CREATE' or
         * 'ALTER' ACL, instead, read and write access on universe
         * was used to allow create/alter.
         * For backward compatibility, if a user has read and write
         * access on the universe, grant it CREATE access
         * automatically.
         * The legacy fix does not affect sequences since they
         * were added in 1.7.7 only.
         */
        if (is_17_compat_mode && has_access & PRIV_R && has_access & PRIV_W)
                has_access |= PRIV_C | PRIV_A;

        user_access_t access = ((PRIV_U | (user_access_t) priv_type) &
                                ~has_access);
        bool is_owner = owner_uid == cr->uid || cr->uid == ADMIN;
        if (access == 0)
                return; /* Access granted. */
        /* Check for specific entity access. */
        struct access *object = entity_access_get(type);
        if (object) {
                access &= ~object[cr->auth_token].effective;
        }
        /*
         * Only the owner of the object or someone who has
         * specific DDL privilege on the object can execute
         * DDL. If a user has no USAGE access and is owner,
         * deny access as well.
         * If a user wants to CREATE an object, they're of course
         * the owner of the object, but this should be ignored --
         * CREATE privilege is required.
         */
        if (access == 0 || (is_owner && !(access & (PRIV_U | PRIV_C))))
                return; /* Access granted. */
        /*
         * USAGE can be granted only globally.
         */
        if (!(access & PRIV_U)) {
                /* Check for privileges on a single object. */
                struct access *object = access_find(type, object_id);
                if (object != NULL)
                        access &= ~object[cr->auth_token].effective;
                if (access == 0)
                        return; /* Access granted. */
        }
        /* Create a meaningful error message. */
        struct user *user = user_find_xc(cr->uid);
        const char *object_name;
        const char *pname;
        if (access & PRIV_U) {
                object_name = schema_object_name(SC_UNIVERSE);
                pname = priv_name(PRIV_U);
                name = "";
        } else {
                object_name = schema_object_name(type);
                pname = priv_name(access);
        }
        tnt_raise(AccessDeniedError, pname, object_name, name,
                  user->def->name);
}

/**
 * Throw an exception if the given index definition
 * is incompatible with a sequence.
 */
static void
index_def_check_sequence(struct index_def *index_def, const char *space_name)
{
        enum field_type type = index_def->key_def->parts[0].type;
        if (type != FIELD_TYPE_UNSIGNED && type != FIELD_TYPE_INTEGER) {
                tnt_raise(ClientError, ER_MODIFY_INDEX, index_def->name,
                          space_name, "sequence cannot be used with "
                          "a non-integer key");
        }
}

/**
 * Support function for index_def_new_from_tuple(..)
 * Checks tuple (of _index space) and throws a nice error if it is invalid
 * Checks only types of fields and their count!
 * Additionally determines version of tuple structure
 * is_166plus is set as true if tuple structure is 1.6.6+
 * is_166plus is set as false if tuple structure is 1.6.5-
 */
static void
index_def_check_tuple(const struct tuple *tuple, bool *is_166plus)
{
        *is_166plus = true;
        const mp_type common_template[] = {MP_UINT, MP_UINT, MP_STR, MP_STR};
        const char *data = tuple_data(tuple);
        uint32_t field_count = mp_decode_array(&data);
        const char *field_start = data;
        if (field_count < 6)
                goto err;
        for (size_t i = 0; i < lengthof(common_template); i++) {
                enum mp_type type = mp_typeof(*data);
                if (type != common_template[i])
                        goto err;
                mp_next(&data);
        }
        if (mp_typeof(*data) == MP_UINT) {
                /* old 1.6.5- version */
                /* TODO: removed it in newer versions, find all 1.6.5- */
                *is_166plus = false;
                mp_next(&data);
                if (mp_typeof(*data) != MP_UINT)
                        goto err;
                if (field_count % 2)
                        goto err;
                mp_next(&data);
                for (uint32_t i = 6; i < field_count; i += 2) {
                        if (mp_typeof(*data) != MP_UINT)
                                goto err;
                        mp_next(&data);
                        if (mp_typeof(*data) != MP_STR)
                                goto err;
                        mp_next(&data);
                }
        } else {
                if (field_count != 6)
                        goto err;
                if (mp_typeof(*data) != MP_MAP)
                        goto err;
                mp_next(&data);
                if (mp_typeof(*data) != MP_ARRAY)
                        goto err;
        }
        return;

err:
        char got[DIAG_ERRMSG_MAX];
        char *p = got, *e = got + sizeof(got);
        data = field_start;
        for (uint32_t i = 0; i < field_count && p < e; i++) {
                enum mp_type type = mp_typeof(*data);
                mp_next(&data);
                const char *type_name;
                switch (type) {
                case MP_UINT:
                        type_name = "number";
                        break;
                case MP_STR:
                        type_name = "string";
                        break;
                case MP_ARRAY:
                        type_name = "array";
                        break;
                case MP_MAP:
                        type_name = "map";
                        break;
                default:
                        type_name = "unknown";
                        break;
                }
                p += snprintf(p, e - p, i ? ", %s" : "%s", type_name);
        }
        const char *expected;
        if (*is_166plus) {
                expected = "space id (number), index id (number), "
                        "name (string), type (string), "
                        "options (map), parts (array)";
        } else {
                expected = "space id (number), index id (number), "
                        "name (string), type (string), "
                        "is_unique (number), part count (number) "
                        "part0 field no (number), "
                        "part0 field type (string), ...";
        }
        tnt_raise(ClientError, ER_WRONG_INDEX_RECORD, got, expected);
}

/**
 * Fill index_opts structure from opts field in tuple of space _index
 * Throw an error is unrecognized option.
 */
static void
index_opts_decode(struct index_opts *opts, const char *map)
{
        index_opts_create(opts);
        if (opts_decode(opts, index_opts_reg, &map, ER_WRONG_INDEX_OPTIONS,
                        BOX_INDEX_FIELD_OPTS, NULL) != 0)
                diag_raise();
        if (opts->distance == rtree_index_distance_type_MAX) {
                tnt_raise(ClientError, ER_WRONG_INDEX_OPTIONS,
                          BOX_INDEX_FIELD_OPTS, "distance must be either "\
                          "'euclid' or 'manhattan'");
        }
        if (opts->page_size <= 0 || (opts->range_size > 0 &&
                                     opts->page_size > opts->range_size)) {
                tnt_raise(ClientError, ER_WRONG_INDEX_OPTIONS,
                          BOX_INDEX_FIELD_OPTS,
                          "page_size must be greater than 0 and "
                          "less than or equal to range_size");
        }
        if (opts->run_count_per_level <= 0) {
                tnt_raise(ClientError, ER_WRONG_INDEX_OPTIONS,
                          BOX_INDEX_FIELD_OPTS,
                          "run_count_per_level must be greater than 0");
        }
        if (opts->run_size_ratio <= 1) {
                tnt_raise(ClientError, ER_WRONG_INDEX_OPTIONS,
                          BOX_INDEX_FIELD_OPTS,
                          "run_size_ratio must be greater than 1");
        }
        if (opts->bloom_fpr <= 0 || opts->bloom_fpr > 1) {
                tnt_raise(ClientError, ER_WRONG_INDEX_OPTIONS,
                          BOX_INDEX_FIELD_OPTS,
                          "bloom_fpr must be greater than 0 and "
                          "less than or equal to 1");
        }
}

/**
 * Create a index_def object from a record in _index
 * system space.
 *
 * Check that:
 * - index id is within range
 * - index type is supported
 * - part count > 0
 * - there are parts for the specified part count
 * - types of parts in the parts array are known to the system
 * - fieldno of each part in the parts array is within limits
 */
static struct index_def *
index_def_new_from_tuple(struct tuple *tuple, struct space *space)
{
        bool is_166plus;
        index_def_check_tuple(tuple, &is_166plus);

        struct index_opts opts;
        index_opts_create(&opts);
        uint32_t id = tuple_field_u32_xc(tuple, BOX_INDEX_FIELD_SPACE_ID);
        uint32_t index_id = tuple_field_u32_xc(tuple, BOX_INDEX_FIELD_ID);
        enum index_type type =
                STR2ENUM(index_type, tuple_field_cstr_xc(tuple,
                                                         BOX_INDEX_FIELD_TYPE));
        uint32_t name_len;
        const char *name = tuple_field_str_xc(tuple, BOX_INDEX_FIELD_NAME,
                                              &name_len);
        uint32_t part_count;
        const char *parts;
        if (is_166plus) {
                /* 1.6.6+ _index space structure */
                const char *opts_field =
                        tuple_field_with_type_xc(tuple, BOX_INDEX_FIELD_OPTS,
                                                 MP_MAP);
                index_opts_decode(&opts, opts_field);
                parts = tuple_field(tuple, BOX_INDEX_FIELD_PARTS);
                part_count = mp_decode_array(&parts);
        } else {
                /* 1.6.5- _index space structure */
                /* TODO: remove it in newer versions, find all 1.6.5- */
                opts.is_unique =
                        tuple_field_u32_xc(tuple,
                                           BOX_INDEX_FIELD_IS_UNIQUE_165);
                part_count = tuple_field_u32_xc(tuple,
                                                BOX_INDEX_FIELD_PART_COUNT_165);
                parts = tuple_field(tuple, BOX_INDEX_FIELD_PARTS_165);
        }
        if (name_len > BOX_NAME_MAX) {
                tnt_raise(ClientError, ER_MODIFY_INDEX,
                          tt_cstr(name, BOX_INVALID_NAME_MAX),
                          space_name(space), "index name is too long");
        }
        identifier_check_xc(name, name_len);
        struct key_def *key_def = NULL;
        struct key_part_def *part_def = (struct key_part_def *)
                        malloc(sizeof(*part_def) * part_count);
        if (part_def == NULL) {
                tnt_raise(OutOfMemory, sizeof(*part_def) * part_count,
                          "malloc", "key_part_def");
        }
        auto key_def_guard = make_scoped_guard([&] {
                free(part_def);
                if (key_def != NULL)
                        key_def_delete(key_def);
        });
        if (is_166plus) {
                /* 1.6.6+ */
                if (key_def_decode_parts(part_def, part_count, &parts,
                                         space->def->fields,
                                         space->def->field_count) != 0)
                        diag_raise();
        } else {
                /* 1.6.5- TODO: remove it in newer versions, find all 1.6.5- */
                if (key_def_decode_parts_160(part_def, part_count, &parts,
                                             space->def->fields,
                                             space->def->field_count) != 0)
                        diag_raise();
        }
        key_def = key_def_new(part_def, part_count);
        if (key_def == NULL)
                diag_raise();
        struct index_def *index_def =
                index_def_new(id, index_id, name, name_len, type,
                              &opts, key_def, space_index_key_def(space, 0));
        if (index_def == NULL)
                diag_raise();
        auto index_def_guard = make_scoped_guard([=] { index_def_delete(index_def); });
        index_def_check_xc(index_def, space_name(space));
        space_check_index_def_xc(space, index_def);
        if (index_def->iid == 0 && space->sequence != NULL)
                index_def_check_sequence(index_def, space_name(space));
        index_def_guard.is_active = false;
        return index_def;
}

/**
 * Fill space opts from the msgpack stream (MP_MAP field in the
 * tuple).
 */
static void
space_opts_decode(struct space_opts *opts, const char *data)
{
        space_opts_create(opts);
        if (data == NULL)
                return;

        bool is_170_plus = (mp_typeof(*data) == MP_MAP);
        if (!is_170_plus) {
                /* Tarantool < 1.7.0 compatibility */
                if (mp_typeof(*data) != MP_STR) {
                        tnt_raise(ClientError, ER_FIELD_TYPE,
                                  BOX_SPACE_FIELD_OPTS + TUPLE_INDEX_BASE,
                                  mp_type_strs[MP_STR]);
                }
                uint32_t len;
                const char *flags = mp_decode_str(&data, &len);
                flags = tt_cstr(flags, len);
                while (flags && *flags) {
                        while (isspace(*flags)) /* skip space */
                                flags++;
                        if (strncmp(flags, "temporary", strlen("temporary")) == 0)
                                opts->is_temporary = true;
                        flags = strchr(flags, ',');
                        if (flags)
                                flags++;
                }
        } else if (opts_decode(opts, space_opts_reg, &data,
                               ER_WRONG_SPACE_OPTIONS,
                               BOX_SPACE_FIELD_OPTS, NULL) != 0) {
                diag_raise();
        }
}

/**
 * Decode field definition from MessagePack map. Format:
 * {name: <string>, type: <string>}. Type is optional.
 * @param[out] field Field to decode to.
 * @param data MessagePack map to decode.
 * @param space_name Name of a space, from which the field is got.
 *        Used in error messages.
 * @param name_len Length of @a space_name.
 * @param errcode Error code to use for client errors. Either
 *        create or modify space errors.
 * @param fieldno Field number to decode. Used in error messages.
 * @param region Region to allocate field name.
 */
static void
field_def_decode(struct field_def *field, const char **data,
                 const char *space_name, uint32_t name_len,
                 uint32_t errcode, uint32_t fieldno, struct region *region)
{
        if (mp_typeof(**data) != MP_MAP) {
                tnt_raise(ClientError, errcode, tt_cstr(space_name, name_len),
                          tt_sprintf("field %d is not map",
                                     fieldno + TUPLE_INDEX_BASE));
        }
        int count = mp_decode_map(data);
        *field = field_def_default;
        for (int i = 0; i < count; ++i) {
                if (mp_typeof(**data) != MP_STR) {
                        tnt_raise(ClientError, errcode,
                                  tt_cstr(space_name, name_len),
                                  tt_sprintf("field %d format is not map"\
                                             " with string keys",
                                             fieldno + TUPLE_INDEX_BASE));
                }
                uint32_t key_len;
                const char *key = mp_decode_str(data, &key_len);
                if (opts_parse_key(field, field_def_reg, key, key_len, data,
                                   ER_WRONG_SPACE_FORMAT,
                                   fieldno + TUPLE_INDEX_BASE, region,
                                   true) != 0)
                        diag_raise();
        }
        if (field->name == NULL) {
                tnt_raise(ClientError, errcode, tt_cstr(space_name, name_len),
                          tt_sprintf("field %d name is not specified",
                                     fieldno + TUPLE_INDEX_BASE));
        }
        size_t field_name_len = strlen(field->name);
        if (field_name_len > BOX_NAME_MAX) {
                tnt_raise(ClientError, errcode, tt_cstr(space_name, name_len),
                          tt_sprintf("field %d name is too long",
                                     fieldno + TUPLE_INDEX_BASE));
        }
        identifier_check_xc(field->name, field_name_len);
        if (field->type == field_type_MAX) {
                tnt_raise(ClientError, errcode, tt_cstr(space_name, name_len),
                          tt_sprintf("field %d has unknown field type",
                                     fieldno + TUPLE_INDEX_BASE));
        }
}

/**
 * Decode MessagePack array of fields.
 * @param data MessagePack array of fields.
 * @param[out] out_count Length of a result array.
 * @param space_name Space name to use in error messages.
 * @param errcode Errcode for client errors.
 * @param region Region to allocate result array.
 *
 * @retval Array of fields.
 */
static struct field_def *
space_format_decode(const char *data, uint32_t *out_count,
                    const char *space_name, uint32_t name_len,
                    uint32_t errcode, struct region *region)
{
        /* Type is checked by _space format. */
        assert(mp_typeof(*data) == MP_ARRAY);
        uint32_t count = mp_decode_array(&data);
        *out_count = count;
        if (count == 0)
                return NULL;
        size_t size = count * sizeof(struct field_def);
        struct field_def *region_defs =
                (struct field_def *) region_alloc_xc(region, size);
        for (uint32_t i = 0; i < count; ++i) {
                field_def_decode(&region_defs[i], &data, space_name, name_len,
                                 errcode, i, region);
        }
        return region_defs;
}

/**
 * Fill space_def structure from struct tuple.
 */
static struct space_def *
space_def_new_from_tuple(struct tuple *tuple, uint32_t errcode,
                         struct region *region)
{
        uint32_t name_len;
        const char *name =
                tuple_field_str_xc(tuple, BOX_SPACE_FIELD_NAME, &name_len);
        if (name_len > BOX_NAME_MAX)
                tnt_raise(ClientError, errcode,
                          tt_cstr(name, BOX_INVALID_NAME_MAX),
                          "space name is too long");
        identifier_check_xc(name, name_len);
        uint32_t id = tuple_field_u32_xc(tuple, BOX_SPACE_FIELD_ID);
        if (id > BOX_SPACE_MAX) {
                tnt_raise(ClientError, errcode, tt_cstr(name, name_len),
                          "space id is too big");
        }
        if (id == 0) {
                tnt_raise(ClientError, errcode, tt_cstr(name, name_len),
                          "space id 0 is reserved");
        }
        uint32_t uid = tuple_field_u32_xc(tuple, BOX_SPACE_FIELD_UID);
        uint32_t exact_field_count =
                tuple_field_u32_xc(tuple, BOX_SPACE_FIELD_FIELD_COUNT);
        uint32_t engine_name_len;
        const char *engine_name =
                tuple_field_str_xc(tuple, BOX_SPACE_FIELD_ENGINE,
                                   &engine_name_len);
        /*
         * Engines are compiled-in so their names are known in
         * advance to be shorter than names of other identifiers.
         */
        if (engine_name_len > ENGINE_NAME_MAX) {
                tnt_raise(ClientError, errcode, tt_cstr(name, name_len),
                          "space engine name is too long");
        }
        identifier_check_xc(engine_name, engine_name_len);
        const char *space_opts;
        struct field_def *fields;
        uint32_t field_count;
        if (dd_version_id >= version_id(1, 7, 6)) {
                /* Check space opts. */
                space_opts =
                        tuple_field_with_type_xc(tuple, BOX_SPACE_FIELD_OPTS,
                                                 MP_MAP);
                /* Check space format */
                const char *format =
                        tuple_field_with_type_xc(tuple, BOX_SPACE_FIELD_FORMAT,
                                                 MP_ARRAY);
                fields = space_format_decode(format, &field_count, name,
                                             name_len, errcode, region);
                if (exact_field_count != 0 &&
                    exact_field_count < field_count) {
                        tnt_raise(ClientError, errcode, tt_cstr(name, name_len),
                                  "exact_field_count must be either 0 or >= "\
                                  "formatted field count");
                }
        } else {
                fields = NULL;
                field_count = 0;
                space_opts = tuple_field(tuple, BOX_SPACE_FIELD_OPTS);
        }
        struct space_opts opts;
        space_opts_decode(&opts, space_opts);
        /*
         * Currently, only predefined replication groups
         * are supported.
         */
        if (opts.group_id != GROUP_DEFAULT &&
            opts.group_id != GROUP_LOCAL) {
                tnt_raise(ClientError, ER_NO_SUCH_GROUP,
                          int2str(opts.group_id));
        }
        struct space_def *def =
                space_def_new_xc(id, uid, exact_field_count, name, name_len,
                                 engine_name, engine_name_len, &opts, fields,
                                 field_count);
        auto def_guard = make_scoped_guard([=] { space_def_delete(def); });
        struct engine *engine = engine_find_xc(def->engine_name);
        engine_check_space_def_xc(engine, def);
        def_guard.is_active = false;
        return def;
}

/**
 * Space old and new space triggers (move the original triggers
 * to the new space, or vice versa, restore the original triggers
 * in the old space).
 */
static void
space_swap_triggers(struct space *new_space, struct space *old_space)
{
        rlist_swap(&new_space->before_replace, &old_space->before_replace);
        rlist_swap(&new_space->on_replace, &old_space->on_replace);
        rlist_swap(&new_space->on_stmt_begin, &old_space->on_stmt_begin);
}

/**
 * True if the space has records identified by key 'uid'.
 * Uses 'iid' index.
 */
bool
space_has_data(uint32_t id, uint32_t iid, uint32_t uid)
{
        struct space *space = space_by_id(id);
        if (space == NULL)
                return false;

        if (space_index(space, iid) == NULL)
                return false;

        struct index *index = index_find_system_xc(space, iid);
        char key[6];
        assert(mp_sizeof_uint(BOX_SYSTEM_ID_MIN) <= sizeof(key));
        mp_encode_uint(key, uid);
        struct iterator *it = index_create_iterator_xc(index, ITER_EQ, key, 1);
        IteratorGuard iter_guard(it);
        if (iterator_next_xc(it) != NULL)
                return true;
        return false;
}

/* }}} */

/* {{{ struct alter_space - the body of a full blown alter */
struct alter_space;

class AlterSpaceOp {
public:
        AlterSpaceOp(struct alter_space *alter);

        /** Link in alter_space::ops. */
        struct rlist link;
        /**
         * Called before creating the new space. Used to update
         * the space definition and/or key list that will be used
         * for creating the new space. Must not yield or fail.
         */
        virtual void alter_def(struct alter_space * /* alter */) {}
        /**
         * Called after creating a new space. Used for performing
         * long-lasting operations, such as index rebuild or format
         * check. May yield. May throw an exception. Must not modify
         * the old space.
         */
        virtual void prepare(struct alter_space * /* alter */) {}
        /**
         * Called after all registered operations have completed
         * the preparation phase. Used to propagate the old space
         * state to the new space (e.g. move unchanged indexes).
         * Must not yield or fail.
         */
        virtual void alter(struct alter_space * /* alter */) {}
        /**
         * Called after the change has been successfully written
         * to WAL. Must not fail.
         */
        virtual void commit(struct alter_space * /* alter */,
                            int64_t /* signature */) {}
        /**
         * Called in case a WAL error occurred. It is supposed to undo
         * the effect of AlterSpaceOp::prepare and AlterSpaceOp::alter.
         * Must not fail.
         */
        virtual void rollback(struct alter_space * /* alter */) {}

        virtual ~AlterSpaceOp() {}

        void *operator new(size_t size)
        {
                return region_aligned_calloc_xc(&fiber()->gc, size,
                                                alignof(uint64_t));
        }
        void operator delete(void * /* ptr */) {}
};

/**
 * A trigger installed on transaction commit/rollback events of
 * the transaction which initiated the alter.
 */
static struct trigger *
txn_alter_trigger_new(trigger_f run, void *data)
{
        struct trigger *trigger = (struct trigger *)
                region_calloc_object_xc(&fiber()->gc, struct trigger);
        trigger->run = run;
        trigger->data = data;
        trigger->destroy = NULL;
        return trigger;
}

struct alter_space {
        /** List of alter operations */
        struct rlist ops;
        /** Definition of the new space - space_def. */
        struct space_def *space_def;
        /** Definition of the new space - keys. */
        struct rlist key_list;
        /** Old space. */
        struct space *old_space;
        /** New space. */
        struct space *new_space;
        /**
         * Assigned to the new primary key definition if we're
         * rebuilding the primary key, i.e. changing its key parts
         * substantially.
         */
        struct key_def *pk_def;
        /**
         * Min field count of a new space. It is calculated before
         * the new space is created and used to update optionality
         * of key_defs and key_parts.
         */
        uint32_t new_min_field_count;
};

static struct alter_space *
alter_space_new(struct space *old_space)
{
        struct alter_space *alter =
                region_calloc_object_xc(&fiber()->gc, struct alter_space);
        rlist_create(&alter->ops);
        alter->old_space = old_space;
        alter->space_def = space_def_dup_xc(alter->old_space->def);
        if (old_space->format != NULL)
                alter->new_min_field_count = old_space->format->min_field_count;
        else
                alter->new_min_field_count = 0;
        return alter;
}

/** Destroy alter. */
static void
alter_space_delete(struct alter_space *alter)
{
        /* Destroy the ops. */
        while (! rlist_empty(&alter->ops)) {
                AlterSpaceOp *op = rlist_shift_entry(&alter->ops,
                                                     AlterSpaceOp, link);
                delete op;
        }
        /* Delete the new space, if any. */
        if (alter->new_space)
                space_delete(alter->new_space);
        space_def_delete(alter->space_def);
}

AlterSpaceOp::AlterSpaceOp(struct alter_space *alter)
{
        /* Add to the tail: operations must be processed in order. */
        rlist_add_tail_entry(&alter->ops, this, link);
}

/**
 * Commit the alter.
 *
 * Move all unchanged indexes from the old space to the new space.
 * Set the newly built indexes in the new space, or free memory
 * of the dropped indexes.
 * Replace the old space with a new one in the space cache.
 */
static void
alter_space_commit(struct trigger *trigger, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct alter_space *alter = (struct alter_space *) trigger->data;
        /*
         * Commit alter ops, this will move the changed
         * indexes into their new places.
         */
        class AlterSpaceOp *op;
        rlist_foreach_entry(op, &alter->ops, link) {
                op->commit(alter, txn->signature);
        }

        trigger_run_xc(&on_alter_space, alter->new_space);

        alter->new_space = NULL; /* for alter_space_delete(). */
        /*
         * Delete the old version of the space, we are not
         * going to use it.
         */
        space_delete(alter->old_space);
        alter_space_delete(alter);
}

/**
 * Rollback all effects of space alter. This is
 * a transaction trigger, and it fires most likely
 * upon a failed write to the WAL.
 *
 * Keep in mind that we may end up here in case of
 * alter_space_commit() failure (unlikely)
 */
static void
alter_space_rollback(struct trigger *trigger, void * /* event */)
{
        struct alter_space *alter = (struct alter_space *) trigger->data;
        /* Rollback alter ops */
        class AlterSpaceOp *op;
        rlist_foreach_entry(op, &alter->ops, link) {
                op->rollback(alter);
        }
        /* Rebuild index maps once for all indexes. */
        space_fill_index_map(alter->old_space);
        space_fill_index_map(alter->new_space);
        /*
         * Don't forget about space triggers.
         */
        space_swap_triggers(alter->new_space, alter->old_space);
        space_cache_replace(alter->new_space, alter->old_space);
        alter_space_delete(alter);
}

/**
 * alter_space_do() - do all the work necessary to
 * create a new space.
 *
 * If something may fail during alter, it must be done here,
 * before a record is written to the Write Ahead Log.  Only
 * trivial and infallible actions are left to the commit phase
 * of the alter.
 *
 * The implementation of this function follows "Template Method"
 * pattern, providing a skeleton of the alter, while all the
 * details are encapsulated in AlterSpaceOp methods.
 *
 * These are the major steps of alter defining the structure of
 * the algorithm and performed regardless of what is altered:
 *
 * - a copy of the definition of the old space is created
 * - the definition of the old space is altered, to get
 *   definition of a new space
 * - an instance of the new space is created, according to the new
 *   definition; the space is so far empty
 * - data structures of the new space are built; sometimes, it
 *   doesn't need to happen, e.g. when alter only changes the name
 *   of a space or an index, or other accidental property.
 *   If any data structure needs to be built, e.g. a new index,
 *   only this index is built, not the entire space with all its
 *   indexes.
 * - at commit, the new space is coalesced with the old one.
 *   On rollback, the new space is deleted.
 */
static void
alter_space_do(struct txn *txn, struct alter_space *alter)
{
        /*
         * Prepare triggers while we may fail. Note, we don't have to
         * free them in case of failure, because they are allocated on
         * the region.
         */
        struct trigger *on_commit, *on_rollback;
        on_commit = txn_alter_trigger_new(alter_space_commit, alter);
        on_rollback = txn_alter_trigger_new(alter_space_rollback, alter);

        /* Create a definition of the new space. */
        space_dump_def(alter->old_space, &alter->key_list);
        class AlterSpaceOp *op;
        /*
         * Alter the definition of the old space, so that
         * a new space can be created with a new definition.
         */
        rlist_foreach_entry(op, &alter->ops, link)
                op->alter_def(alter);
        /*
         * Create a new (empty) space for the new definition.
         * Sic: the triggers are not moved over yet.
         */
        alter->new_space = space_new_xc(alter->space_def, &alter->key_list);
        /*
         * Copy the replace function, the new space is at the same recovery
         * phase as the old one. This hack is especially necessary for
         * system spaces, which may be altered in some row in the
         * snapshot/xlog, but needs to continue staying "fully
         * built".
         */
        space_prepare_alter_xc(alter->old_space, alter->new_space);

        alter->new_space->sequence = alter->old_space->sequence;
        memcpy(alter->new_space->access, alter->old_space->access,
               sizeof(alter->old_space->access));

        /*
         * Build new indexes, check if tuples conform to
         * the new space format.
         */
        rlist_foreach_entry(op, &alter->ops, link)
                op->prepare(alter);

        /*
         * This function must not throw exceptions or yield after
         * this point.
         */

        /* Move old indexes, update space format. */
        rlist_foreach_entry(op, &alter->ops, link)
                op->alter(alter);

        /* Rebuild index maps once for all indexes. */
        space_fill_index_map(alter->old_space);
        space_fill_index_map(alter->new_space);
        /*
         * Don't forget about space triggers.
         */
        space_swap_triggers(alter->new_space, alter->old_space);
        /*
         * The new space is ready. Time to update the space
         * cache with it.
         */
        space_cache_replace(alter->old_space, alter->new_space);

        /*
         * Install transaction commit/rollback triggers to either
         * finish or rollback the DDL depending on the results of
         * writing to WAL.
         */
        txn_on_commit(txn, on_commit);
        txn_on_rollback(txn, on_rollback);
}

/* }}}  */

/* {{{ AlterSpaceOp descendants - alter operations, such as Add/Drop index */

/**
 * This operation does not modify the space, it just checks that
 * tuples stored in it conform to the new format.
 */
class CheckSpaceFormat: public AlterSpaceOp
{
public:
        CheckSpaceFormat(struct alter_space *alter)
                :AlterSpaceOp(alter) {}
        virtual void prepare(struct alter_space *alter);
};

void
CheckSpaceFormat::prepare(struct alter_space *alter)
{
        struct space *new_space = alter->new_space;
        struct space *old_space = alter->old_space;
        struct tuple_format *new_format = new_space->format;
        struct tuple_format *old_format = old_space->format;
        if (old_format != NULL) {
                assert(new_format != NULL);
                if (!tuple_format1_can_store_format2_tuples(new_format,
                                                            old_format))
                    space_check_format_xc(old_space, new_format);
        }
}

/** Change non-essential properties of a space. */
class ModifySpace: public AlterSpaceOp
{
public:
        ModifySpace(struct alter_space *alter, struct space_def *def)
                :AlterSpaceOp(alter), new_def(def), new_dict(NULL) {}
        /* New space definition. */
        struct space_def *new_def;
        /**
         * Newely created field dictionary. When new space_def is
         * created, it allocates new dictionary. Alter moves new
         * names into an old dictionary and deletes new one.
         */
        struct tuple_dictionary *new_dict;
        virtual void alter_def(struct alter_space *alter);
        virtual void alter(struct alter_space *alter);
        virtual void rollback(struct alter_space *alter);
        virtual ~ModifySpace();
};

/** Amend the definition of the new space. */
void
ModifySpace::alter_def(struct alter_space *alter)
{
        /*
         * Use the old dictionary for the new space, because
         * it is already referenced by existing tuple formats.
         * We will update it in place in ModifySpace::alter.
         */
        new_dict = new_def->dict;
        new_def->dict = alter->old_space->def->dict;
        tuple_dictionary_ref(new_def->dict);

        space_def_delete(alter->space_def);
        alter->space_def = new_def;
        /* Now alter owns the def. */
        new_def = NULL;
}

void
ModifySpace::alter(struct alter_space *alter)
{
        /*
         * Move new names into an old dictionary, which already is
         * referenced by existing tuple formats. New dictionary
         * object is deleted later, in destructor.
         */
        tuple_dictionary_swap(alter->new_space->def->dict, new_dict);
}

void
ModifySpace::rollback(struct alter_space *alter)
{
        tuple_dictionary_swap(alter->new_space->def->dict, new_dict);
}

ModifySpace::~ModifySpace()
{
        if (new_dict != NULL)
                tuple_dictionary_unref(new_dict);
        if (new_def != NULL)
                space_def_delete(new_def);
}

/** DropIndex - remove an index from space. */

class DropIndex: public AlterSpaceOp
{
public:
        DropIndex(struct alter_space *alter, struct index_def *def_arg)
                :AlterSpaceOp(alter), old_index_def(def_arg) {}
        /** A reference to the definition of the dropped index. */
        struct index_def *old_index_def;
        virtual void alter_def(struct alter_space *alter);
        virtual void prepare(struct alter_space *alter);
        virtual void commit(struct alter_space *alter, int64_t lsn);
};

/*
 * Alter the definition of the new space and remove
 * the new index from it.
 */
void
DropIndex::alter_def(struct alter_space * /* alter */)
{
        rlist_del_entry(old_index_def, link);
}

/* Do the drop. */
void
DropIndex::prepare(struct alter_space *alter)
{
        if (old_index_def->iid == 0)
                space_drop_primary_key(alter->new_space);
}

void
DropIndex::commit(struct alter_space *alter, int64_t signature)
{
        struct index *index = space_index(alter->old_space,
                                          old_index_def->iid);
        assert(index != NULL);
        index_commit_drop(index, signature);
}

/**
 * A no-op to preserve the old index data in the new space.
 * Added to the alter specification when the index at hand
 * is not affected by alter in any way.
 */
class MoveIndex: public AlterSpaceOp
{
public:
        MoveIndex(struct alter_space *alter, uint32_t iid_arg)
                :AlterSpaceOp(alter), iid(iid_arg) {}
        /** id of the index on the move. */
        uint32_t iid;
        virtual void alter(struct alter_space *alter);
        virtual void rollback(struct alter_space *alter);
};

void
MoveIndex::alter(struct alter_space *alter)
{
        space_swap_index(alter->old_space, alter->new_space, iid, iid);
}

void
MoveIndex::rollback(struct alter_space *alter)
{
        space_swap_index(alter->old_space, alter->new_space, iid, iid);
}

/**
 * Change non-essential properties of an index, i.e.
 * properties not involving index data or layout on disk.
 */
class ModifyIndex: public AlterSpaceOp
{
public:
        ModifyIndex(struct alter_space *alter,
                    struct index_def *new_index_def_arg,
                    struct index_def *old_index_def_arg)
                : AlterSpaceOp(alter),new_index_def(new_index_def_arg),
                  old_index_def(old_index_def_arg) {
                if (new_index_def->iid == 0 &&
                    key_part_cmp(new_index_def->key_def->parts,
                                 new_index_def->key_def->part_count,
                                 old_index_def->key_def->parts,
                                 old_index_def->key_def->part_count) != 0) {
                        /*
                         * Primary parts have been changed -
                         * update secondary indexes.
                         */
                        alter->pk_def = new_index_def->key_def;
                }
        }
        struct index_def *new_index_def;
        struct index_def *old_index_def;
        virtual void alter_def(struct alter_space *alter);
        virtual void alter(struct alter_space *alter);
        virtual void commit(struct alter_space *alter, int64_t lsn);
        virtual void rollback(struct alter_space *alter);
        virtual ~ModifyIndex();
};

/** Update the definition of the new space */
void
ModifyIndex::alter_def(struct alter_space *alter)
{
        rlist_del_entry(old_index_def, link);
        index_def_list_add(&alter->key_list, new_index_def);
}

void
ModifyIndex::alter(struct alter_space *alter)
{
        assert(old_index_def->iid == new_index_def->iid);
        /*
         * Move the old index to the new space to preserve the
         * original data, but use the new definition.
         */
        space_swap_index(alter->old_space, alter->new_space,
                         old_index_def->iid, new_index_def->iid);
        struct index *old_index = space_index(alter->old_space,
                                              old_index_def->iid);
        assert(old_index != NULL);
        struct index *new_index = space_index(alter->new_space,
                                              new_index_def->iid);
        assert(new_index != NULL);
        SWAP(old_index->def, new_index->def);
        index_update_def(new_index);
}

void
ModifyIndex::commit(struct alter_space *alter, int64_t signature)
{
        struct index *new_index = space_index(alter->new_space,
                                              new_index_def->iid);
        assert(new_index != NULL);
        index_commit_modify(new_index, signature);
}

void
ModifyIndex::rollback(struct alter_space *alter)
{
        assert(old_index_def->iid == new_index_def->iid);
        /*
         * Restore indexes.
         */
        space_swap_index(alter->old_space, alter->new_space,
                         old_index_def->iid, new_index_def->iid);
        struct index *old_index = space_index(alter->old_space,
                                              old_index_def->iid);
        assert(old_index != NULL);
        struct index *new_index = space_index(alter->new_space,
                                              new_index_def->iid);
        assert(new_index != NULL);
        SWAP(old_index->def, new_index->def);
        index_update_def(old_index);
}

ModifyIndex::~ModifyIndex()
{
        index_def_delete(new_index_def);
}

/** CreateIndex - add a new index to the space. */
class CreateIndex: public AlterSpaceOp
{
public:
        CreateIndex(struct alter_space *alter)
                :AlterSpaceOp(alter), new_index(NULL), new_index_def(NULL)
        {}
        /** New index. */
        struct index *new_index;
        /** New index index_def. */
        struct index_def *new_index_def;
        virtual void alter_def(struct alter_space *alter);
        virtual void prepare(struct alter_space *alter);
        virtual void commit(struct alter_space *alter, int64_t lsn);
        virtual ~CreateIndex();
};

/** Add definition of the new key to the new space def. */
void
CreateIndex::alter_def(struct alter_space *alter)
{
        index_def_list_add(&alter->key_list, new_index_def);
}

/**
 * Optionally build the new index.
 *
 * During recovery the space is often not fully constructed yet
 * anyway, so there is no need to fully populate index with data,
 * it is done at the end of recovery.
 *
 * Note, that system spaces are exception to this, since
 * they are fully enabled at all times.
 */
void
CreateIndex::prepare(struct alter_space *alter)
{
        /* Get the new index and build it.  */
        new_index = space_index(alter->new_space, new_index_def->iid);
        assert(new_index != NULL);

        if (new_index_def->iid == 0) {
                /*
                 * Adding a primary key: bring the space
                 * up to speed with the current recovery
                 * state. During snapshot recovery it
                 * means preparing the primary key for
                 * build (beginBuild()). During xlog
                 * recovery, it means building the primary
                 * key. After recovery, it means building
                 * all keys.
                 */
                space_add_primary_key_xc(alter->new_space);
                return;
        }
        space_build_index_xc(alter->old_space, new_index,
                             alter->new_space->format);
}

void
CreateIndex::commit(struct alter_space *alter, int64_t signature)
{
        (void) alter;
        assert(new_index != NULL);
        index_commit_create(new_index, signature);
        new_index = NULL;
}

CreateIndex::~CreateIndex()
{
        if (new_index != NULL)
                index_abort_create(new_index);
        if (new_index_def != NULL)
                index_def_delete(new_index_def);
}

/**
 * RebuildIndex - drop the old index data and rebuild index
 * from by reading the primary key. Used when key_def of
 * an index is changed.
 */
class RebuildIndex: public AlterSpaceOp
{
public:
        RebuildIndex(struct alter_space *alter,
                     struct index_def *new_index_def_arg,
                     struct index_def *old_index_def_arg)
                :AlterSpaceOp(alter), new_index(NULL),
                new_index_def(new_index_def_arg),
                old_index_def(old_index_def_arg)
        {
                /* We may want to rebuild secondary keys as well. */
                if (new_index_def->iid == 0)
                        alter->pk_def = new_index_def->key_def;
        }
        /** New index. */
        struct index *new_index;
        /** New index index_def. */
        struct index_def *new_index_def;
        /** Old index index_def. */
        struct index_def *old_index_def;
        virtual void alter_def(struct alter_space *alter);
        virtual void prepare(struct alter_space *alter);
        virtual void commit(struct alter_space *alter, int64_t signature);
        virtual ~RebuildIndex();
};

/** Add definition of the new key to the new space def. */
void
RebuildIndex::alter_def(struct alter_space *alter)
{
        rlist_del_entry(old_index_def, link);
        index_def_list_add(&alter->key_list, new_index_def);
}

void
RebuildIndex::prepare(struct alter_space *alter)
{
        /* Get the new index and build it.  */
        new_index = space_index(alter->new_space, new_index_def->iid);
        assert(new_index != NULL);
        space_build_index_xc(alter->old_space, new_index,
                             alter->new_space->format);
}

void
RebuildIndex::commit(struct alter_space *alter, int64_t signature)
{
        struct index *old_index = space_index(alter->old_space,
                                              old_index_def->iid);
        assert(old_index != NULL);
        index_commit_drop(old_index, signature);
        assert(new_index != NULL);
        index_commit_create(new_index, signature);
        new_index = NULL;
}

RebuildIndex::~RebuildIndex()
{
        if (new_index != NULL)
                index_abort_create(new_index);
        if (new_index_def != NULL)
                index_def_delete(new_index_def);
}

/** TruncateIndex - truncate an index. */
class TruncateIndex: public AlterSpaceOp
{
public:
        TruncateIndex(struct alter_space *alter, uint32_t iid)
                : AlterSpaceOp(alter), iid(iid) {}
        /** id of the index to truncate. */
        uint32_t iid;
        virtual void prepare(struct alter_space *alter);
        virtual void commit(struct alter_space *alter, int64_t signature);
};

void
TruncateIndex::prepare(struct alter_space *alter)
{
        if (iid == 0) {
                /*
                 * Notify the engine that the primary index
                 * was truncated.
                 */
                space_drop_primary_key(alter->new_space);
                space_add_primary_key_xc(alter->new_space);
                return;
        }

        /*
         * Although the new index is empty, we still need to call
         * space_build_index() to let the engine know that the
         * index was recreated. For example, Vinyl uses this
         * callback to load indexes during local recovery.
         */
        struct index *new_index = space_index(alter->new_space, iid);
        assert(new_index != NULL);
        space_build_index_xc(alter->new_space, new_index,
                             alter->new_space->format);
}

void
TruncateIndex::commit(struct alter_space *alter, int64_t signature)
{
        struct index *old_index = space_index(alter->old_space, iid);
        struct index *new_index = space_index(alter->new_space, iid);

        index_commit_drop(old_index, signature);
        index_commit_create(new_index, signature);
}

/**
 * UpdateSchemaVersion - increment schema_version. Used on
 * in alter_space_do(), i.e. when creating or dropping
 * an index, altering a space.
 */
class UpdateSchemaVersion: public AlterSpaceOp
{
public:
        UpdateSchemaVersion(struct alter_space * alter)
                :AlterSpaceOp(alter) {}
        virtual void alter(struct alter_space *alter);
};

void
UpdateSchemaVersion::alter(struct alter_space *alter)
{
    (void)alter;
    ++schema_version;
}

/* }}} */

/**
 * Delete the space. It is already removed from the space cache.
 */
static void
on_drop_space_commit(struct trigger *trigger, void *event)
{
        (void) event;
        struct space *space = (struct space *)trigger->data;
        trigger_run_xc(&on_alter_space, space);
        space_delete(space);
}

/**
 * Return the original space back into the cache. The effect
 * of all other events happened after the space was removed were
 * reverted by the cascading rollback.
 */
static void
on_drop_space_rollback(struct trigger *trigger, void *event)
{
        (void) event;
        struct space *space = (struct space *)trigger->data;
        space_cache_replace(NULL, space);
}

/**
 * Run the triggers registered on commit of a change in _space.
 */
static void
on_create_space_commit(struct trigger *trigger, void *event)
{
        (void) event;
        struct space *space = (struct space *)trigger->data;
        trigger_run_xc(&on_alter_space, space);
}

/**
 * A trigger invoked on commit/rollback of DROP/ADD space.
 * The trigger removes the space from the space cache.
 *
 * By the time the space is removed, it should be empty: we
 * rely on cascading rollback.
 */
static void
on_create_space_rollback(struct trigger *trigger, void *event)
{
        (void) event;
        struct space *space = (struct space *)trigger->data;
        space_cache_replace(space, NULL);
        space_delete(space);
}

/**
 * Create MoveIndex operation for a range of indexes in a space
 * for range [begin, end)
 */
void
alter_space_move_indexes(struct alter_space *alter, uint32_t begin,
                         uint32_t end)
{
        struct space *old_space = alter->old_space;
        bool is_min_field_count_changed;
        if (old_space->format != NULL) {
                is_min_field_count_changed =
                        old_space->format->min_field_count !=
                        alter->new_min_field_count;
        } else {
                is_min_field_count_changed = false;
        }
        for (uint32_t index_id = begin; index_id < end; ++index_id) {
                struct index *old_index = space_index(old_space, index_id);
                if (old_index == NULL)
                        continue;
                struct index_def *old_def = old_index->def;
                struct index_def *new_def;
                uint32_t min_field_count = alter->new_min_field_count;
                if (alter->pk_def == NULL || !index_depends_on_pk(old_index)) {
                        if (is_min_field_count_changed) {
                                new_def = index_def_dup(old_def);
                                index_def_update_optionality(new_def,
                                                             min_field_count);
                                (void) new ModifyIndex(alter, new_def, old_def);
                        } else {
                                (void) new MoveIndex(alter, old_def->iid);
                        }
                        continue;
                }
                /*
                 * Rebuild secondary indexes that depend on the
                 * primary key since primary key parts have changed.
                 */
                new_def = index_def_new(old_def->space_id, old_def->iid,
                                        old_def->name, strlen(old_def->name),
                                        old_def->type, &old_def->opts,
                                        old_def->key_def, alter->pk_def);
                index_def_update_optionality(new_def, min_field_count);
                auto guard = make_scoped_guard([=] { index_def_delete(new_def); });
                if (!index_def_change_requires_rebuild(old_index, new_def))
                        (void) new ModifyIndex(alter, new_def, old_def);
                else
                        (void) new RebuildIndex(alter, new_def, old_def);
                guard.is_active = false;
        }
}

/**
 * A trigger which is invoked on replace in a data dictionary
 * space _space.
 *
 * Generally, whenever a data dictionary change occurs
 * 2 things should be done:
 *
 * - space cache should be updated
 *
 * - the space which is changed should be rebuilt according
 *   to the nature of the modification, i.e. indexes added/dropped,
 *   tuple format changed, etc.
 *
 * When dealing with an update of _space space, we have 3 major
 * cases:
 *
 * 1) insert a new tuple: creates a new space
 *    The trigger prepares a space structure to insert
 *    into the  space cache and registers an on commit
 *    hook to perform the registration. Should the statement
 *    itself fail, transaction is rolled back, the transaction
 *    rollback hook must be there to delete the created space
 *    object, avoiding a memory leak. The hooks are written
 *    in a way that excludes the possibility of a failure.
 *
 * 2) delete a tuple: drops an existing space.
 *
 *    A space can be dropped only if it has no indexes.
 *    The only reason for this restriction is that there
 *    must be no tuples in _index without a corresponding tuple
 *    in _space. It's not possible to delete such tuples
 *    automatically (this would require multi-statement
 *    transactions), so instead the trigger verifies that the
 *    records have been deleted by the user.
 *
 *    Then the trigger registers transaction commit hook to
 *    perform the deletion from the space cache.  No rollback hook
 *    is required: if the transaction is rolled back, nothing is
 *    done.
 *
 * 3) modify an existing tuple: some space
 *    properties are immutable, but it's OK to change
 *    space name or field count. This is done in WAL-error-
 *    safe mode.
 *
 * A note about memcached_space: Tarantool 1.4 had a check
 * which prevented re-definition of memcached_space. With
 * dynamic space configuration such a check would be particularly
 * clumsy, so it is simply not done.
 */
static void
on_replace_dd_space(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        txn_check_singlestatement_xc(txn, "Space _space");
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;
        struct region *region = &fiber()->gc;
        /*
         * Things to keep in mind:
         * - old_tuple is set only in case of UPDATE.  For INSERT
         *   or REPLACE it is NULL.
         * - the trigger may be called inside recovery from a snapshot,
         *   when index look up is not possible
         * - _space, _index and other metaspaces initially don't
         *   have a tuple which represents it, this tuple is only
         *   created during recovery from a snapshot.
         *
         * Let's establish whether an old space exists. Use
         * old_tuple ID field, if old_tuple is set, since UPDATE
         * may have changed space id.
         */
        uint32_t old_id = tuple_field_u32_xc(old_tuple ? old_tuple : new_tuple,
                                             BOX_SPACE_FIELD_ID);
        struct space *old_space = space_by_id(old_id);
        if (new_tuple != NULL && old_space == NULL) { /* INSERT */
                struct space_def *def =
                        space_def_new_from_tuple(new_tuple, ER_CREATE_SPACE,
                                                 region);
                access_check_ddl(def->name, def->id, def->uid, SC_SPACE,
                                 PRIV_C, true);
                auto def_guard =
                        make_scoped_guard([=] { space_def_delete(def); });
                RLIST_HEAD(empty_list);
                struct space *space = space_new_xc(def, &empty_list);
                /**
                 * The new space must be inserted in the space
                 * cache right away to achieve linearisable
                 * execution on a replica.
                 */
                space_cache_replace(NULL, space);
                /*
                 * Do not forget to update schema_version right after
                 * inserting the space to the space_cache, since no
                 * AlterSpaceOps are registered in case of space
                 * create.
                 */
                ++schema_version;
                /*
                 * So may happen that until the DDL change record
                 * is written to the WAL, the space is used for
                 * insert/update/delete. All these updates are
                 * rolled back by the pipelined rollback mechanism,
                 * so it's safe to simply drop the space on
                 * rollback.
                 */
                struct trigger *on_commit =
                        txn_alter_trigger_new(on_create_space_commit, space);
                txn_on_commit(txn, on_commit);
                struct trigger *on_rollback =
                        txn_alter_trigger_new(on_create_space_rollback, space);
                txn_on_rollback(txn, on_rollback);
        } else if (new_tuple == NULL) { /* DELETE */
                access_check_ddl(old_space->def->name, old_space->def->id,
                                 old_space->def->uid, SC_SPACE, PRIV_D, true);
                /* Verify that the space is empty (has no indexes) */
                if (old_space->index_count) {
                        tnt_raise(ClientError, ER_DROP_SPACE,
                                  space_name(old_space),
                                  "the space has indexes");
                }
                if (schema_find_grants("space", old_space->def->id)) {
                        tnt_raise(ClientError, ER_DROP_SPACE,
                                  space_name(old_space),
                                  "the space has grants");
                }
                /*
                 * Before 1.7.6 a space record was removed before
                 * the corresponding record in the _truncate system
                 * space so the following check should be disabled.
                 */
                if (dd_version_id >= version_id(1, 7, 6) &&
                    space_has_data(BOX_TRUNCATE_ID, 0, old_space->def->id))
                        tnt_raise(ClientError, ER_DROP_SPACE,
                                  space_name(old_space),
                                  "the space has truncate record");
                /**
                 * The space must be deleted from the space
                 * cache right away to achieve linearisable
                 * execution on a replica.
                 */
                space_cache_replace(old_space, NULL);
                /*
                 * Do not forget to update schema_version right after
                 * deleting the space from the space_cache, since no
                 * AlterSpaceOps are registered in case of space drop.
                 */
                ++schema_version;
                struct trigger *on_commit =
                        txn_alter_trigger_new(on_drop_space_commit, old_space);
                txn_on_commit(txn, on_commit);
                struct trigger *on_rollback =
                        txn_alter_trigger_new(on_drop_space_rollback,
                                              old_space);
                txn_on_rollback(txn, on_rollback);
        } else { /* UPDATE, REPLACE */
                assert(old_space != NULL && new_tuple != NULL);
                struct space_def *def =
                        space_def_new_from_tuple(new_tuple, ER_ALTER_SPACE,
                                                 region);
                access_check_ddl(def->name, def->id, def->uid, SC_SPACE,
                                 PRIV_A, true);
                auto def_guard =
                        make_scoped_guard([=] { space_def_delete(def); });
                if (def->id != space_id(old_space))
                        tnt_raise(ClientError, ER_ALTER_SPACE,
                                  space_name(old_space),
                                  "space id is immutable");
                if (strcmp(def->engine_name, old_space->def->engine_name) != 0)
                        tnt_raise(ClientError, ER_ALTER_SPACE,
                                  space_name(old_space),
                                  "can not change space engine");
                if (def->opts.group_id != space_group_id(old_space))
                        tnt_raise(ClientError, ER_ALTER_SPACE,
                                  space_name(old_space),
                                  "replication group is immutable");
                /*
                 * Allow change of space properties, but do it
                 * in WAL-error-safe mode.
                 */
                struct alter_space *alter = alter_space_new(old_space);
                auto alter_guard =
                        make_scoped_guard([=] {alter_space_delete(alter);});
                /*
                 * Calculate a new min_field_count. It can be
                 * changed by resetting space:format(), if an old
                 * format covers some nullable indexed fields in
                 * the format tail. And when the format is reset,
                 * these fields become optional - index
                 * comparators must be updated.
                 */
                struct key_def **keys;
                size_t bsize = old_space->index_count * sizeof(keys[0]);
                keys = (struct key_def **) region_alloc_xc(&fiber()->gc,
                                                           bsize);
                for (uint32_t i = 0; i < old_space->index_count; ++i)
                        keys[i] = old_space->index[i]->def->key_def;
                alter->new_min_field_count =
                        tuple_format_min_field_count(keys,
                                                     old_space->index_count,
                                                     def->fields,
                                                     def->field_count);
                (void) new CheckSpaceFormat(alter);
                (void) new ModifySpace(alter, def);
                def_guard.is_active = false;
                /* Create MoveIndex ops for all space indexes. */
                alter_space_move_indexes(alter, 0, old_space->index_id_max + 1);
                /* Remember to update schema_version. */
                (void) new UpdateSchemaVersion(alter);
                alter_space_do(txn, alter);
                alter_guard.is_active = false;
        }
}

/**
 * Just like with _space, 3 major cases:
 *
 * - insert a tuple = addition of a new index. The
 *   space should exist.
 *
 * - delete a tuple - drop index.
 *
 * - update a tuple - change of index type or key parts.
 *   Change of index type is the same as deletion of the old
 *   index and addition of the new one.
 *
 *   A new index needs to be built before we attempt to commit
 *   a record to the write ahead log, since:
 *
 *   1) if it fails, it's not good to end up with a corrupt index
 *   which is already committed to WAL
 *
 *   2) Tarantool indexes also work as constraints (min number of
 *   fields in the space, field uniqueness), and it's not good to
 *   commit to WAL a constraint which is not enforced in the
 *   current data set.
 *
 *   When adding a new index, ideally we'd also need to rebuild
 *   all tuple formats in all tuples, since the old format may not
 *   be ideal for the new index. We, however, do not do that,
 *   since that would entail rebuilding all indexes at once.
 *   Instead, the default tuple format of the space is changed,
 *   and as tuples get updated/replaced, all tuples acquire a new
 *   format.
 *
 *   The same is the case with dropping an index: nothing is
 *   rebuilt right away, but gradually the extra space reserved
 *   for offsets is relinquished to the slab allocator as tuples
 *   are modified.
 */
static void
on_replace_dd_index(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        txn_check_singlestatement_xc(txn, "Space _index");
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;
        uint32_t id = tuple_field_u32_xc(old_tuple ? old_tuple : new_tuple,
                                         BOX_INDEX_FIELD_SPACE_ID);
        uint32_t iid = tuple_field_u32_xc(old_tuple ? old_tuple : new_tuple,
                                          BOX_INDEX_FIELD_ID);
        struct space *old_space = space_cache_find_xc(id);
        enum priv_type priv_type = new_tuple ? PRIV_C : PRIV_D;
        if (old_tuple && new_tuple)
                priv_type = PRIV_A;
        access_check_ddl(old_space->def->name, old_space->def->id,
                         old_space->def->uid, SC_SPACE, priv_type, true);
        struct index *old_index = space_index(old_space, iid);

        /*
         * Deal with various cases of dropping of the primary key.
         */
        if (iid == 0 && new_tuple == NULL) {
                /*
                 * Dropping the primary key in a system space: off limits.
                 */
                if (space_is_system(old_space))
                        tnt_raise(ClientError, ER_LAST_DROP,
                                  space_name(old_space));
                /*
                 * Can't drop primary key before secondary keys.
                 */
                if (old_space->index_count > 1) {
                        tnt_raise(ClientError, ER_DROP_PRIMARY_KEY,
                                  space_name(old_space));
                }
                /*
                 * Can't drop primary key before space sequence.
                 */
                if (old_space->sequence != NULL) {
                        tnt_raise(ClientError, ER_ALTER_SPACE,
                                  space_name(old_space),
                                  "can not drop primary key while "
                                  "space sequence exists");
                }
        }

        if (iid != 0 && space_index(old_space, 0) == NULL) {
                /*
                 * A secondary index can not be created without
                 * a primary key.
                 */
                tnt_raise(ClientError, ER_ALTER_SPACE,
                          space_name(old_space),
                          "can not add a secondary key before primary");
        }

        struct alter_space *alter = alter_space_new(old_space);
        auto scoped_guard =
                make_scoped_guard([=] { alter_space_delete(alter); });

        /*
         * Handle the following 4 cases:
         * 1. Simple drop of an index.
         * 2. Creation of a new index: primary or secondary.
         * 3. Change of an index which does not require a rebuild.
         * 4. Change of an index which does require a rebuild.
         */
        /*
         * First, move all unchanged indexes from the old space
         * to the new one.
         */
        /* Case 1: drop the index, if it is dropped. */
        if (old_index != NULL && new_tuple == NULL) {
                alter_space_move_indexes(alter, 0, iid);
                (void) new DropIndex(alter, old_index->def);
        }
        /* Case 2: create an index, if it is simply created. */
        if (old_index == NULL && new_tuple != NULL) {
                alter_space_move_indexes(alter, 0, iid);
                CreateIndex *create_index = new CreateIndex(alter);
                create_index->new_index_def =
                        index_def_new_from_tuple(new_tuple, old_space);
                index_def_update_optionality(create_index->new_index_def,
                                             alter->new_min_field_count);
        }
        /* Case 3 and 4: check if we need to rebuild index data. */
        if (old_index != NULL && new_tuple != NULL) {
                struct index_def *index_def;
                index_def = index_def_new_from_tuple(new_tuple, old_space);
                auto index_def_guard =
                        make_scoped_guard([=] { index_def_delete(index_def); });
                /*
                 * To detect which key parts are optional,
                 * min_field_count is required. But
                 * min_field_count from the old space format can
                 * not be used. For example, consider the case,
                 * when a space has no format, has a primary index
                 * on the first field and has a single secondary
                 * index on a non-nullable second field. Min field
                 * count here is 2. Now alter the secondary index
                 * to make its part be nullable. In the
                 * 'old_space' min_field_count is still 2, but
                 * actually it is already 1. Actual
                 * min_field_count must be calculated using old
                 * unchanged indexes, NEW definition of an updated
                 * index and a space format, defined by a user.
                 */
                struct key_def **keys;
                size_t bsize = old_space->index_count * sizeof(keys[0]);
                keys = (struct key_def **) region_alloc_xc(&fiber()->gc,
                                                           bsize);
                for (uint32_t i = 0, j = 0; i < old_space->index_count;
                     ++i) {
                        struct index_def *d = old_space->index[i]->def;
                        if (d->iid != index_def->iid)
                                keys[j++] = d->key_def;
                        else
                                keys[j++] = index_def->key_def;
                }
                struct space_def *def = old_space->def;
                alter->new_min_field_count =
                        tuple_format_min_field_count(keys,
                                                     old_space->index_count,
                                                     def->fields,
                                                     def->field_count);
                index_def_update_optionality(index_def,
                                             alter->new_min_field_count);
                alter_space_move_indexes(alter, 0, iid);
                if (index_def_cmp(index_def, old_index->def) == 0) {
                        /* Index is not changed so just move it. */
                        (void) new MoveIndex(alter, old_index->def->iid);
                } else if (index_def_change_requires_rebuild(old_index,
                                                             index_def)) {
                        /*
                         * Operation demands an index rebuild.
                         */
                        (void) new RebuildIndex(alter, index_def,
                                                old_index->def);
                        index_def_guard.is_active = false;
                } else {
                        /*
                         * Operation can be done without index rebuild,
                         * but we still need to check that tuples stored
                         * in the space conform to the new format.
                         */
                        (void) new CheckSpaceFormat(alter);
                        (void) new ModifyIndex(alter, index_def,
                                               old_index->def);
                        index_def_guard.is_active = false;
                }
        }
        /*
         * Create MoveIndex ops for the remaining indexes in the
         * old space.
         */
        alter_space_move_indexes(alter, iid + 1, old_space->index_id_max + 1);
        /* Add an op to update schema_version on commit. */
        (void) new UpdateSchemaVersion(alter);
        alter_space_do(txn, alter);
        scoped_guard.is_active = false;
}

/**
 * A trigger invoked on replace in space _truncate.
 *
 * In a nutshell, we truncate a space by replacing it with
 * a new empty space with the same definition and indexes.
 * Note, although we instantiate the new space before WAL
 * write, we don't propagate changes to the old space in
 * case a WAL write error happens and we have to rollback.
 * This is OK, because a WAL write error implies cascading
 * rollback of all transactions following this one.
 */
static void
on_replace_dd_truncate(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct txn_stmt *stmt = txn_current_stmt(txn);
        txn_check_singlestatement_xc(txn, "Space _truncate");
        struct tuple *new_tuple = stmt->new_tuple;

        if (new_tuple == NULL) {
                /* Space drop - nothing to do. */
                return;
        }

        uint32_t space_id =
                tuple_field_u32_xc(new_tuple, BOX_TRUNCATE_FIELD_SPACE_ID);
        struct space *old_space = space_cache_find_xc(space_id);

        if (stmt->row->type == IPROTO_INSERT) {
                /*
                 * Space creation during initial recovery -
                 * nothing to do.
                 */
                return;
        }

        /*
         * System spaces use triggers to keep records in sync
         * with internal objects. Since space truncation doesn't
         * invoke triggers, we don't permit it for system spaces.
         */
        if (space_is_system(old_space))
                tnt_raise(ClientError, ER_TRUNCATE_SYSTEM_SPACE,
                          space_name(old_space));

        /*
         * Check if a write privilege was given, raise an error if not.
         */
        access_check_space_xc(old_space, PRIV_W);

        struct alter_space *alter = alter_space_new(old_space);
        auto scoped_guard =
                make_scoped_guard([=] { alter_space_delete(alter); });

        /*
         * Recreate all indexes of the truncated space.
         */
        for (uint32_t i = 0; i < old_space->index_count; i++) {
                struct index *old_index = old_space->index[i];
                (void) new TruncateIndex(alter, old_index->def->iid);
        }

        alter_space_do(txn, alter);
        scoped_guard.is_active = false;
}

/* {{{ access control */

bool
user_has_data(struct user *user)
{
        uint32_t uid = user->def->uid;
        uint32_t spaces[] = { BOX_SPACE_ID, BOX_FUNC_ID, BOX_SEQUENCE_ID,
                              BOX_PRIV_ID, BOX_PRIV_ID };
        /*
         * owner index id #1 for _space and _func and _priv.
         * For _priv also check that the user has no grants.
         */
        uint32_t indexes[] = { 1, 1, 1, 1, 0 };
        uint32_t count = sizeof(spaces)/sizeof(*spaces);
        for (uint32_t i = 0; i < count; i++) {
                if (space_has_data(spaces[i], indexes[i], uid))
                        return true;
        }
        if (! user_map_is_empty(&user->users))
                return true;
        /*
         * If there was a role, the previous check would have
         * returned true.
         */
        assert(user_map_is_empty(&user->roles));
        return false;
}

/**
 * Supposedly a user may have many authentication mechanisms
 * defined, but for now we only support chap-sha1. Get
 * password of chap-sha1 from the _user space.
 */
void
user_def_fill_auth_data(struct user_def *user, const char *auth_data)
{
        uint8_t type = mp_typeof(*auth_data);
        if (type == MP_ARRAY || type == MP_NIL) {
                /*
                 * Nothing useful.
                 * MP_ARRAY is a special case since Lua arrays are
                 * indistinguishable from tables, so an empty
                 * table may well be encoded as an msgpack array.
                 * Treat as no data.
                 */
                return;
        }
        if (mp_typeof(*auth_data) != MP_MAP) {
                /** Prevent users from making silly mistakes */
                tnt_raise(ClientError, ER_CREATE_USER,
                          user->name, "invalid password format, "
                          "use box.schema.user.passwd() to reset password");
        }
        uint32_t mech_count = mp_decode_map(&auth_data);
        for (uint32_t i = 0; i < mech_count; i++) {
                if (mp_typeof(*auth_data) != MP_STR) {
                        mp_next(&auth_data);
                        mp_next(&auth_data);
                        continue;
                }
                uint32_t len;
                const char *mech_name = mp_decode_str(&auth_data, &len);
                if (strncasecmp(mech_name, "chap-sha1", 9) != 0) {
                        mp_next(&auth_data);
                        continue;
                }
                const char *hash2_base64 = mp_decode_str(&auth_data, &len);
                if (len != 0 && len != SCRAMBLE_BASE64_SIZE) {
                        tnt_raise(ClientError, ER_CREATE_USER,
                                  user->name, "invalid user password");
                }
                if (user->uid == GUEST) {
                    /** Guest user is permitted to have empty password */
                    if (strncmp(hash2_base64, CHAP_SHA1_EMPTY_PASSWORD, len))
                        tnt_raise(ClientError, ER_GUEST_USER_PASSWORD);
                }

                base64_decode(hash2_base64, len, user->hash2,
                              sizeof(user->hash2));
                break;
        }
}

static struct user_def *
user_def_new_from_tuple(struct tuple *tuple)
{
        uint32_t name_len;
        const char *name = tuple_field_str_xc(tuple, BOX_USER_FIELD_NAME,
                                              &name_len);
        if (name_len > BOX_NAME_MAX) {
                tnt_raise(ClientError, ER_CREATE_USER,
                          tt_cstr(name, BOX_INVALID_NAME_MAX),
                          "user name is too long");
        }
        size_t size = user_def_sizeof(name_len);
        /* Use calloc: in case user password is empty, fill it with \0 */
        struct user_def *user = (struct user_def *) malloc(size);
        if (user == NULL)
                tnt_raise(OutOfMemory, size, "malloc", "user");
        auto def_guard = make_scoped_guard([=] { free(user); });
        user->uid = tuple_field_u32_xc(tuple, BOX_USER_FIELD_ID);
        user->owner = tuple_field_u32_xc(tuple, BOX_USER_FIELD_UID);
        const char *user_type =
                tuple_field_cstr_xc(tuple, BOX_USER_FIELD_TYPE);
        user->type= schema_object_type(user_type);
        memcpy(user->name, name, name_len);
        user->name[name_len] = 0;
        if (user->type != SC_ROLE && user->type != SC_USER) {
                tnt_raise(ClientError, ER_CREATE_USER,
                          user->name, "unknown user type");
        }
        identifier_check_xc(user->name, name_len);
        /*
         * AUTH_DATA field in _user space should contain
         * chap-sha1 -> base64_encode(sha1(sha1(password), 0).
         * Check for trivial errors when a plain text
         * password is saved in this field instead.
         */
        if (tuple_field_count(tuple) > BOX_USER_FIELD_AUTH_MECH_LIST) {
                const char *auth_data =
                        tuple_field(tuple, BOX_USER_FIELD_AUTH_MECH_LIST);
                const char *tmp = auth_data;
                bool is_auth_empty;
                if (mp_typeof(*auth_data) == MP_ARRAY &&
                    mp_decode_array(&tmp) == 0) {
                        is_auth_empty = true;
                } else if (mp_typeof(*auth_data) == MP_MAP &&
                           mp_decode_map(&tmp) == 0) {
                        is_auth_empty = true;
                } else {
                        is_auth_empty = false;
                }
                if (!is_auth_empty && user->type == SC_ROLE)
                        tnt_raise(ClientError, ER_CREATE_ROLE, user->name,
                                  "authentication data can not be set for a "\
                                  "role");
                user_def_fill_auth_data(user, auth_data);
        }
        def_guard.is_active = false;
        return user;
}

static void
user_cache_remove_user(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct txn_stmt *stmt = txn_last_stmt(txn);
        uint32_t uid = tuple_field_u32_xc(stmt->old_tuple ?
                                       stmt->old_tuple : stmt->new_tuple,
                                       BOX_USER_FIELD_ID);
        user_cache_delete(uid);
}

static void
user_cache_alter_user(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct txn_stmt *stmt = txn_last_stmt(txn);
        struct user_def *user = user_def_new_from_tuple(stmt->new_tuple);
        auto def_guard = make_scoped_guard([=] { free(user); });
        /* Can throw if, e.g. too many users. */
        user_cache_replace(user);
        def_guard.is_active = false;
}

/**
 * A trigger invoked on replace in the user table.
 */
static void
on_replace_dd_user(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct txn_stmt *stmt = txn_current_stmt(txn);
        txn_check_singlestatement_xc(txn, "Space _user");
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;

        uint32_t uid = tuple_field_u32_xc(old_tuple ? old_tuple : new_tuple,
                                          BOX_USER_FIELD_ID);
        struct user *old_user = user_by_id(uid);
        if (new_tuple != NULL && old_user == NULL) { /* INSERT */
                struct user_def *user = user_def_new_from_tuple(new_tuple);
                access_check_ddl(user->name, user->uid, user->owner, user->type,
                                 PRIV_C, true);
                auto def_guard = make_scoped_guard([=] { free(user); });
                (void) user_cache_replace(user);
                def_guard.is_active = false;
                struct trigger *on_rollback =
                        txn_alter_trigger_new(user_cache_remove_user, NULL);
                txn_on_rollback(txn, on_rollback);
        } else if (new_tuple == NULL) { /* DELETE */
                access_check_ddl(old_user->def->name, old_user->def->uid,
                                 old_user->def->owner, old_user->def->type,
                                 PRIV_D, true);
                /* Can't drop guest or super user */
                if (uid <= (uint32_t) BOX_SYSTEM_USER_ID_MAX || uid == SUPER) {
                        tnt_raise(ClientError, ER_DROP_USER,
                                  old_user->def->name,
                                  "the user or the role is a system");
                }
                /*
                 * Can only delete user if it has no spaces,
                 * no functions and no grants.
                 */
                if (user_has_data(old_user)) {
                        tnt_raise(ClientError, ER_DROP_USER,
                                  old_user->def->name, "the user has objects");
                }
                struct trigger *on_commit =
                        txn_alter_trigger_new(user_cache_remove_user, NULL);
                txn_on_commit(txn, on_commit);
        } else { /* UPDATE, REPLACE */
                assert(old_user != NULL && new_tuple != NULL);
                /*
                 * Allow change of user properties (name,
                 * password) but first check that the change is
                 * correct.
                 */
                struct user_def *user = user_def_new_from_tuple(new_tuple);
                access_check_ddl(user->name, user->uid, user->uid,
                                 old_user->def->type, PRIV_A, true);
                auto def_guard = make_scoped_guard([=] { free(user); });
                struct trigger *on_commit =
                        txn_alter_trigger_new(user_cache_alter_user, NULL);
                txn_on_commit(txn, on_commit);
        }
}

/**
 * Get function identifiers from a tuple.
 *
 * @param tuple Tuple to get ids from.
 * @param[out] fid Function identifier.
 * @param[out] uid Owner identifier.
 */
static inline void
func_def_get_ids_from_tuple(const struct tuple *tuple, uint32_t *fid,
                            uint32_t *uid)
{
        *fid = tuple_field_u32_xc(tuple, BOX_FUNC_FIELD_ID);
        *uid = tuple_field_u32_xc(tuple, BOX_FUNC_FIELD_UID);
}

/** Create a function definition from tuple. */
static struct func_def *
func_def_new_from_tuple(const struct tuple *tuple)
{
        uint32_t len;
        const char *name = tuple_field_str_xc(tuple, BOX_FUNC_FIELD_NAME,
                                              &len);
        if (len > BOX_NAME_MAX)
                tnt_raise(ClientError, ER_CREATE_FUNCTION,
                          tt_cstr(name, BOX_INVALID_NAME_MAX),
                          "function name is too long");
        identifier_check_xc(name, len);
        struct func_def *def = (struct func_def *) malloc(func_def_sizeof(len));
        if (def == NULL)
                tnt_raise(OutOfMemory, func_def_sizeof(len), "malloc", "def");
        auto def_guard = make_scoped_guard([=] { free(def); });
        func_def_get_ids_from_tuple(tuple, &def->fid, &def->uid);
        memcpy(def->name, name, len);
        def->name[len] = 0;
        if (tuple_field_count(tuple) > BOX_FUNC_FIELD_SETUID)
                def->setuid = tuple_field_u32_xc(tuple, BOX_FUNC_FIELD_SETUID);
        else
                def->setuid = false;
        if (tuple_field_count(tuple) > BOX_FUNC_FIELD_LANGUAGE) {
                const char *language =
                        tuple_field_cstr_xc(tuple, BOX_FUNC_FIELD_LANGUAGE);
                def->language = STR2ENUM(func_language, language);
                if (def->language == func_language_MAX) {
                        tnt_raise(ClientError, ER_FUNCTION_LANGUAGE,
                                  language, def->name);
                }
        } else {
                /* Lua is the default. */
                def->language = FUNC_LANGUAGE_LUA;
        }
        def_guard.is_active = false;
        return def;
}

/** Remove a function from function cache */
static void
func_cache_remove_func(struct trigger * /* trigger */, void *event)
{
        struct txn_stmt *stmt = txn_last_stmt((struct txn *) event);
        uint32_t fid = tuple_field_u32_xc(stmt->old_tuple ?
                                       stmt->old_tuple : stmt->new_tuple,
                                       BOX_FUNC_FIELD_ID);
        func_cache_delete(fid);
}

/** Replace a function in the function cache */
static void
func_cache_replace_func(struct trigger * /* trigger */, void *event)
{
        struct txn_stmt *stmt = txn_last_stmt((struct txn*) event);
        struct func_def *def = func_def_new_from_tuple(stmt->new_tuple);
        auto def_guard = make_scoped_guard([=] { free(def); });
        func_cache_replace(def);
        def_guard.is_active = false;
}

/**
 * A trigger invoked on replace in a space containing
 * functions on which there were defined any grants.
 */
static void
on_replace_dd_func(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        txn_check_singlestatement_xc(txn, "Space _func");
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;

        uint32_t fid = tuple_field_u32_xc(old_tuple ? old_tuple : new_tuple,
                                          BOX_FUNC_FIELD_ID);
        struct func *old_func = func_by_id(fid);
        if (new_tuple != NULL && old_func == NULL) { /* INSERT */
                struct func_def *def = func_def_new_from_tuple(new_tuple);
                access_check_ddl(def->name, def->fid, def->uid, SC_FUNCTION,
                                 PRIV_C, true);
                auto def_guard = make_scoped_guard([=] { free(def); });
                func_cache_replace(def);
                def_guard.is_active = false;
                struct trigger *on_rollback =
                        txn_alter_trigger_new(func_cache_remove_func, NULL);
                txn_on_rollback(txn, on_rollback);
        } else if (new_tuple == NULL) {         /* DELETE */
                uint32_t uid;
                func_def_get_ids_from_tuple(old_tuple, &fid, &uid);
                /*
                 * Can only delete func if you're the one
                 * who created it or a superuser.
                 */
                access_check_ddl(old_func->def->name, fid, uid, SC_FUNCTION,
                                 PRIV_D, true);
                /* Can only delete func if it has no grants. */
                if (schema_find_grants("function", old_func->def->fid)) {
                        tnt_raise(ClientError, ER_DROP_FUNCTION,
                                  (unsigned) old_func->def->uid,
                                  "function has grants");
                }
                struct trigger *on_commit =
                        txn_alter_trigger_new(func_cache_remove_func, NULL);
                txn_on_commit(txn, on_commit);
        } else {                                /* UPDATE, REPLACE */
                struct func_def *def = func_def_new_from_tuple(new_tuple);
                auto def_guard = make_scoped_guard([=] { free(def); });
                access_check_ddl(def->name, def->fid, def->uid, SC_FUNCTION,
                                 PRIV_A, true);
                struct trigger *on_commit =
                        txn_alter_trigger_new(func_cache_replace_func, NULL);
                txn_on_commit(txn, on_commit);
        }
}

/** Create a collation identifier definition from tuple. */
void
coll_id_def_new_from_tuple(const struct tuple *tuple, struct coll_id_def *def)
{
        memset(def, 0, sizeof(*def));
        uint32_t name_len, locale_len, type_len;
        def->id = tuple_field_u32_xc(tuple, BOX_COLLATION_FIELD_ID);
        def->name = tuple_field_str_xc(tuple, BOX_COLLATION_FIELD_NAME, &name_len);
        def->name_len = name_len;
        if (name_len > BOX_NAME_MAX)
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "collation name is too long");
        identifier_check_xc(def->name, name_len);

        def->owner_id = tuple_field_u32_xc(tuple, BOX_COLLATION_FIELD_UID);
        struct coll_def *base = &def->base;
        const char *type = tuple_field_str_xc(tuple, BOX_COLLATION_FIELD_TYPE,
                                              &type_len);
        base->type = STRN2ENUM(coll_type, type, type_len);
        if (base->type == coll_type_MAX)
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "unknown collation type");
        const char *locale =
                tuple_field_str_xc(tuple, BOX_COLLATION_FIELD_LOCALE,
                                   &locale_len);
        if (locale_len > COLL_LOCALE_LEN_MAX)
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "collation locale is too long");
        if (locale_len > 0)
                identifier_check_xc(locale, locale_len);
        snprintf(base->locale, sizeof(base->locale), "%.*s", locale_len,
                 locale);
        const char *options =
                tuple_field_with_type_xc(tuple, BOX_COLLATION_FIELD_OPTIONS,
                                         MP_MAP);

        assert(base->type == COLL_TYPE_ICU);
        if (opts_decode(&base->icu, coll_icu_opts_reg, &options,
                        ER_WRONG_COLLATION_OPTIONS,
                        BOX_COLLATION_FIELD_OPTIONS, NULL) != 0)
                diag_raise();

        if (base->icu.french_collation == coll_icu_on_off_MAX) {
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "ICU wrong french_collation option setting, "
                                  "expected ON | OFF");
        }

        if (base->icu.alternate_handling == coll_icu_alternate_handling_MAX) {
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "ICU wrong alternate_handling option setting, "
                                  "expected NON_IGNORABLE | SHIFTED");
        }

        if (base->icu.case_first == coll_icu_case_first_MAX) {
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "ICU wrong case_first option setting, "
                                  "expected OFF | UPPER_FIRST | LOWER_FIRST");
        }

        if (base->icu.case_level == coll_icu_on_off_MAX) {
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "ICU wrong case_level option setting, "
                                  "expected ON | OFF");
        }

        if (base->icu.normalization_mode == coll_icu_on_off_MAX) {
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "ICU wrong normalization_mode option setting, "
                                  "expected ON | OFF");
        }

        if (base->icu.strength == coll_icu_strength_MAX) {
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "ICU wrong strength option setting, "
                                  "expected PRIMARY | SECONDARY | "
                                  "TERTIARY | QUATERNARY | IDENTICAL");
        }

        if (base->icu.numeric_collation == coll_icu_on_off_MAX) {
                tnt_raise(ClientError, ER_CANT_CREATE_COLLATION,
                          "ICU wrong numeric_collation option setting, "
                                  "expected ON | OFF");
        }
}

/**
 * Rollback a change in collation space.
 * A change is only INSERT or DELETE, UPDATE is not supported.
 */
static void
coll_id_cache_rollback(struct trigger *trigger, void *event)
{
        struct coll_id *coll_id = (struct coll_id *) trigger->data;
        struct txn_stmt *stmt = txn_last_stmt((struct txn*) event);

        if (stmt->new_tuple == NULL) {
                /* DELETE: put the collation identifier back. */
                assert(stmt->old_tuple != NULL);
                struct coll_id *replaced_id;
                if (coll_id_cache_replace(coll_id, &replaced_id) != 0) {
                        panic("Out of memory on insertion into collation "\
                              "cache");
                }
                assert(replaced_id == NULL);
        } else {
                /* INSERT: delete the new collation identifier. */
                assert(stmt->old_tuple == NULL);
                coll_id_cache_delete(coll_id);
                coll_id_delete(coll_id);
        }
}


/** Free a deleted collation identifier on commit. */
static void
coll_id_cache_commit(struct trigger *trigger, void *event)
{
        (void) event;
        struct coll_id *coll_id = (struct coll_id *) trigger->data;
        coll_id_delete(coll_id);
}

/**
 * A trigger invoked on replace in a space containing
 * collations that a user defined.
 */
static void
on_replace_dd_collation(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;
        txn_check_singlestatement_xc(txn, "Space _collation");
        struct trigger *on_rollback =
                txn_alter_trigger_new(coll_id_cache_rollback, NULL);
        struct trigger *on_commit =
                txn_alter_trigger_new(coll_id_cache_commit, NULL);
        if (new_tuple == NULL && old_tuple != NULL) {
                /* DELETE */
                /*
                 * TODO: Check that no index uses the collation
                 * identifier.
                 */
                int32_t old_id = tuple_field_u32_xc(old_tuple,
                                                    BOX_COLLATION_FIELD_ID);
                struct coll_id *old_coll_id = coll_by_id(old_id);
                assert(old_coll_id != NULL);
                access_check_ddl(old_coll_id->name, old_coll_id->id,
                                 old_coll_id->owner_id, SC_COLLATION,
                                 PRIV_D, false);
                /*
                 * Set on_commit/on_rollback triggers after
                 * deletion from the cache to make trigger logic
                 * simple.
                 */
                coll_id_cache_delete(old_coll_id);
                on_rollback->data = old_coll_id;
                on_commit->data = old_coll_id;
                txn_on_rollback(txn, on_rollback);
                txn_on_commit(txn, on_commit);
        } else if (new_tuple != NULL && old_tuple == NULL) {
                /* INSERT */
                struct coll_id_def new_def;
                coll_id_def_new_from_tuple(new_tuple, &new_def);
                access_check_ddl(new_def.name, new_def.id, new_def.owner_id,
                                 SC_COLLATION, PRIV_C, false);
                struct coll_id *new_coll_id = coll_id_new(&new_def);
                if (new_coll_id == NULL)
                        diag_raise();
                struct coll_id *replaced_id;
                if (coll_id_cache_replace(new_coll_id, &replaced_id) != 0) {
                        coll_id_delete(new_coll_id);
                        diag_raise();
                }
                assert(replaced_id == NULL);
                on_rollback->data = new_coll_id;
                txn_on_rollback(txn, on_rollback);
        } else {
                /* UPDATE */
                assert(new_tuple != NULL && old_tuple != NULL);
                tnt_raise(ClientError, ER_UNSUPPORTED, "collation", "alter");
        }
}

/**
 * Create a privilege definition from tuple.
 */
void
priv_def_create_from_tuple(struct priv_def *priv, struct tuple *tuple)
{
        priv->grantor_id = tuple_field_u32_xc(tuple, BOX_PRIV_FIELD_ID);
        priv->grantee_id = tuple_field_u32_xc(tuple, BOX_PRIV_FIELD_UID);

        const char *object_type =
                tuple_field_cstr_xc(tuple, BOX_PRIV_FIELD_OBJECT_TYPE);
        priv->object_type = schema_object_type(object_type);

        const char *data = tuple_field(tuple, BOX_PRIV_FIELD_OBJECT_ID);
        if (data == NULL) {
                tnt_raise(ClientError, ER_NO_SUCH_FIELD,
                          BOX_PRIV_FIELD_OBJECT_ID + TUPLE_INDEX_BASE);
        }
        /*
         * When granting or revoking privileges on a whole entity
         * we pass empty string ('') to object_id to indicate
         * grant on every object of that entity.
         * So check for that first.
         */
        switch (mp_typeof(*data)) {
        case MP_STR:
                if (mp_decode_strl(&data) == 0) {
                        /* Entity-wide privilege. */
                        priv->object_id = 0;
                        priv->object_type = schema_entity_type(priv->object_type);
                        break;
                }
                FALLTHROUGH;
        default:
                priv->object_id = tuple_field_u32_xc(tuple,
                                                     BOX_PRIV_FIELD_OBJECT_ID);
        }
        if (priv->object_type == SC_UNKNOWN) {
                tnt_raise(ClientError, ER_UNKNOWN_SCHEMA_OBJECT,
                          object_type);
        }
        priv->access = tuple_field_u32_xc(tuple, BOX_PRIV_FIELD_ACCESS);
}

/*
 * This function checks that:
 * - a privilege is granted from an existing user to an existing
 *   user on an existing object
 * - the grantor has the right to grant (is the owner of the object)
 *
 * @XXX Potentially there is a race in case of rollback, since an
 * object can be changed during WAL write.
 * In the future we must protect grant/revoke with a logical lock.
 */
static void
priv_def_check(struct priv_def *priv, enum priv_type priv_type)
{
        struct user *grantor = user_find_xc(priv->grantor_id);
        /* May be a role */
        struct user *grantee = user_by_id(priv->grantee_id);
        if (grantee == NULL) {
                tnt_raise(ClientError, ER_NO_SUCH_USER,
                          int2str(priv->grantee_id));
        }
        const char *name = schema_find_name(priv->object_type, priv->object_id);
        access_check_ddl(name, priv->object_id, grantor->def->uid,
                         priv->object_type, priv_type, false);
        switch (priv->object_type) {
        case SC_UNIVERSE:
                if (grantor->def->uid != ADMIN) {
                        tnt_raise(AccessDeniedError,
                                  priv_name(priv_type),
                                  schema_object_name(SC_UNIVERSE),
                                  name,
                                  grantor->def->name);
                }
                break;
        case SC_SPACE:
        {
                struct space *space = space_cache_find_xc(priv->object_id);
                if (space->def->uid != grantor->def->uid &&
                    grantor->def->uid != ADMIN) {
                        tnt_raise(AccessDeniedError,
                                  priv_name(priv_type),
                                  schema_object_name(SC_SPACE), name,
                                  grantor->def->name);
                }
                break;
        }
        case SC_FUNCTION:
        {
                struct func *func = func_cache_find(priv->object_id);
                if (func->def->uid != grantor->def->uid &&
                    grantor->def->uid != ADMIN) {
                        tnt_raise(AccessDeniedError,
                                  priv_name(priv_type),
                                  schema_object_name(SC_FUNCTION), name,
                                  grantor->def->name);
                }
                break;
        }
        case SC_SEQUENCE:
        {
                struct sequence *seq = sequence_cache_find(priv->object_id);
                if (seq->def->uid != grantor->def->uid &&
                    grantor->def->uid != ADMIN) {
                        tnt_raise(AccessDeniedError,
                                  priv_name(priv_type),
                                  schema_object_name(SC_SEQUENCE), name,
                                  grantor->def->name);
                }
                break;
        }
        case SC_ROLE:
        {
                struct user *role = user_by_id(priv->object_id);
                if (role == NULL || role->def->type != SC_ROLE) {
                        tnt_raise(ClientError, ER_NO_SUCH_ROLE,
                                  role ? role->def->name :
                                  int2str(priv->object_id));
                }
                /*
                 * Only the creator of the role can grant or revoke it.
                 * Everyone can grant 'PUBLIC' role.
                 */
                if (role->def->owner != grantor->def->uid &&
                    grantor->def->uid != ADMIN &&
                    (role->def->uid != PUBLIC || priv->access != PRIV_X)) {
                        tnt_raise(AccessDeniedError,
                                  priv_name(priv_type),
                                  schema_object_name(SC_ROLE), name,
                                  grantor->def->name);
                }
                /* Not necessary to do during revoke, but who cares. */
                role_check(grantee, role);
                break;
        }
        case SC_USER:
        {
                struct user *user = user_by_id(priv->object_id);
                if (user == NULL || user->def->type != SC_USER) {
                        tnt_raise(ClientError, ER_NO_SUCH_USER,
                                  user ? user->def->name :
                                  int2str(priv->object_id));
                }
                if (user->def->owner != grantor->def->uid &&
                    grantor->def->uid != ADMIN) {
                        tnt_raise(AccessDeniedError,
                                  priv_name(priv_type),
                                  schema_object_name(SC_USER), name,
                                  grantor->def->name);
                }
                break;
        }
        case SC_ENTITY_SPACE:
        case SC_ENTITY_FUNCTION:
        case SC_ENTITY_SEQUENCE:
        case SC_ENTITY_ROLE:
        case SC_ENTITY_USER:
        {
                /* Only admin may grant privileges on an entire entity. */
                if (grantor->def->uid != ADMIN) {
                        tnt_raise(AccessDeniedError, priv_name(priv_type),
                                  schema_object_name(priv->object_type), name,
                                  grantor->def->name);
                }
        }
        default:
                break;
        }
        if (priv->access == 0) {
                tnt_raise(ClientError, ER_GRANT,
                          "the grant tuple has no privileges");
        }
}

/**
 * Update a metadata cache object with the new access
 * data.
 */
static void
grant_or_revoke(struct priv_def *priv)
{
        struct user *grantee = user_by_id(priv->grantee_id);
        if (grantee == NULL)
                return;
        /*
         * Grant a role to a user only when privilege type is 'execute'
         * and the role is specified.
         */
        if (priv->object_type == SC_ROLE && !(priv->access & ~PRIV_X)) {
                struct user *role = user_by_id(priv->object_id);
                if (role == NULL || role->def->type != SC_ROLE)
                        return;
                if (priv->access)
                        role_grant(grantee, role);
                else
                        role_revoke(grantee, role);
        } else {
                priv_grant(grantee, priv);
        }
}

/** A trigger called on rollback of grant, or on commit of revoke. */
static void
revoke_priv(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct txn_stmt *stmt = txn_last_stmt(txn);
        struct tuple *tuple = (stmt->new_tuple ?
                               stmt->new_tuple : stmt->old_tuple);
        struct priv_def priv;
        priv_def_create_from_tuple(&priv, tuple);
        /*
         * Access to the object has been removed altogether so
         * there should be no grants at all. If only some grants
         * were removed, modify_priv trigger would have been
         * invoked.
         */
        priv.access = 0;
        grant_or_revoke(&priv);
}

/** A trigger called on rollback of grant, or on commit of revoke. */
static void
modify_priv(struct trigger * /* trigger */, void *event)
{
        struct txn_stmt *stmt = txn_last_stmt((struct txn *) event);
        struct priv_def priv;
        priv_def_create_from_tuple(&priv, stmt->new_tuple);
        grant_or_revoke(&priv);
}

/**
 * A trigger invoked on replace in the space containing
 * all granted privileges.
 */
static void
on_replace_dd_priv(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        txn_check_singlestatement_xc(txn, "Space _priv");
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;
        struct priv_def priv;

        if (new_tuple != NULL && old_tuple == NULL) {        /* grant */
                priv_def_create_from_tuple(&priv, new_tuple);
                /*
                 * Add system privileges explicitly to the
                 * universe grant issued prior to 1.7.7 in
                 * case upgrade script has not been invoked.
                 */
                if (priv.object_type == SC_UNIVERSE &&
                    dd_version_id < version_id(1, 7, 7)) {

                        priv.access |= PRIV_S;
                        priv.access |= PRIV_U;
                        /*
                         * For admin we have to set his privileges
                         * explicitly because he needs them in upgrade and
                         * bootstrap script
                         */
                        if (priv.grantor_id == ADMIN) {
                                priv.access = admin_credentials.universal_access;
                        }
                }
                priv_def_check(&priv, PRIV_GRANT);
                grant_or_revoke(&priv);
                struct trigger *on_rollback =
                        txn_alter_trigger_new(revoke_priv, NULL);
                txn_on_rollback(txn, on_rollback);
        } else if (new_tuple == NULL) {                /* revoke */
                assert(old_tuple);
                priv_def_create_from_tuple(&priv, old_tuple);
                priv_def_check(&priv, PRIV_REVOKE);
                struct trigger *on_commit =
                        txn_alter_trigger_new(revoke_priv, NULL);
                txn_on_commit(txn, on_commit);
        } else {                                       /* modify */
                priv_def_create_from_tuple(&priv, new_tuple);
                priv_def_check(&priv, PRIV_GRANT);
                struct trigger *on_commit =
                        txn_alter_trigger_new(modify_priv, NULL);
                txn_on_commit(txn, on_commit);
        }
}

/* }}} access control */

/* {{{ cluster configuration */

/**
 * This trigger is invoked only upon initial recovery, when
 * reading contents of the system spaces from the snapshot.
 *
 * Before a cluster is assigned a cluster id it's read only.
 * Since during recovery state of the WAL doesn't
 * concern us, we can safely change the cluster id in before-replace
 * event, not in after-replace event.
 */
static void
on_replace_dd_schema(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        txn_check_singlestatement_xc(txn, "Space _schema");
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;
        const char *key = tuple_field_cstr_xc(new_tuple ? new_tuple : old_tuple,
                                              BOX_SCHEMA_FIELD_KEY);
        if (strcmp(key, "cluster") == 0) {
                if (new_tuple == NULL)
                        tnt_raise(ClientError, ER_REPLICASET_UUID_IS_RO);
                tt_uuid uu;
                tuple_field_uuid_xc(new_tuple, BOX_CLUSTER_FIELD_UUID, &uu);
                REPLICASET_UUID = uu;
        } else if (strcmp(key, "version") == 0) {
                if (new_tuple != NULL) {
                        uint32_t major, minor, patch;
                        if (tuple_field_u32(new_tuple, 1, &major) != 0 ||
                            tuple_field_u32(new_tuple, 2, &minor) != 0)
                                tnt_raise(ClientError, ER_WRONG_DD_VERSION);
                        /* Version can be major.minor with no patch. */
                        if (tuple_field_u32(new_tuple, 3, &patch) != 0)
                                patch = 0;
                        dd_version_id = version_id(major, minor, patch);
                } else {
                        assert(old_tuple != NULL);
                        /*
                         * _schema:delete({'version'}) for
                         * example, for box.internal.bootstrap().
                         */
                        dd_version_id = tarantool_version_id();
                }
        }
}

/**
 * A record with id of the new instance has been synced to the
 * write ahead log. Update the cluster configuration cache
 * with it.
 */
static void
on_commit_dd_cluster(struct trigger *trigger, void *event)
{
        (void) trigger;
        struct txn_stmt *stmt = txn_last_stmt((struct txn *) event);
        struct tuple *new_tuple = stmt->new_tuple;
        struct tuple *old_tuple = stmt->old_tuple;

        if (new_tuple == NULL) {
                struct tt_uuid old_uuid;
                tuple_field_uuid_xc(stmt->old_tuple, BOX_CLUSTER_FIELD_UUID,
                                    &old_uuid);
                struct replica *replica = replica_by_uuid(&old_uuid);
                assert(replica != NULL);
                replica_clear_id(replica);
                return;
        } else if (old_tuple != NULL) {
                return; /* nothing to change */
        }

        uint32_t id = tuple_field_u32_xc(new_tuple, BOX_CLUSTER_FIELD_ID);
        tt_uuid uuid;
        tuple_field_uuid_xc(new_tuple, BOX_CLUSTER_FIELD_UUID, &uuid);
        struct replica *replica = replica_by_uuid(&uuid);
        if (replica != NULL) {
                replica_set_id(replica, id);
        } else {
                try {
                        replica = replicaset_add(id, &uuid);
                        /* Can't throw exceptions from on_commit trigger */
                } catch(Exception *e) {
                        panic("Can't register replica: %s", e->errmsg);
                }
        }
}

/**
 * A trigger invoked on replace in the space _cluster,
 * which contains cluster configuration.
 *
 * This space is modified by JOIN command in IPROTO
 * protocol.
 *
 * The trigger updates the cluster configuration cache
 * with uuid of the newly joined instance.
 *
 * During recovery, it acts the same way, loading identifiers
 * of all instances into the cache. Instance globally unique
 * identifiers are used to keep track of cluster configuration,
 * so that a replica that previously joined a replica set can
 * follow updates, and a replica that belongs to a different
 * replica set can not by mistake join/follow another replica
 * set without first being reset (emptied).
 */
static void
on_replace_dd_cluster(struct trigger *trigger, void *event)
{
        (void) trigger;
        struct txn *txn = (struct txn *) event;
        txn_check_singlestatement_xc(txn, "Space _cluster");
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;
        if (new_tuple != NULL) { /* Insert or replace */
                /* Check fields */
                uint32_t replica_id =
                        tuple_field_u32_xc(new_tuple, BOX_CLUSTER_FIELD_ID);
                replica_check_id(replica_id);
                tt_uuid replica_uuid;
                tuple_field_uuid_xc(new_tuple, BOX_CLUSTER_FIELD_UUID,
                                    &replica_uuid);
                if (tt_uuid_is_nil(&replica_uuid))
                        tnt_raise(ClientError, ER_INVALID_UUID,
                                  tt_uuid_str(&replica_uuid));
                if (old_tuple != NULL) {
                        /*
                         * Forbid changes of UUID for a registered instance:
                         * it requires an extra effort to keep _cluster
                         * in sync with appliers and relays.
                         */
                        tt_uuid old_uuid;
                        tuple_field_uuid_xc(old_tuple, BOX_CLUSTER_FIELD_UUID,
                                            &old_uuid);
                        if (!tt_uuid_is_equal(&replica_uuid, &old_uuid)) {
                                tnt_raise(ClientError, ER_UNSUPPORTED,
                                          "Space _cluster",
                                          "updates of instance uuid");
                        }
                }
        } else {
                /*
                 * Don't allow deletion of the record for this instance
                 * from _cluster.
                 */
                assert(old_tuple != NULL);
                uint32_t replica_id =
                        tuple_field_u32_xc(old_tuple, BOX_CLUSTER_FIELD_ID);
                replica_check_id(replica_id);
        }

        struct trigger *on_commit =
                        txn_alter_trigger_new(on_commit_dd_cluster, NULL);
        txn_on_commit(txn, on_commit);
}

/* }}} cluster configuration */

/* {{{ sequence */

/** Create a sequence definition from a tuple. */
static struct sequence_def *
sequence_def_new_from_tuple(struct tuple *tuple, uint32_t errcode)
{
        uint32_t name_len;
        const char *name = tuple_field_str_xc(tuple, BOX_USER_FIELD_NAME,
                                              &name_len);
        if (name_len > BOX_NAME_MAX) {
                tnt_raise(ClientError, errcode,
                          tt_cstr(name, BOX_INVALID_NAME_MAX),
                          "sequence name is too long");
        }
        identifier_check_xc(name, name_len);
        size_t sz = sequence_def_sizeof(name_len);
        struct sequence_def *def = (struct sequence_def *) malloc(sz);
        if (def == NULL)
                tnt_raise(OutOfMemory, sz, "malloc", "sequence");
        auto def_guard = make_scoped_guard([=] { free(def); });
        memcpy(def->name, name, name_len);
        def->name[name_len] = '\0';
        def->id = tuple_field_u32_xc(tuple, BOX_SEQUENCE_FIELD_ID);
        def->uid = tuple_field_u32_xc(tuple, BOX_SEQUENCE_FIELD_UID);
        def->step = tuple_field_i64_xc(tuple, BOX_SEQUENCE_FIELD_STEP);
        def->min = tuple_field_i64_xc(tuple, BOX_SEQUENCE_FIELD_MIN);
        def->max = tuple_field_i64_xc(tuple, BOX_SEQUENCE_FIELD_MAX);
        def->start = tuple_field_i64_xc(tuple, BOX_SEQUENCE_FIELD_START);
        def->cache = tuple_field_i64_xc(tuple, BOX_SEQUENCE_FIELD_CACHE);
        def->cycle = tuple_field_bool_xc(tuple, BOX_SEQUENCE_FIELD_CYCLE);
        if (def->step == 0)
                tnt_raise(ClientError, errcode, def->name,
                          "step option must be non-zero");
        if (def->min > def->max)
                tnt_raise(ClientError, errcode, def->name,
                          "max must be greater than or equal to min");
        if (def->start < def->min || def->start > def->max)
                tnt_raise(ClientError, errcode, def->name,
                          "start must be between min and max");
        def_guard.is_active = false;
        return def;
}

/** Argument passed to on_commit_dd_sequence() trigger. */
struct alter_sequence {
        /** Trigger invoked on commit in the _sequence space. */
        struct trigger on_commit;
        /** Trigger invoked on rollback in the _sequence space. */
        struct trigger on_rollback;
        /** Old sequence definition or NULL if create. */
        struct sequence_def *old_def;
        /** New sequence defitition or NULL if drop. */
        struct sequence_def *new_def;
};

/**
 * Trigger invoked on commit in the _sequence space.
 */
static void
on_commit_dd_sequence(struct trigger *trigger, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct alter_sequence *alter = (struct alter_sequence *) trigger->data;

        if (alter->new_def != NULL && alter->old_def != NULL) {
                /* Alter a sequence. */
                sequence_cache_replace(alter->new_def);
        } else if (alter->new_def == NULL) {
                /* Drop a sequence. */
                sequence_cache_delete(alter->old_def->id);
        }

        trigger_run_xc(&on_alter_sequence, txn_last_stmt(txn));
}

/**
 * Trigger invoked on rollback in the _sequence space.
 */
static void
on_rollback_dd_sequence(struct trigger *trigger, void * /* event */)
{
        struct alter_sequence *alter = (struct alter_sequence *) trigger->data;

        if (alter->new_def != NULL && alter->old_def == NULL) {
                /* Rollback creation of a sequence. */
                sequence_cache_delete(alter->new_def->id);
        }
}

/**
 * A trigger invoked on replace in space _sequence.
 * Used to alter a sequence definition.
 */
static void
on_replace_dd_sequence(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        txn_check_singlestatement_xc(txn, "Space _sequence");
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;

        struct alter_sequence *alter =
                region_calloc_object_xc(&fiber()->gc, struct alter_sequence);

        struct sequence_def *new_def = NULL;
        auto def_guard = make_scoped_guard([=] { free(new_def); });

        if (old_tuple == NULL && new_tuple != NULL) {                /* INSERT */
                new_def = sequence_def_new_from_tuple(new_tuple,
                                                      ER_CREATE_SEQUENCE);
                assert(sequence_by_id(new_def->id) == NULL);
                access_check_ddl(new_def->name, new_def->id, new_def->uid,
                                 SC_SEQUENCE, PRIV_C, false);
                sequence_cache_replace(new_def);
                alter->new_def = new_def;
        } else if (old_tuple != NULL && new_tuple == NULL) {        /* DELETE */
                uint32_t id = tuple_field_u32_xc(old_tuple,
                                                 BOX_SEQUENCE_DATA_FIELD_ID);
                struct sequence *seq = sequence_by_id(id);
                assert(seq != NULL);
                access_check_ddl(seq->def->name, seq->def->id, seq->def->uid,
                                 SC_SEQUENCE, PRIV_D, false);
                if (space_has_data(BOX_SEQUENCE_DATA_ID, 0, id))
                        tnt_raise(ClientError, ER_DROP_SEQUENCE,
                                  seq->def->name, "the sequence has data");
                if (space_has_data(BOX_SPACE_SEQUENCE_ID, 1, id))
                        tnt_raise(ClientError, ER_DROP_SEQUENCE,
                                  seq->def->name, "the sequence is in use");
                if (schema_find_grants("sequence", seq->def->id))
                        tnt_raise(ClientError, ER_DROP_SEQUENCE,
                                  seq->def->name, "the sequence has grants");
                alter->old_def = seq->def;
        } else {                                                /* UPDATE */
                new_def = sequence_def_new_from_tuple(new_tuple,
                                                      ER_ALTER_SEQUENCE);
                struct sequence *seq = sequence_by_id(new_def->id);
                assert(seq != NULL);
                access_check_ddl(seq->def->name, seq->def->id, seq->def->uid,
                                 SC_SEQUENCE, PRIV_A, false);
                alter->old_def = seq->def;
                alter->new_def = new_def;
        }

        def_guard.is_active = false;

        trigger_create(&alter->on_commit,
                       on_commit_dd_sequence, alter, NULL);
        txn_on_commit(txn, &alter->on_commit);
        trigger_create(&alter->on_rollback,
                       on_rollback_dd_sequence, alter, NULL);
        txn_on_rollback(txn, &alter->on_rollback);
}

/**
 * A trigger invoked on replace in space _sequence_data.
 * Used to update a sequence value.
 */
static void
on_replace_dd_sequence_data(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *old_tuple = stmt->old_tuple;
        struct tuple *new_tuple = stmt->new_tuple;

        uint32_t id = tuple_field_u32_xc(old_tuple ?: new_tuple,
                                         BOX_SEQUENCE_DATA_FIELD_ID);
        struct sequence *seq = sequence_cache_find(id);
        if (seq == NULL)
                diag_raise();
        if (new_tuple != NULL) {                        /* INSERT, UPDATE */
                int64_t value = tuple_field_i64_xc(new_tuple,
                                BOX_SEQUENCE_DATA_FIELD_VALUE);
                if (sequence_set(seq, value) != 0)
                        diag_raise();
        } else {                                        /* DELETE */
                sequence_reset(seq);
        }
}

/**
 * Run the triggers registered on commit of a change in _space.
 */
static void
on_commit_dd_space_sequence(struct trigger *trigger, void * /* event */)
{
        struct space *space = (struct space *) trigger->data;
        trigger_run_xc(&on_alter_space, space);
}

/**
 * A trigger invoked on replace in space _space_sequence.
 * Used to update space <-> sequence mapping.
 */
static void
on_replace_dd_space_sequence(struct trigger * /* trigger */, void *event)
{
        struct txn *txn = (struct txn *) event;
        txn_check_singlestatement_xc(txn, "Space _space_sequence");
        struct txn_stmt *stmt = txn_current_stmt(txn);
        struct tuple *tuple = stmt->new_tuple ? stmt->new_tuple : stmt->old_tuple;

        uint32_t space_id = tuple_field_u32_xc(tuple,
                                               BOX_SPACE_SEQUENCE_FIELD_ID);
        uint32_t sequence_id = tuple_field_u32_xc(tuple,
                                BOX_SPACE_SEQUENCE_FIELD_SEQUENCE_ID);
        bool is_generated = tuple_field_bool_xc(tuple,
                                BOX_SPACE_SEQUENCE_FIELD_IS_GENERATED);

        struct space *space = space_cache_find_xc(space_id);
        struct sequence *seq = sequence_cache_find(sequence_id);

        enum priv_type priv_type = stmt->new_tuple ? PRIV_C : PRIV_D;
        if (stmt->new_tuple && stmt->old_tuple)
                priv_type = PRIV_A;

        /* Check we have the correct access type on the sequence.  * */
        if (is_generated || !stmt->new_tuple) {
                access_check_ddl(seq->def->name, seq->def->id, seq->def->uid,
                                 SC_SEQUENCE, priv_type, false);
        } else {
                /*
                 * In case user wants to attach an existing sequence,
                 * check that it has read and write access.
                 */
                access_check_ddl(seq->def->name, seq->def->id, seq->def->uid,
                                 SC_SEQUENCE, PRIV_R, false);
                access_check_ddl(seq->def->name, seq->def->id, seq->def->uid,
                                 SC_SEQUENCE, PRIV_W, false);
        }
        /** Check we have alter access on space. */
        access_check_ddl(space->def->name, space->def->id, space->def->uid,
                         SC_SPACE, PRIV_A, false);

        struct trigger *on_commit =
                txn_alter_trigger_new(on_commit_dd_space_sequence, space);
        txn_on_commit(txn, on_commit);

        if (stmt->new_tuple != NULL) {                        /* INSERT, UPDATE */
                struct index *pk = index_find_xc(space, 0);
                index_def_check_sequence(pk->def, space_name(space));
                if (seq->is_generated) {
                        tnt_raise(ClientError, ER_ALTER_SPACE,
                                  space_name(space),
                                  "can not attach generated sequence");
                }
                seq->is_generated = is_generated;
                space->sequence = seq;
        } else {                                        /* DELETE */
                assert(space->sequence == seq);
                space->sequence = NULL;
        }
}

/* }}} sequence */

static void
unlock_after_dd(struct trigger *trigger, void *event)
{
        (void) trigger;
        (void) event;
        latch_unlock(&schema_lock);
        /*
         * There can be a some count of other latch awaiting fibers. All of
         * these fibers should continue their job before current fiber fires
         * next request. It is important especially for replication - if some
         * rows are applied out of order then lsn order will be broken. This
         * can be done with locking latch one more time - it guarantees that
         * all "queued" fibers did their job before current fiber wakes next
         * time. If there is no waiting fibers then locking will be done without
         * any yields.
         */
        latch_lock(&schema_lock);
        latch_unlock(&schema_lock);
}

static void
lock_before_dd(struct trigger *trigger, void *event)
{
        (void) trigger;
        if (fiber() == latch_owner(&schema_lock))
                return;
        struct txn *txn = (struct txn *)event;
        /*
         * This trigger is executed before any check and may yield
         * on the latch lock. But a yield in a non-autocommit
         * memtx transaction will roll it back silently, rather
         * than produce an error, which is very confusing.
         * So don't try to lock a latch if there is
         * a multi-statement transaction.
         */
        txn_check_singlestatement_xc(txn, "DDL");
        struct trigger *on_commit =
                txn_alter_trigger_new(unlock_after_dd, NULL);
        struct trigger *on_rollback =
                txn_alter_trigger_new(unlock_after_dd, NULL);
        /*
         * Setting triggers doesn't fail. Lock the latch last
         * to avoid leaking the latch in case of exception.
         */
        txn_on_commit(txn, on_commit);
        txn_on_rollback(txn, on_rollback);
        latch_lock(&schema_lock);
}

struct trigger alter_space_on_replace_space = {
        RLIST_LINK_INITIALIZER, on_replace_dd_space, NULL, NULL
};

struct trigger alter_space_on_replace_index = {
        RLIST_LINK_INITIALIZER, on_replace_dd_index, NULL, NULL
};

struct trigger on_replace_truncate = {
        RLIST_LINK_INITIALIZER, on_replace_dd_truncate, NULL, NULL
};

struct trigger on_replace_schema = {
        RLIST_LINK_INITIALIZER, on_replace_dd_schema, NULL, NULL
};

struct trigger on_replace_user = {
        RLIST_LINK_INITIALIZER, on_replace_dd_user, NULL, NULL
};

struct trigger on_replace_func = {
        RLIST_LINK_INITIALIZER, on_replace_dd_func, NULL, NULL
};

struct trigger on_replace_collation = {
        RLIST_LINK_INITIALIZER, on_replace_dd_collation, NULL, NULL
};

struct trigger on_replace_priv = {
        RLIST_LINK_INITIALIZER, on_replace_dd_priv, NULL, NULL
};

struct trigger on_replace_cluster = {
        RLIST_LINK_INITIALIZER, on_replace_dd_cluster, NULL, NULL
};

struct trigger on_replace_sequence = {
        RLIST_LINK_INITIALIZER, on_replace_dd_sequence, NULL, NULL
};

struct trigger on_replace_sequence_data = {
        RLIST_LINK_INITIALIZER, on_replace_dd_sequence_data, NULL, NULL
};

struct trigger on_replace_space_sequence = {
        RLIST_LINK_INITIALIZER, on_replace_dd_space_sequence, NULL, NULL
};

struct trigger on_stmt_begin_space = {
        RLIST_LINK_INITIALIZER, lock_before_dd, NULL, NULL
};

struct trigger on_stmt_begin_index = {
        RLIST_LINK_INITIALIZER, lock_before_dd, NULL, NULL
};

struct trigger on_stmt_begin_truncate = {
        RLIST_LINK_INITIALIZER, lock_before_dd, NULL, NULL
};

/* vim: set foldmethod=marker */

tarantool / tarantool / 15365

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