Re: [tarantool-patches] Re: [PATCH v2 2/2] lua: add key_def lua module

[Alexander Turenko <alexander.turenko@tarantool.org>]

Changes I made (squashed and force-pushed):

* Updated a comment (see below).
* Removed #4025 from commit messages as duplicate of #3398.
* Fixed the typo: `sk:pairs({1}))` -> `sk:pairs({1})`

Vladimir, can you, please, look into the patchset?

branch: kshch/gh-4025-lua-key-kef-methods

WBR, Alexander Turenko.

On Thu, Mar 28, 2019 at 12:01:24PM +0300, Kirill Shcherbatov wrote:
> I've decided do not introduce public helper box_tuple_validate_parts, just
> inline this code.
> =================================================
> 
> There are several reasons to add this module:
> 
> * Factor out key parts parsing code from the tuples merger (#3276).
> * Support comparing a tuple with a key / a tuple, support merging
>   key_defs from Lua (#3398).
> * Support extracting a key from a tuple (#4025).
> 
> The format of `parts` parameter in the `key_def.new(parts)` call is
> compatible with the following structures:
> 
> * box.space[...].index[...].parts;
> * net_box_conn.space[...].index[...].parts.
> 
> A key_def instance has the following methods:
> 
> * :extract_key(tuple)           -> key (as tuple)
> * :compare(tuple_a, tuple_b)    -> number
> * :compare_with_key(tuple, key) -> number
> * :merge(another_key_def)       -> new key_def instance
> * :totable()                    -> table
> 
> Note functions that accept tuple(s) also allow to pass Lua
> table(s) instead.
> 
> Needed for #3276.
> Fixes #3398.
> Fixes #4025.
> 
> @TarantoolBot document
> Title: lua: key_def module
> 
> See the commit message for an API reference.
> 
> Example for extract_key():
> 
> ```lua
> -- Remove values got from a secondary non-unique index.
> local key_def_lib = require('key_def')
> local s = box.schema.space.create('test')
> local pk = s:create_index('pk')
> local sk = s:create_index('test', {unique = false, parts = {
>     {2, 'number', path = 'a'}, {2, 'number', path = 'b'}}})
> s:insert{1, {a = 1, b = 1}}
> s:insert{2, {a = 1, b = 2}}
> local key_def = key_def_lib.new(pk.parts)
> for _, tuple in sk:pairs({1})) do
>     local key = key_def:extract_key(tuple)
>     pk:delete(key)
> end
> ```
> ---
>  src/CMakeLists.txt            |   1 +
>  src/box/CMakeLists.txt        |   2 +
>  src/box/lua/init.c            |   5 +
>  src/box/lua/key_def.c         | 462 ++++++++++++++++++++++++++++++++++
>  src/box/lua/key_def.h         |  63 +++++
>  src/box/lua/key_def.lua       |  19 ++
>  src/box/lua/space.cc          |  28 +--
>  test/box-tap/key_def.test.lua | 370 +++++++++++++++++++++++++++
>  8 files changed, 924 insertions(+), 26 deletions(-)
>  create mode 100644 src/box/lua/key_def.c
>  create mode 100644 src/box/lua/key_def.h
>  create mode 100644 src/box/lua/key_def.lua
>  create mode 100755 test/box-tap/key_def.test.lua
> 
> diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
> index 7c2395517..a6a18142b 100644
> --- a/src/CMakeLists.txt
> +++ b/src/CMakeLists.txt
> @@ -136,6 +136,7 @@ set(api_headers
>      ${CMAKE_SOURCE_DIR}/src/lua/string.h
>      ${CMAKE_SOURCE_DIR}/src/box/txn.h
>      ${CMAKE_SOURCE_DIR}/src/box/key_def.h
> +    ${CMAKE_SOURCE_DIR}/src/box/lua/key_def.h
>      ${CMAKE_SOURCE_DIR}/src/box/field_def.h
>      ${CMAKE_SOURCE_DIR}/src/box/tuple.h
>      ${CMAKE_SOURCE_DIR}/src/box/tuple_format.h
> diff --git a/src/box/CMakeLists.txt b/src/box/CMakeLists.txt
> index 59e91b65a..906ce22e6 100644
> --- a/src/box/CMakeLists.txt
> +++ b/src/box/CMakeLists.txt
> @@ -12,6 +12,7 @@ lua_source(lua_sources lua/net_box.lua)
>  lua_source(lua_sources lua/upgrade.lua)
>  lua_source(lua_sources lua/console.lua)
>  lua_source(lua_sources lua/xlog.lua)
> +lua_source(lua_sources lua/key_def.lua)
>  set(bin_sources)
>  bin_source(bin_sources bootstrap.snap bootstrap.h)
>  
> @@ -139,6 +140,7 @@ add_library(box STATIC
>      lua/net_box.c
>      lua/xlog.c
>      lua/sql.c
> +    lua/key_def.c
>      ${bin_sources})
>  
>  target_link_libraries(box box_error tuple stat xrow xlog vclock crc32 scramble
> diff --git a/src/box/lua/init.c b/src/box/lua/init.c
> index 744b2c895..68ef58909 100644
> --- a/src/box/lua/init.c
> +++ b/src/box/lua/init.c
> @@ -59,9 +59,11 @@
>  #include "box/lua/console.h"
>  #include "box/lua/tuple.h"
>  #include "box/lua/sql.h"
> +#include "box/lua/key_def.h"
>  
>  extern char session_lua[],
>  	tuple_lua[],
> +	key_def_lua[],
>  	schema_lua[],
>  	load_cfg_lua[],
>  	xlog_lua[],
> @@ -80,6 +82,7 @@ static const char *lua_sources[] = {
>  	"box/console", console_lua,
>  	"box/load_cfg", load_cfg_lua,
>  	"box/xlog", xlog_lua,
> +	"box/key_def", key_def_lua,
>  	NULL
>  };
>  
> @@ -312,6 +315,8 @@ box_lua_init(struct lua_State *L)
>  	lua_pop(L, 1);
>  	tarantool_lua_console_init(L);
>  	lua_pop(L, 1);
> +	luaopen_key_def(L);
> +	lua_pop(L, 1);
>  
>  	/* Load Lua extension */
>  	for (const char **s = lua_sources; *s; s += 2) {
> diff --git a/src/box/lua/key_def.c b/src/box/lua/key_def.c
> new file mode 100644
> index 000000000..4df6c2045
> --- /dev/null
> +++ b/src/box/lua/key_def.c
> @@ -0,0 +1,462 @@
> +/*
> + * Copyright 2010-2019, 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 "box/coll_id_cache.h"
> +#include "box/lua/key_def.h"
> +#include "box/tuple.h"
> +#include "diag.h"
> +#include "fiber.h"
> +#include "lua/utils.h"
> +#include "tuple.h"
> +
> +static uint32_t key_def_type_id = 0;
> +
> +/**
> + * Set key_part_def from a table on top of a Lua stack.
> + *
> + * When successful return 0, otherwise return -1 and set a diag.
> + */
> +static int
> +luaT_key_def_set_part(struct lua_State *L, struct key_part_def *parts,
> +		      int part_idx, struct region *region)
> +{
> +	struct key_part_def *part = &parts[part_idx];
> +	*part = key_part_def_default;
> +
> +	/* Set part->fieldno. */
> +	lua_pushstring(L, "fieldno");
> +	lua_gettable(L, -2);
> +	if (lua_isnil(L, -1)) {
> +		diag_set(IllegalParams, "fieldno must not be nil");
> +		return -1;
> +	}
> +	/*
> +	 * Transform one-based Lua fieldno to zero-based
> +	 * fieldno to use in key_def_new().
> +	 */
> +	part->fieldno = lua_tointeger(L, -1) - TUPLE_INDEX_BASE;
> +	lua_pop(L, 1);
> +
> +	/* Set part->type. */
> +	lua_pushstring(L, "type");
> +	lua_gettable(L, -2);
> +	if (lua_isnil(L, -1)) {
> +		diag_set(IllegalParams, "type must not be nil");
> +		return -1;
> +	}
> +	size_t type_len;
> +	const char *type_name = lua_tolstring(L, -1, &type_len);
> +	lua_pop(L, 1);
> +	part->type = field_type_by_name(type_name, type_len);
> +	switch (part->type) {
> +	case FIELD_TYPE_ANY:
> +	case FIELD_TYPE_ARRAY:
> +	case FIELD_TYPE_MAP:
> +		/* Tuple comparators don't support these types. */
> +		diag_set(IllegalParams, "Unsupported field type: %s",
> +			 type_name);
> +		return -1;
> +	case field_type_MAX:
> +		diag_set(IllegalParams, "Unknown field type: %s", type_name);
> +		return -1;
> +	default:
> +		/* Pass though. */
> +		break;
> +	}
> +
> +	/* Set part->is_nullable and part->nullable_action. */
> +	lua_pushstring(L, "is_nullable");
> +	lua_gettable(L, -2);
> +	if (!lua_isnil(L, -1) && lua_toboolean(L, -1) != 0) {
> +		part->is_nullable = true;
> +		part->nullable_action = ON_CONFLICT_ACTION_NONE;
> +	}
> +	lua_pop(L, 1);
> +
> +	/*
> +	 * Set part->coll_id using collation_id.
> +	 *
> +	 * The value will be checked in key_def_new().
> +	 */
> +	lua_pushstring(L, "collation_id");
> +	lua_gettable(L, -2);
> +	if (!lua_isnil(L, -1))
> +		part->coll_id = lua_tointeger(L, -1);
> +	lua_pop(L, 1);
> +
> +	/* Set part->coll_id using collation. */
> +	lua_pushstring(L, "collation");
> +	lua_gettable(L, -2);
> +	if (!lua_isnil(L, -1)) {
> +		/* Check for conflicting options. */
> +		if (part->coll_id != COLL_NONE) {
> +			diag_set(IllegalParams, "Conflicting options: "
> +				 "collation_id and collation");
> +			return -1;
> +		}
> +
> +		size_t coll_name_len;
> +		const char *coll_name = lua_tolstring(L, -1, &coll_name_len);
> +		struct coll_id *coll_id = coll_by_name(coll_name,
> +						       coll_name_len);
> +		if (coll_id == NULL) {
> +			diag_set(IllegalParams, "Unknown collation: \"%s\"",
> +				 coll_name);
> +			return -1;
> +		}
> +		part->coll_id = coll_id->id;
> +	}
> +	lua_pop(L, 1);
> +
> +	/* Set part->path (JSON path). */
> +	lua_pushstring(L, "path");
> +	lua_gettable(L, -2);
> +	if (!lua_isnil(L, -1)) {
> +		size_t path_len;
> +		const char *path = lua_tolstring(L, -1, &path_len);
> +		if (json_path_validate(path, path_len, TUPLE_INDEX_BASE) != 0) {
> +			diag_set(ClientError, ER_WRONG_INDEX_OPTIONS,
> +				 part_idx + TUPLE_INDEX_BASE, "invalid path");
> +			return -1;
> +		}
> +		char *tmp = region_alloc(region, path_len + 1);
> +		if (tmp == NULL) {
> +			diag_set(OutOfMemory, path_len + 1, "region", "path");
> +			return -1;
> +		}
> +		/*
> +		 * lua_tolstring() guarantees that a string have
> +		 * trailing '\0'.
> +		 */
> +		memcpy(tmp, path, path_len + 1);
> +		part->path = tmp;
> +	} else {
> +		part->path = NULL;
> +	}
> +	lua_pop(L, 1);
> +	return 0;
> +}
> +
> +void
> +lbox_push_key_part(struct lua_State *L, const struct key_part *part)
> +{
> +	lua_newtable(L);
> +
> +	lua_pushstring(L, field_type_strs[part->type]);
> +	lua_setfield(L, -2, "type");
> +
> +	lua_pushnumber(L, part->fieldno + TUPLE_INDEX_BASE);
> +	lua_setfield(L, -2, "fieldno");
> +
> +	if (part->path != NULL) {
> +		lua_pushlstring(L, part->path, part->path_len);
> +		lua_setfield(L, -2, "path");
> +	}
> +
> +	lua_pushboolean(L, key_part_is_nullable(part));
> +	lua_setfield(L, -2, "is_nullable");
> +
> +	if (part->coll_id != COLL_NONE) {
> +		struct coll_id *coll_id = coll_by_id(part->coll_id);
> +		assert(coll_id != NULL);
> +		lua_pushstring(L, coll_id->name);
> +		lua_setfield(L, -2, "collation");
> +	}
> +}
> +
> +struct key_def *
> +check_key_def(struct lua_State *L, int idx)
> +{
> +	if (lua_type(L, idx) != LUA_TCDATA)
> +		return NULL;
> +
> +	uint32_t cdata_type;
> +	struct key_def **key_def_ptr = luaL_checkcdata(L, idx, &cdata_type);
> +	if (key_def_ptr == NULL || cdata_type != key_def_type_id)
> +		return NULL;
> +	return *key_def_ptr;
> +}
> +
> +/**
> + * Free a key_def from a Lua code.
> + */
> +static int
> +lbox_key_def_gc(struct lua_State *L)
> +{
> +	struct key_def *key_def = check_key_def(L, 1);
> +	if (key_def == NULL)
> +		return 0;
> +	box_key_def_delete(key_def);
> +	return 0;
> +}
> +
> +/**
> + * Take existent tuple from LUA stack or build a new tuple with
> + * default format from table, check for compatibility with a
> + * given key_def. Take tuple reference pointer on success.
> + */

Updated a bit:

diff --git a/src/box/lua/key_def.c b/src/box/lua/key_def.c
index 4df6c2045..dc610007b 100644
--- a/src/box/lua/key_def.c
+++ b/src/box/lua/key_def.c
@@ -219,9 +219,13 @@ lbox_key_def_gc(struct lua_State *L)
 }
 
 /**
- * Take existent tuple from LUA stack or build a new tuple with
- * default format from table, check for compatibility with a
- * given key_def. Take tuple reference pointer on success.
+ * Validate a tuple from a given index on a Lua stack against
+ * a key def and return the tuple.
+ *
+ * If a table is passed instead of a tuple, create a new tuple
+ * from it.
+ *
+ * Return a refcounted tuple (either provided one or a new one).
  */
 static struct tuple *
 lbox_key_def_check_tuple(struct lua_State *L, struct key_def *key_def, int idx)

> +static struct tuple *
> +lbox_key_def_check_tuple(struct lua_State *L, struct key_def *key_def, int idx)
> +{
> +	struct tuple *tuple = luaT_istuple(L, idx);
> +	if (tuple == NULL)
> +		tuple = luaT_tuple_new(L, idx, box_tuple_format_default());
> +	if (tuple == NULL)
> +		return NULL;
> +	/* Check that tuple match with the key definition. */
> +	uint32_t min_field_count =
> +		tuple_format_min_field_count(&key_def, 1, NULL, 0);
> +	uint32_t field_count = tuple_field_count(tuple);
> +	if (field_count < min_field_count) {
> +		diag_set(ClientError, ER_NO_SUCH_FIELD_NO, field_count + 1);
> +		return NULL;
> +	}
> +	for (uint32_t idx = 0; idx < key_def->part_count; idx++) {
> +		struct key_part *part = &key_def->parts[idx];
> +		const char *field = tuple_field_by_part(tuple, part);
> +		if (field == NULL) {
> +			assert(key_def->has_optional_parts);
> +			continue;
> +		}
> +		if (key_part_validate(part->type, field, idx,
> +				      key_part_is_nullable(part)) != 0)
> +			return NULL;
> +	}
> +	tuple_ref(tuple);
> +	return tuple;
> +}
> +
> +static int
> +lbox_key_def_extract_key(struct lua_State *L)
> +{
> +	struct key_def *key_def;
> +	if (lua_gettop(L) != 2 || (key_def = check_key_def(L, 1)) == NULL)
> +		return luaL_error(L, "Usage: key_def:extract_key(tuple)");
> +
> +	struct tuple *tuple;
> +	if ((tuple = lbox_key_def_check_tuple(L, key_def, 2)) == NULL)
> +		return luaT_error(L);
> +
> +	uint32_t key_size;
> +	char *key = tuple_extract_key(tuple, key_def, &key_size);
> +	tuple_unref(tuple);
> +	if (key == NULL)
> +		return luaT_error(L);
> +
> +	struct tuple *ret =
> +		box_tuple_new(box_tuple_format_default(), key, key + key_size);
> +	if (ret == NULL)
> +		return luaT_error(L);
> +	luaT_pushtuple(L, ret);
> +	return 1;
> +}
> +
> +static int
> +lbox_key_def_compare(struct lua_State *L)
> +{
> +	struct key_def *key_def;
> +	if (lua_gettop(L) != 3 || (key_def = check_key_def(L, 1)) == NULL) {
> +		return luaL_error(L, "Usage: key_def:"
> +				     "compare(tuple_a, tuple_b)");
> +	}
> +
> +	struct tuple *tuple_a, *tuple_b;
> +	if ((tuple_a = lbox_key_def_check_tuple(L, key_def, 2)) == NULL)
> +		return luaT_error(L);
> +	if ((tuple_b = lbox_key_def_check_tuple(L, key_def, 3)) == NULL) {
> +		tuple_unref(tuple_a);
> +		return luaT_error(L);
> +	}
> +
> +	int rc = tuple_compare(tuple_a, tuple_b, key_def);
> +	tuple_unref(tuple_a);
> +	tuple_unref(tuple_b);
> +	lua_pushinteger(L, rc);
> +	return 1;
> +}
> +
> +static int
> +lbox_key_def_compare_with_key(struct lua_State *L)
> +{
> +	struct key_def *key_def;
> +	if (lua_gettop(L) != 3 || (key_def = check_key_def(L, 1)) == NULL) {
> +		return luaL_error(L, "Usage: key_def:"
> +				     "compare_with_key(tuple, key)");
> +	}
> +
> +	struct tuple *tuple, *key_tuple = NULL;
> +	struct tuple_format *format = box_tuple_format_default();
> +	if ((tuple = lbox_key_def_check_tuple(L, key_def, 2)) == NULL)
> +		return luaT_error(L);
> +	if ((key_tuple = luaT_tuple_new(L, 3, format)) == NULL) {
> +		tuple_unref(tuple);
> +		return luaT_error(L);
> +	}
> +	tuple_ref(key_tuple);
> +
> +	const char *key = tuple_data(key_tuple);
> +	assert(mp_typeof(*key) == MP_ARRAY);
> +	uint32_t part_count = mp_decode_array(&key);
> +	if (key_validate_parts(key_def, key, part_count, true) != 0) {
> +		tuple_unref(tuple);
> +		tuple_unref(key_tuple);
> +		return luaT_error(L);
> +	}
> +
> +	int rc = tuple_compare_with_key(tuple, key, part_count, key_def);
> +	tuple_unref(tuple);
> +	tuple_unref(key_tuple);
> +	lua_pushinteger(L, rc);
> +	return 1;
> +}
> +
> +static int
> +lbox_key_def_merge(struct lua_State *L)
> +{
> +	struct key_def *key_def_a, *key_def_b;
> +	if (lua_gettop(L) != 2 || (key_def_a = check_key_def(L, 1)) == NULL ||
> +	   (key_def_b = check_key_def(L, 2)) == NULL)
> +		return luaL_error(L, "Usage: key_def:merge(second_key_def)");
> +
> +	struct key_def *new_key_def = key_def_merge(key_def_a, key_def_b);
> +	if (new_key_def == NULL)
> +		return luaT_error(L);
> +
> +	*(struct key_def **) luaL_pushcdata(L, key_def_type_id) = new_key_def;
> +	lua_pushcfunction(L, lbox_key_def_gc);
> +	luaL_setcdatagc(L, -2);
> +	return 1;
> +}
> +
> +static int
> +lbox_key_def_to_table(struct lua_State *L)
> +{
> +	struct key_def *key_def;
> +	if (lua_gettop(L) != 1 || (key_def = check_key_def(L, 1)) == NULL)
> +		return luaL_error(L, "Usage: key_def:totable()");
> +
> +	lua_createtable(L, key_def->part_count, 0);
> +	for (uint32_t i = 0; i < key_def->part_count; ++i) {
> +		lbox_push_key_part(L, &key_def->parts[i]);
> +		lua_rawseti(L, -2, i + 1);
> +	}
> +	return 1;
> +}
> +
> +/**
> + * Create a new key_def from a Lua table.
> + *
> + * Expected a table of key parts on the Lua stack. The format is
> + * the same as box.space.<...>.index.<...>.parts or corresponding
> + * net.box's one.
> + *
> + * Push the new key_def as cdata to a Lua stack.
> + */
> +static int
> +lbox_key_def_new(struct lua_State *L)
> +{
> +	if (lua_gettop(L) != 1 || lua_istable(L, 1) != 1)
> +		return luaL_error(L, "Bad params, use: key_def.new({"
> +				  "{fieldno = fieldno, type = type"
> +				  "[, is_nullable = <boolean>]"
> +				  "[, path = <string>]"
> +				  "[, collation_id = <number>]"
> +				  "[, collation = <string>]}, ...}");
> +
> +	uint32_t part_count = lua_objlen(L, 1);
> +	const ssize_t parts_size = sizeof(struct key_part_def) * part_count;
> +
> +	struct region *region = &fiber()->gc;
> +	size_t region_svp = region_used(region);
> +	struct key_part_def *parts = region_alloc(region, parts_size);
> +	if (parts == NULL) {
> +		diag_set(OutOfMemory, parts_size, "region", "parts");
> +		return luaT_error(L);
> +	}
> +
> +	for (uint32_t i = 0; i < part_count; ++i) {
> +		lua_pushinteger(L, i + 1);
> +		lua_gettable(L, 1);
> +		if (luaT_key_def_set_part(L, parts, i, region) != 0) {
> +			region_truncate(region, region_svp);
> +			return luaT_error(L);
> +		}
> +	}
> +
> +	struct key_def *key_def = key_def_new(parts, part_count);
> +	region_truncate(region, region_svp);
> +	if (key_def == NULL)
> +		return luaT_error(L);
> +
> +	/*
> +	 * Calculate minimal field count of tuples with specified
> +	 * key and update key_def optionality to use correct
> +	 * compare/extract functions.
> +	 */
> +	uint32_t min_field_count =
> +		tuple_format_min_field_count(&key_def, 1, NULL, 0);
> +	key_def_update_optionality(key_def, min_field_count);
> +
> +	*(struct key_def **) luaL_pushcdata(L, key_def_type_id) = key_def;
> +	lua_pushcfunction(L, lbox_key_def_gc);
> +	luaL_setcdatagc(L, -2);
> +
> +	return 1;
> +}
> +
> +LUA_API int
> +luaopen_key_def(struct lua_State *L)
> +{
> +	luaL_cdef(L, "struct key_def;");
> +	key_def_type_id = luaL_ctypeid(L, "struct key_def&");
> +
> +	/* Export C functions to Lua. */
> +	static const struct luaL_Reg meta[] = {
> +		{"new", lbox_key_def_new},
> +		{NULL, NULL}
> +	};
> +	luaL_register_module(L, "key_def", meta);
> +
> +	lua_newtable(L); /* key_def.internal */
> +	lua_pushcfunction(L, lbox_key_def_extract_key);
> +	lua_setfield(L, -2, "extract_key");
> +	lua_pushcfunction(L, lbox_key_def_compare);
> +	lua_setfield(L, -2, "compare");
> +	lua_pushcfunction(L, lbox_key_def_compare_with_key);
> +	lua_setfield(L, -2, "compare_with_key");
> +	lua_pushcfunction(L, lbox_key_def_merge);
> +	lua_setfield(L, -2, "merge");
> +	lua_pushcfunction(L, lbox_key_def_to_table);
> +	lua_setfield(L, -2, "totable");
> +	lua_setfield(L, -2, "internal");
> +
> +	return 1;
> +}
> diff --git a/src/box/lua/key_def.h b/src/box/lua/key_def.h
> new file mode 100644
> index 000000000..7dc7158e8
> --- /dev/null
> +++ b/src/box/lua/key_def.h
> @@ -0,0 +1,63 @@
> +#ifndef TARANTOOL_BOX_LUA_KEY_DEF_H_INCLUDED
> +#define TARANTOOL_BOX_LUA_KEY_DEF_H_INCLUDED
> +/*
> + * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
> + *
> + * Redistribution and use in source and binary forms, with or
> + * without modification, are permitted provided that the following
> + * conditions are met:
> + *
> + * 1. Redistributions of source code must retain the above
> + *    copyright notice, this list of conditions and the
> + *    following disclaimer.
> + *
> + * 2. Redistributions in binary form must reproduce the above
> + *    copyright notice, this list of conditions and the following
> + *    disclaimer in the documentation and/or other materials
> + *    provided with the distribution.
> + *
> + * THIS SOFTWARE IS PROVIDED BY AUTHORS ``AS IS'' AND
> + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
> + * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
> + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
> + * AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
> + * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
> + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
> + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
> + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
> + * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
> + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
> + * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
> + * SUCH DAMAGE.
> + */
> +
> +#if defined(__cplusplus)
> +extern "C" {
> +#endif /* defined(__cplusplus) */
> +
> +struct lua_State;
> +struct key_part;
> +
> +/**
> + * Push a new table representing a key_part to a Lua stack.
> + */
> +void
> +lbox_push_key_part(struct lua_State *L, const struct key_part *part);
> +
> +/**
> + * Extract a key_def object from a Lua stack.
> + */
> +struct key_def *
> +check_key_def(struct lua_State *L, int idx);
> +
> +/**
> + * Register the module.
> + */
> +int
> +luaopen_key_def(struct lua_State *L);
> +
> +#if defined(__cplusplus)
> +} /* extern "C" */
> +#endif /* defined(__cplusplus) */
> +
> +#endif /* TARANTOOL_BOX_LUA_KEY_DEF_H_INCLUDED */
> diff --git a/src/box/lua/key_def.lua b/src/box/lua/key_def.lua
> new file mode 100644
> index 000000000..122243cc4
> --- /dev/null
> +++ b/src/box/lua/key_def.lua
> @@ -0,0 +1,19 @@
> +local ffi = require('ffi')
> +local key_def = require('key_def')
> +local key_def_t = ffi.typeof('struct key_def')
> +
> +local methods = {
> +    ['extract_key'] = key_def.internal.extract_key,
> +    ['compare'] = key_def.internal.compare,
> +    ['compare_with_key'] = key_def.internal.compare_with_key,
> +    ['merge'] = key_def.internal.merge,
> +    ['totable'] = key_def.internal.totable,
> +    ['__serialize'] = key_def.internal.totable,
> +}
> +
> +ffi.metatype(key_def_t, {
> +    __index = function(self, key)
> +        return methods[key]
> +    end,
> +    __tostring = function(key_def) return "<struct key_def &>" end,
> +})
> diff --git a/src/box/lua/space.cc b/src/box/lua/space.cc
> index ca793e423..6eecbc756 100644
> --- a/src/box/lua/space.cc
> +++ b/src/box/lua/space.cc
> @@ -30,6 +30,7 @@
>   */
>  #include "box/lua/space.h"
>  #include "box/lua/tuple.h"
> +#include "box/lua/key_def.h"
>  #include "box/sql/sqlLimit.h"
>  #include "lua/utils.h"
>  #include "lua/trigger.h"
> @@ -286,32 +287,7 @@ lbox_fillspace(struct lua_State *L, struct space *space, int i)
>  
>  		for (uint32_t j = 0; j < index_def->key_def->part_count; j++) {
>  			lua_pushnumber(L, j + 1);
> -			lua_newtable(L);
> -			const struct key_part *part =
> -				&index_def->key_def->parts[j];
> -
> -			lua_pushstring(L, field_type_strs[part->type]);
> -			lua_setfield(L, -2, "type");
> -
> -			lua_pushnumber(L, part->fieldno + TUPLE_INDEX_BASE);
> -			lua_setfield(L, -2, "fieldno");
> -
> -			if (part->path != NULL) {
> -				lua_pushlstring(L, part->path, part->path_len);
> -				lua_setfield(L, -2, "path");
> -			}
> -
> -			lua_pushboolean(L, key_part_is_nullable(part));
> -			lua_setfield(L, -2, "is_nullable");
> -
> -			if (part->coll_id != COLL_NONE) {
> -				struct coll_id *coll_id =
> -					coll_by_id(part->coll_id);
> -				assert(coll_id != NULL);
> -				lua_pushstring(L, coll_id->name);
> -				lua_setfield(L, -2, "collation");
> -			}
> -
> +			lbox_push_key_part(L, &index_def->key_def->parts[j]);
>  			lua_settable(L, -3); /* index[k].parts[j] */
>  		}
>  
> diff --git a/test/box-tap/key_def.test.lua b/test/box-tap/key_def.test.lua
> new file mode 100755
> index 000000000..756520b8c
> --- /dev/null
> +++ b/test/box-tap/key_def.test.lua
> @@ -0,0 +1,370 @@
> +#!/usr/bin/env tarantool
> +
> +local tap = require('tap')
> +local ffi = require('ffi')
> +local json = require('json')
> +local fun = require('fun')
> +local key_def_lib = require('key_def')
> +
> +local usage_error = 'Bad params, use: key_def.new({' ..
> +                    '{fieldno = fieldno, type = type' ..
> +                    '[, is_nullable = <boolean>]' ..
> +                    '[, path = <string>]' ..
> +                    '[, collation_id = <number>]' ..
> +                    '[, collation = <string>]}, ...}'
> +
> +local function coll_not_found(fieldno, collation)
> +    if type(collation) == 'number' then
> +        return ('Wrong index options (field %d): ' ..
> +               'collation was not found by ID'):format(fieldno)
> +    end
> +
> +    return ('Unknown collation: "%s"'):format(collation)
> +end
> +
> +local function set_key_part_defaults(parts)
> +    local res = {}
> +    for i, part in ipairs(parts) do
> +        res[i] = table.copy(part)
> +        if res[i].is_nullable == nil then
> +            res[i].is_nullable = false
> +        end
> +    end
> +    return res
> +end
> +
> +local key_def_new_cases = {
> +    -- Cases to call before box.cfg{}.
> +    {
> +        'Pass a field on an unknown type',
> +        parts = {{
> +            fieldno = 2,
> +            type = 'unknown',
> +        }},
> +        exp_err = 'Unknown field type: unknown',
> +    },
> +    {
> +        'Try to use collation_id before box.cfg{}',
> +        parts = {{
> +            fieldno = 1,
> +            type = 'string',
> +            collation_id = 2,
> +        }},
> +        exp_err = coll_not_found(1, 2),
> +    },
> +    {
> +        'Try to use collation before box.cfg{}',
> +        parts = {{
> +            fieldno = 1,
> +            type = 'string',
> +            collation = 'unicode_ci',
> +        }},
> +        exp_err = coll_not_found(1, 'unicode_ci'),
> +    },
> +    function()
> +        -- For collations.
> +        box.cfg{}
> +    end,
> +    -- Cases to call after box.cfg{}.
> +    {
> +        'Try to use both collation_id and collation',
> +        parts = {{
> +            fieldno = 1,
> +            type = 'string',
> +            collation_id = 2,
> +            collation = 'unicode_ci',
> +        }},
> +        exp_err = 'Conflicting options: collation_id and collation',
> +    },
> +    {
> +        'Unknown collation_id',
> +        parts = {{
> +            fieldno = 1,
> +            type = 'string',
> +            collation_id = 42,
> +        }},
> +        exp_err = coll_not_found(1, 42),
> +    },
> +    {
> +        'Unknown collation name',
> +        parts = {{
> +            fieldno = 1,
> +            type = 'string',
> +            collation = 'unknown',
> +        }},
> +        exp_err = 'Unknown collation: "unknown"',
> +    },
> +    {
> +        'Bad parts parameter type',
> +        parts = 1,
> +        exp_err = usage_error,
> +    },
> +    {
> +        'No parameters',
> +        params = {},
> +        exp_err = usage_error,
> +    },
> +    {
> +        'Two parameters',
> +        params = {{}, {}},
> +        exp_err = usage_error,
> +    },
> +    {
> +        'Invalid JSON path',
> +        parts = {{
> +            fieldno = 1,
> +            type = 'string',
> +            path = '[3[',
> +        }},
> +        exp_err = 'Wrong index options (field 1): invalid path',
> +    },
> +    {
> +        'Success case; zero parts',
> +        parts = {},
> +        exp_err = nil,
> +    },
> +    {
> +        'Success case; one part',
> +        parts = {{
> +            fieldno = 1,
> +            type = 'string',
> +        }},
> +        exp_err = nil,
> +    },
> +    {
> +        'Success case; one part with a JSON path',
> +        parts = {{
> +            fieldno = 1,
> +            type = 'string',
> +            path = '[3]',
> +        }},
> +        exp_err = nil,
> +    },
> +}
> +
> +local test = tap.test('key_def')
> +
> +test:plan(#key_def_new_cases - 1 + 7)
> +for _, case in ipairs(key_def_new_cases) do
> +    if type(case) == 'function' then
> +        case()
> +    else
> +        local ok, res
> +        if case.params then
> +            ok, res = pcall(key_def_lib.new, unpack(case.params))
> +        else
> +            ok, res = pcall(key_def_lib.new, case.parts)
> +        end
> +        if case.exp_err == nil then
> +            ok = ok and type(res) == 'cdata' and
> +                ffi.istype('struct key_def', res)
> +            test:ok(ok, case[1])
> +        else
> +            local err = tostring(res) -- cdata -> string
> +            test:is_deeply({ok, err}, {false, case.exp_err}, case[1])
> +        end
> +    end
> +end
> +
> +-- Prepare source data for test cases.
> +local parts_a = {
> +    {type = 'unsigned', fieldno = 1},
> +}
> +local parts_b = {
> +    {type = 'number', fieldno = 2},
> +    {type = 'number', fieldno = 3},
> +}
> +local parts_c = {
> +    {type = 'scalar', fieldno = 2},
> +    {type = 'scalar', fieldno = 1},
> +    {type = 'string', fieldno = 4, is_nullable = true},
> +}
> +local key_def_a = key_def_lib.new(parts_a)
> +local key_def_b = key_def_lib.new(parts_b)
> +local key_def_c = key_def_lib.new(parts_c)
> +local tuple_a = box.tuple.new({1, 1, 22})
> +local tuple_b = box.tuple.new({2, 1, 11})
> +local tuple_c = box.tuple.new({3, 1, 22})
> +
> +-- Case: extract_key().
> +test:test('extract_key()', function(test)
> +    test:plan(9)
> +
> +    test:is_deeply(key_def_a:extract_key(tuple_a):totable(), {1}, 'case 1')
> +    test:is_deeply(key_def_b:extract_key(tuple_a):totable(), {1, 22}, 'case 2')
> +
> +    -- JSON path.
> +    local res = key_def_lib.new({
> +        {type = 'string', fieldno = 1, path = 'a.b'},
> +    }):extract_key(box.tuple.new({{a = {b = 'foo'}}})):totable()
> +    test:is_deeply(res, {'foo'}, 'JSON path (tuple argument)')
> +
> +    local res = key_def_lib.new({
> +        {type = 'string', fieldno = 1, path = 'a.b'},
> +    }):extract_key({{a = {b = 'foo'}}}):totable()
> +    test:is_deeply(res, {'foo'}, 'JSON path (table argument)')
> +
> +    -- A key def has a **nullable** part with a field that is over
> +    -- a tuple size.
> +    --
> +    -- The key def options are:
> +    --
> +    -- * is_nullable = true;
> +    -- * has_optional_parts = true.
> +    test:is_deeply(key_def_c:extract_key(tuple_a):totable(), {1, 1, box.NULL},
> +        'short tuple with a nullable part')
> +
> +    -- A key def has a **non-nullable** part with a field that is
> +    -- over a tuple size.
> +    --
> +    -- The key def options are:
> +    --
> +    -- * is_nullable = false;
> +    -- * has_optional_parts = false.
> +    local exp_err = 'Field 2 was not found in the tuple'
> +    local key_def = key_def_lib.new({
> +        {type = 'string', fieldno = 1},
> +        {type = 'string', fieldno = 2},
> +    })
> +    local ok, err = pcall(key_def.extract_key, key_def,
> +        box.tuple.new({'foo'}))
> +    test:is_deeply({ok, tostring(err)}, {false, exp_err},
> +        'short tuple with a non-nullable part (case 1)')
> +
> +    -- Same as before, but a max fieldno is over tuple:len() + 1.
> +    local exp_err = 'Field 2 was not found in the tuple'
> +    local key_def = key_def_lib.new({
> +        {type = 'string', fieldno = 1},
> +        {type = 'string', fieldno = 2},
> +        {type = 'string', fieldno = 3},
> +    })
> +    local ok, err = pcall(key_def.extract_key, key_def,
> +        box.tuple.new({'foo'}))
> +    test:is_deeply({ok, tostring(err)}, {false, exp_err},
> +        'short tuple with a non-nullable part (case 2)')
> +
> +    -- Same as before, but with another key def options:
> +    --
> +    -- * is_nullable = true;
> +    -- * has_optional_parts = false.
> +    local exp_err = 'Field 2 was not found in the tuple'
> +    local key_def = key_def_lib.new({
> +        {type = 'string', fieldno = 1, is_nullable = true},
> +        {type = 'string', fieldno = 2},
> +    })
> +    local ok, err = pcall(key_def.extract_key, key_def,
> +        box.tuple.new({'foo'}))
> +    test:is_deeply({ok, tostring(err)}, {false, exp_err},
> +        'short tuple with a non-nullable part (case 3)')
> +
> +    -- A tuple has a field that does not match corresponding key
> +    -- part type.
> +    local exp_err = 'Supplied key type of part 2 does not match index ' ..
> +                    'part type: expected string'
> +    local key_def = key_def_lib.new({
> +        {type = 'string', fieldno = 1},
> +        {type = 'string', fieldno = 2},
> +        {type = 'string', fieldno = 3},
> +    })
> +    local ok, err = pcall(key_def.extract_key, key_def, {'one', 'two', 3})
> +    test:is_deeply({ok, tostring(err)}, {false, exp_err},
> +        'wrong field type')
> +end)
> +
> +-- Case: compare().
> +test:test('compare()', function(test)
> +    test:plan(8)
> +
> +    test:is(key_def_a:compare(tuple_b, tuple_a), 1,
> +            'case 1: great (tuple argument)')
> +    test:is(key_def_a:compare(tuple_b, tuple_c), -1,
> +            'case 2: less (tuple argument)')
> +    test:is(key_def_b:compare(tuple_b, tuple_a), -1,
> +            'case 3: less (tuple argument)')
> +    test:is(key_def_b:compare(tuple_a, tuple_c), 0,
> +            'case 4: equal (tuple argument)')
> +
> +    test:is(key_def_a:compare(tuple_b:totable(), tuple_a:totable()), 1,
> +            'case 1: great (table argument)')
> +    test:is(key_def_a:compare(tuple_b:totable(), tuple_c:totable()), -1,
> +            'case 2: less (table argument)')
> +    test:is(key_def_b:compare(tuple_b:totable(), tuple_a:totable()), -1,
> +            'case 3: less (table argument)')
> +    test:is(key_def_b:compare(tuple_a:totable(), tuple_c:totable()), 0,
> +            'case 4: equal (table argument)')
> +end)
> +
> +-- Case: compare_with_key().
> +test:test('compare_with_key()', function(test)
> +    test:plan(2)
> +
> +    local key = {1, 22}
> +    test:is(key_def_b:compare_with_key(tuple_a:totable(), key), 0, 'table')
> +
> +    local key = box.tuple.new({1, 22})
> +    test:is(key_def_b:compare_with_key(tuple_a, key), 0, 'tuple')
> +end)
> +
> +-- Case: totable().
> +test:test('totable()', function(test)
> +    test:plan(2)
> +
> +    local exp = set_key_part_defaults(parts_a)
> +    test:is_deeply(key_def_a:totable(), exp, 'case 1')
> +
> +    local exp = set_key_part_defaults(parts_b)
> +    test:is_deeply(key_def_b:totable(), exp, 'case 2')
> +end)
> +
> +-- Case: __serialize().
> +test:test('__serialize()', function(test)
> +    test:plan(2)
> +
> +    local exp = set_key_part_defaults(parts_a)
> +    test:is(json.encode(key_def_a), json.encode(exp), 'case 1')
> +
> +    local exp = set_key_part_defaults(parts_b)
> +    test:is(json.encode(key_def_b), json.encode(exp), 'case 2')
> +end)
> +
> +-- Case: tostring().
> +test:test('tostring()', function(test)
> +    test:plan(2)
> +
> +    local exp = '<struct key_def &>'
> +    test:is(tostring(key_def_a), exp, 'case 1')
> +    test:is(tostring(key_def_b), exp, 'case 2')
> +end)
> +
> +-- Case: merge().
> +test:test('merge()', function(test)
> +    test:plan(6)
> +
> +    local key_def_ab = key_def_a:merge(key_def_b)
> +    local exp_parts = fun.iter(key_def_a:totable())
> +        :chain(fun.iter(key_def_b:totable())):totable()
> +    test:is_deeply(key_def_ab:totable(), exp_parts,
> +        'case 1: verify with :totable()')
> +    test:is_deeply(key_def_ab:extract_key(tuple_a):totable(), {1, 1, 22},
> +        'case 1: verify with :extract_key()')
> +
> +    local key_def_ba = key_def_b:merge(key_def_a)
> +    local exp_parts = fun.iter(key_def_b:totable())
> +        :chain(fun.iter(key_def_a:totable())):totable()
> +    test:is_deeply(key_def_ba:totable(), exp_parts,
> +        'case 2: verify with :totable()')
> +    test:is_deeply(key_def_ba:extract_key(tuple_a):totable(), {1, 22, 1},
> +        'case 2: verify with :extract_key()')
> +
> +    -- Intersecting parts + NULL parts.
> +    local key_def_cb = key_def_c:merge(key_def_b)
> +    local exp_parts = key_def_c:totable()
> +    exp_parts[#exp_parts + 1] = {type = 'number', fieldno = 3,
> +        is_nullable = false}
> +    test:is_deeply(key_def_cb:totable(), exp_parts,
> +        'case 3: verify with :totable()')
> +    test:is_deeply(key_def_cb:extract_key(tuple_a):totable(),
> +        {1, 1, box.NULL, 22}, 'case 3: verify with :extract_key()')
> +end)
> +
> +os.exit(test:check() and 0 or 1)
> -- 
> 2.21.0
>