[PATCH v3 1/1] lua: add key_def lua module

[PATCH v3 1/1] lua: add key_def lua module

[Kirill Shcherbatov]

Needed for #3276.
Fixes #3398.
Fixes #4025.

@TarantoolBot document
Title: lua: key_def module

It is convenient to have tuple compare function into lua-land for the
following case:
 - exporting key from tuple to iterate over secondary non-unique index
   and delete tuples from space
 - support comparing a tuple with a key / a tuple, support merging
   key_defs from Lua
 - factor out key parts parsing code from the tuples merger

A key_def instance has the following methods:
 - :extract_key(tuple)           -> key (as tuple)
    Receives tuple or Lua table. Returns tuple representing extracted
    key.
 - :compare(tuple_a, tuple_b)    -> number
    Receives tuples or Lua tables.
    Returns:
    - a value > 0 when tuple_a > tuple_b,
    - a value == 0 when tuple_a == tuple_b,
    - a value < 0 otherwise.
 - :compare_with_key(tuple, key) -> number
    - a value > 0 when key(tuple) > key,
    - a value == 0 when key(tuple) == key,
    - a value < 0 otherwise.
 - :merge(another_key_def)       -> new key_def instance
    Constructs a new key definition with a set union of key parts
    from first and second key defs
 - :totable()                    -> table
    Dump key_def object as a Lua table (needed to support __serialize
    method)

The root key_def library exports all instance methods directly.

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.

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
```
---
http://github.com/tarantool/tarantool/tree/kshch/gh-4025-lua-key-kef-methods
https://github.com/tarantool/tarantool/issues/4025

 src/CMakeLists.txt            |   1 +
 src/box/CMakeLists.txt        |   2 +
 src/box/errcode.h             |   1 +
 src/box/lua/init.c            |   5 +
 src/box/lua/key_def.c         | 425 ++++++++++++++++++++++++++++++++
 src/box/lua/key_def.h         |  63 +++++
 src/box/lua/key_def.lua       |  19 ++
 src/box/lua/space.cc          |  34 +--
 src/box/tuple.h               |  28 +++
 test/box-tap/key_def.test.lua | 448 ++++++++++++++++++++++++++++++++++
 test/box/misc.result          |   1 +
 11 files changed, 995 insertions(+), 32 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 31600745a..7fbbc7803 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)
 
@@ -140,6 +141,7 @@ add_library(box STATIC
     lua/net_box.c
     lua/xlog.c
     lua/execute.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/errcode.h b/src/box/errcode.h
index 3f8cb8e0e..72485d3c8 100644
--- a/src/box/errcode.h
+++ b/src/box/errcode.h
@@ -246,6 +246,7 @@ struct errcode_record {
 	/*191 */_(ER_SQL_PARSER_LIMIT,		"%s %d exceeds the limit (%d)") \
 	/*192 */_(ER_INDEX_DEF_UNSUPPORTED,	"%s are prohibited in an index definition") \
 	/*193 */_(ER_CK_DEF_UNSUPPORTED,	"%s are prohibited in a CHECK constraint definition") \
+	/*194 */_(ER_TUPLE_KEY_PART_MISSED,	"Supplied tuple field for part %u does not exists") \
 
 /*
  * !IMPORTANT! Please follow instructions at start of the file
diff --git a/src/box/lua/init.c b/src/box/lua/init.c
index 3ff4f3227..6b8be5096 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/execute.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..d5fe237c0
--- /dev/null
+++ b/src/box/lua/key_def.c
@@ -0,0 +1,425 @@
+/*
+ * 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 "misc.h"
+#include "tuple.h"
+
+static uint32_t key_def_type_id = 0;
+
+void
+luaT_push_key_def(struct lua_State *L, const struct key_def *key_def)
+{
+	lua_createtable(L, key_def->part_count, 0);
+	for (uint32_t i = 0; i < key_def->part_count; ++i) {
+		const struct key_part *part = &key_def->parts[i];
+		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");
+		}
+		lua_rawseti(L, -2, i + 1);
+	}
+}
+
+/**
+ * Set key_part_def from a table on top of a Lua stack.
+ * The region argument is used to allocate a JSON path when
+ * required.
+ * 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 *part,
+		      struct region *region)
+{
+	*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(IllegalParams, "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;
+}
+
+static struct tuple *
+luaT_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 || tuple_validate_parts(key_def, tuple) != 0)
+		return NULL;
+	tuple_ref(tuple);
+	return tuple;
+}
+
+struct key_def *
+luaT_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 = luaT_check_key_def(L, 1);
+	if (key_def == NULL)
+		return 0;
+	box_key_def_delete(key_def);
+	return 0;
+}
+
+static int
+lbox_key_def_extract_key(struct lua_State *L)
+{
+	struct key_def *key_def;
+	if (lua_gettop(L) != 2 || (key_def = luaT_check_key_def(L, 1)) == NULL)
+		return luaL_error(L, "Usage: key_def:extract_key(tuple)");
+
+	struct tuple *tuple;
+	if ((tuple = luaT_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 = luaT_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 = luaT_key_def_check_tuple(L, key_def, 2)) == NULL)
+		return luaT_error(L);
+	if ((tuple_b = luaT_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 = luaT_check_key_def(L, 1)) == NULL) {
+		return luaL_error(L, "Usage: key_def:"
+				     "compare_with_key(tuple, key)");
+	}
+
+	struct tuple *tuple = luaT_key_def_check_tuple(L, key_def, 2);
+	if (tuple == NULL)
+		return luaT_error(L);
+
+	size_t key_len;
+	const char *key = lbox_encode_tuple_on_gc(L, 3, &key_len);
+	uint32_t part_count = mp_decode_array(&key);
+	if (key_validate_parts(key_def, key, part_count, true) != 0) {
+		tuple_unref(tuple);
+		return luaT_error(L);
+	}
+
+	int rc = tuple_compare_with_key(tuple, key, part_count, key_def);
+	tuple_unref(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 = luaT_check_key_def(L, 1)) == NULL ||
+	    (key_def_b = luaT_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 = luaT_check_key_def(L, 1)) == NULL)
+		return luaL_error(L, "Usage: key_def:totable()");
+
+	luaT_push_key_def(L, key_def);
+	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);
+
+	/*
+	 * Compare and extract key_def methods must work even with
+	 * tuples with omitted (optional) fields. As there is no
+	 * space format which would guarantee certain minimal
+	 * field_count, pass min_field_count = 0 to ensure that
+	 * functions will work correctly in such case.
+	 */
+	key_def_update_optionality(key_def, 0);
+
+	*(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},
+		{"extract_key", lbox_key_def_extract_key},
+		{"compare", lbox_key_def_compare},
+		{"compare_with_key", lbox_key_def_compare_with_key},
+		{"merge", lbox_key_def_merge},
+		{"totable", lbox_key_def_to_table},
+		{NULL, NULL}
+	};
+	luaL_register_module(L, "key_def", meta);
+	return 1;
+}
diff --git a/src/box/lua/key_def.h b/src/box/lua/key_def.h
new file mode 100644
index 000000000..2cecfaae5
--- /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_def;
+
+/**
+ * Push a new table representing a key_def to a Lua stack.
+ */
+void
+luaT_push_key_def(struct lua_State *L, const struct key_def *key_def);
+
+/**
+ * Extract a key_def object from a Lua stack.
+ */
+struct key_def *
+luaT_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..586005709
--- /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.extract_key,
+    ['compare'] = key_def.compare,
+    ['compare_with_key'] = key_def.compare_with_key,
+    ['merge'] = key_def.merge,
+    ['totable'] = key_def.totable,
+    ['__serialize'] = key_def.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 93269b72b..100da0a79 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,38 +287,7 @@ lbox_fillspace(struct lua_State *L, struct space *space, int i)
 		lua_setfield(L, -2, "name");
 
 		lua_pushstring(L, "parts");
-		lua_newtable(L);
-
-		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");
-			}
-
-			lua_settable(L, -3); /* index[k].parts[j] */
-		}
+		luaT_push_key_def(L, index_def->key_def);
 
 		lua_settable(L, -3); /* space.index[k].parts */
 
diff --git a/src/box/tuple.h b/src/box/tuple.h
index eed8e1a34..a3ac45574 100644
--- a/src/box/tuple.h
+++ b/src/box/tuple.h
@@ -670,6 +670,34 @@ tuple_field_by_part(struct tuple *tuple, struct key_part *part)
 				       tuple_field_map(tuple), part);
 }
 
+/**
+ * Check that tuple fields match with given key definition
+ * key_def.
+ * @param key_def Key definition.
+ * @param tuple Tuple to validate.
+ *
+ * @retval 0  The tuple is valid.
+ * @retval -1 The tuple is invalid.
+ */
+static inline int
+tuple_validate_parts(struct key_def *key_def, struct tuple *tuple)
+{
+	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) {
+			if (key_part_is_nullable(part))
+				continue;
+			diag_set(ClientError, ER_TUPLE_KEY_PART_MISSED, idx);
+			return -1;
+		}
+		if (key_part_validate(part->type, field, idx,
+				      key_part_is_nullable(part)) != 0)
+			return -1;
+	}
+	return 0;
+}
+
 /**
  * @brief Tuple Interator
  */
diff --git a/test/box-tap/key_def.test.lua b/test/box-tap/key_def.test.lua
new file mode 100755
index 000000000..5fc9cc67b
--- /dev/null
+++ b/test/box-tap/key_def.test.lua
@@ -0,0 +1,448 @@
+#!/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 = 999,
+        }},
+        exp_err = coll_not_found(1, 999),
+    },
+    {
+        '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 = '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.
+
+-- Case: extract_key().
+test:test('extract_key()', function(test)
+    test:plan(13)
+
+    local key_def_a = key_def_lib.new({
+        {type = 'unsigned', fieldno = 1},
+    })
+    local key_def_b = key_def_lib.new({
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    })
+    local key_def_c = key_def_lib.new({
+        {type = 'scalar', fieldno = 2},
+        {type = 'scalar', fieldno = 1},
+        {type = 'string', fieldno = 4, is_nullable = true},
+    })
+    local tuple_a = box.tuple.new({1, 1, 22})
+
+    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 = 'Supplied tuple field for part 1 does not exists'
+    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 = 'Supplied tuple field for part 1 does not exists'
+    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 = 'Supplied tuple field for part 1 does not exists'
+    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')
+
+    local key_def = key_def_lib.new({
+        {type = 'number', fieldno = 1, path='a'},
+        {type = 'number', fieldno = 1, path='b'},
+        {type = 'number', fieldno = 1, path='c', is_nullable=true},
+        {type = 'number', fieldno = 3, is_nullable=true},
+    })
+    local ok, err = pcall(key_def.extract_key, key_def,
+                          box.tuple.new({1, 1, 22}))
+    test:is_deeply({ok, tostring(err)},
+                   {false, 'Supplied tuple field for part 0 does not exists'},
+                   'invalid JSON structure')
+    test:is_deeply(key_def:extract_key({{a=1, b=2}, 1}):totable(),
+                   {1, 2, box.NULL, box.NULL},
+                   'tuple with optional parts - case 1')
+    test:is_deeply(key_def:extract_key({{a=1, b=2, c=3}, 1}):totable(),
+                   {1, 2, 3, box.NULL},
+                   'tuple with optional parts - case 2')
+    test:is_deeply(key_def:extract_key({{a=1, b=2}, 1, 3}):totable(),
+                   {1, 2, box.NULL, 3},
+                   'tuple with optional parts - case 3')
+end)
+
+-- Case: compare().
+test:test('compare()', function(test)
+    test:plan(8)
+
+    local key_def_a = key_def_lib.new({
+        {type = 'unsigned', fieldno = 1},
+    })
+    local key_def_b = key_def_lib.new({
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    })
+    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})
+
+    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_def_b = key_def_lib.new({
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    })
+    local tuple_a = box.tuple.new({1, 1, 22})
+
+    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 parts_a = {
+        {type = 'unsigned', fieldno = 1}
+    }
+    local parts_b = {
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    }
+    local key_def_a = key_def_lib.new(parts_a)
+    local key_def_b = key_def_lib.new(parts_b)
+
+    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 parts_a = {
+        {type = 'unsigned', fieldno = 1}
+    }
+    local parts_b = {
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    }
+    local key_def_a = key_def_lib.new(parts_a)
+    local key_def_b = key_def_lib.new(parts_b)
+
+    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 parts_a = {
+        {type = 'unsigned', fieldno = 1}
+    }
+    local parts_b = {
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    }
+    local key_def_a = key_def_lib.new(parts_a)
+    local key_def_b = key_def_lib.new(parts_b)
+
+    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_a = key_def_lib.new({
+        {type = 'unsigned', fieldno = 1},
+    })
+    local key_def_b = key_def_lib.new({
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    })
+    local key_def_c = key_def_lib.new({
+        {type = 'scalar', fieldno = 2},
+        {type = 'scalar', fieldno = 1},
+        {type = 'string', fieldno = 4, is_nullable = true},
+    })
+    local tuple_a = box.tuple.new({1, 1, 22})
+
+    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)
diff --git a/test/box/misc.result b/test/box/misc.result
index a1f7a0990..7d6d54241 100644
--- a/test/box/misc.result
+++ b/test/box/misc.result
@@ -522,6 +522,7 @@ t;
   191: box.error.SQL_PARSER_LIMIT
   192: box.error.INDEX_DEF_UNSUPPORTED
   193: box.error.CK_DEF_UNSUPPORTED
+  194: box.error.TUPLE_KEY_PART_MISSED
 ...
 test_run:cmd("setopt delimiter ''");
 ---
-- 
2.21.0

Re: [tarantool-patches] [PATCH v3 1/1] lua: add key_def lua module

[Konstantin Osipov]

* Kirill Shcherbatov <kshcherbatov@tarantool.org> [19/04/22 17:40]:
> --- a/src/box/errcode.h
> +++ b/src/box/errcode.h
> @@ -246,6 +246,7 @@ struct errcode_record {
>  	/*191 */_(ER_SQL_PARSER_LIMIT,		"%s %d exceeds the limit (%d)") \
>  	/*192 */_(ER_INDEX_DEF_UNSUPPORTED,	"%s are prohibited in an index definition") \
>  	/*193 */_(ER_CK_DEF_UNSUPPORTED,	"%s are prohibited in a CHECK constraint definition") \
> +	/*194 */_(ER_TUPLE_KEY_PART_MISSED,	"Supplied tuple field for part %u does not exists") \

I don't understand this error message. 
Besides key part number, the message should contain field name or
number.

Besides, we already have ER_NO_SUCH_FIELD_NO and
why not use it?


ER_NO_FIELD_FOR_KEY_PART "The supplied tuple has no field %s for key part %u"

> +
> +static uint32_t key_def_type_id = 0;

Please rename to key_def_ctype_id

> +static int
> +lbox_key_def_compare(struct lua_State *L)
> +{
> +	struct key_def *key_def;
> +	if (lua_gettop(L) != 3 ||
> +	    (key_def = luaT_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 = luaT_key_def_check_tuple(L, key_def, 2)) == NULL)
> +		return luaT_error(L);
> +	if ((tuple_b = luaT_key_def_check_tuple(L, key_def, 3)) == NULL) {
> +		tuple_unref(tuple_a);
> +		return luaT_error(L);
> +	}

Invoking tuple_validate and possibly tuple_new on each compare is
awfully slow.
> +static int
> +lbox_key_def_compare_with_key(struct lua_State *L)
> +{
> +	struct key_def *key_def;
> +	if (lua_gettop(L) != 3 ||
> +	    (key_def = luaT_check_key_def(L, 1)) == NULL) {
> +		return luaL_error(L, "Usage: key_def:"
> +				     "compare_with_key(tuple, key)");
> +	}
> +
> +	struct tuple *tuple = luaT_key_def_check_tuple(L, key_def, 2);
> +	if (tuple == NULL)
> +		return luaT_error(L);
> +
> +	size_t key_len;
> +	const char *key = lbox_encode_tuple_on_gc(L, 3, &key_len);
> +	uint32_t part_count = mp_decode_array(&key);
> +	if (key_validate_parts(key_def, key, part_count, true) != 0) {
> +		tuple_unref(tuple);
> +		return luaT_error(L);
> +	}
> +
> +	int rc = tuple_compare_with_key(tuple, key, part_count, key_def);
> +	tuple_unref(tuple);
> +	lua_pushinteger(L, rc);
> +	return 1;
> +}

This also looks as a terribly inefficient implementation for
compare.

Overall, the API looks good to me, while the implementation seems
to be too inefficient. I would consider changing extract_key() to
return char *, not struct tuple, the buffer could be allocated on
transaction region. I also think that compare should not allocate
memory or create tuples, and it should not call tuple_validate()
either.

If this is urgent, I would push since the code quality is very
good and the api would stable, but I don't see how soo inefficient
compare functions could be useful.

-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov

Re: [tarantool-patches] Re: [PATCH v3 1/1] lua: add key_def lua module

[Alexander Turenko]

> > +static int
> > +lbox_key_def_compare_with_key(struct lua_State *L)
> > +{
> > +	struct key_def *key_def;
> > +	if (lua_gettop(L) != 3 ||
> > +	    (key_def = luaT_check_key_def(L, 1)) == NULL) {
> > +		return luaL_error(L, "Usage: key_def:"
> > +				     "compare_with_key(tuple, key)");
> > +	}
> > +
> > +	struct tuple *tuple = luaT_key_def_check_tuple(L, key_def, 2);
> > +	if (tuple == NULL)
> > +		return luaT_error(L);
> > +
> > +	size_t key_len;
> > +	const char *key = lbox_encode_tuple_on_gc(L, 3, &key_len);
> > +	uint32_t part_count = mp_decode_array(&key);
> > +	if (key_validate_parts(key_def, key, part_count, true) != 0) {
> > +		tuple_unref(tuple);
> > +		return luaT_error(L);
> > +	}
> > +
> > +	int rc = tuple_compare_with_key(tuple, key, part_count, key_def);
> > +	tuple_unref(tuple);
> > +	lua_pushinteger(L, rc);
> > +	return 1;
> > +}
> 
> This also looks as a terribly inefficient implementation for
> compare.
> 
> Overall, the API looks good to me, while the implementation seems
> to be too inefficient. I would consider changing extract_key() to
> return char *, not struct tuple, the buffer could be allocated on
> transaction region. I also think that compare should not allocate
> memory or create tuples, and it should not call tuple_validate()
> either.
> 
> If this is urgent, I would push since the code quality is very
> good and the api would stable, but I don't see how soo inefficient
> compare functions could be useful.

We can add *_unchecked() functions, which will not perform validation.

I'm against returning char *, because a user can do nothing with it in
Lua.

We can consider a cursor implementation as motivating example. For a
cursor that navigates over an unique index we need :extract_key(), see
[1]. Comparing of tuples is needed if we'll try to implement cursors on
a non-unique index (but maybe better to implement #3898 ('Allow
specifying primary key in iteration by secondary key').

I don't know what Michael F. want when he file #3398 ('lua: introduce
key_def module'). Maybe he just needs merger. Maybe we should introduce
built-in tuple sorter instead of exposing comparators into Lua, which
can be either fast or safe, but not both.

[1]: https://github.com/Totktonada/tarantool-merger-examples/blob/695fc9511685033f4b4b22c0df537a12ddf2eaf6/chunked_example_fast/storage.lua#L97

WBR, Alexander Turenko.

Re: [PATCH v3 1/1] lua: add key_def lua module

[Vladimir Davydov]

On Mon, Apr 22, 2019 at 03:28:55PM +0300, Kirill Shcherbatov wrote:
> diff --git a/src/box/errcode.h b/src/box/errcode.h
> index 3f8cb8e0e..72485d3c8 100644
> --- a/src/box/errcode.h
> +++ b/src/box/errcode.h
> @@ -246,6 +246,7 @@ struct errcode_record {
>  	/*191 */_(ER_SQL_PARSER_LIMIT,		"%s %d exceeds the limit (%d)") \
>  	/*192 */_(ER_INDEX_DEF_UNSUPPORTED,	"%s are prohibited in an index definition") \
>  	/*193 */_(ER_CK_DEF_UNSUPPORTED,	"%s are prohibited in a CHECK constraint definition") \
> +	/*194 */_(ER_TUPLE_KEY_PART_MISSED,	"Supplied tuple field for part %u does not exists") \

We have ER_FIELD_MISSING, ER_NO_SUCH_FIELD_NO, ER_NO_SUCN_FIELD_NAME.
Please reuse rather than introducing yet another error code. Also the
error message should include a path to the missing field.

>  
>  /*
>   * !IMPORTANT! Please follow instructions at start of the file
> diff --git a/src/box/lua/init.c b/src/box/lua/init.c
> index 3ff4f3227..6b8be5096 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/execute.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..d5fe237c0
> --- /dev/null
> +++ b/src/box/lua/key_def.c
> @@ -0,0 +1,425 @@
> +/*
> + * 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 "misc.h"
> +#include "tuple.h"
> +
> +static uint32_t key_def_type_id = 0;

Here are the names of other variables storing Lua type ids:

  CTID_STRUCT_IBUF
  CTID_STRUCT_PORT_PTR
  CTID_STRUCT_ITERATOR_REF
  CTID_STRUCT_ERROR_REF

Please rename this one to conform.

> +
> +void
> +luaT_push_key_def(struct lua_State *L, const struct key_def *key_def)
> +{
> +	lua_createtable(L, key_def->part_count, 0);
> +	for (uint32_t i = 0; i < key_def->part_count; ++i) {
> +		const struct key_part *part = &key_def->parts[i];
> +		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");
> +		}
> +		lua_rawseti(L, -2, i + 1);
> +	}
> +}
> +
> +/**
> + * Set key_part_def from a table on top of a Lua stack.
> + * The region argument is used to allocate a JSON path when
> + * required.
> + * 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 *part,
> +		      struct region *region)
> +{

I typed this by mistake:

	tarantool> key_def.new({fieldno = 1, type = 'unsigned'})

and got a valid key def:

	---
	- []
	...

Can we possibly raise an error in this case?

> +	*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(IllegalParams, "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;
> +}
> +
> +static struct tuple *
> +luaT_key_def_check_tuple(struct lua_State *L, struct key_def *key_def, int idx)

Please write a comment explaining briefly what this function does.
Relevant to all the static functions below.

> +{
> +	struct tuple *tuple = luaT_istuple(L, idx);
> +	if (tuple == NULL)
> +		tuple = luaT_tuple_new(L, idx, box_tuple_format_default());
> +	if (tuple == NULL || tuple_validate_parts(key_def, tuple) != 0)
> +		return NULL;
> +	tuple_ref(tuple);
> +	return tuple;
> +}
> +
> +struct key_def *
> +luaT_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 = luaT_check_key_def(L, 1);
> +	if (key_def == NULL)
> +		return 0;

Hmm, how's that even possible?

> +	box_key_def_delete(key_def);
> +	return 0;
> +}
> +
> +static int
> +lbox_key_def_extract_key(struct lua_State *L)
> +{
> +	struct key_def *key_def;
> +	if (lua_gettop(L) != 2 || (key_def = luaT_check_key_def(L, 1)) == NULL)
> +		return luaL_error(L, "Usage: key_def:extract_key(tuple)");
> +
> +	struct tuple *tuple;
> +	if ((tuple = luaT_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_extract_key allocates a key on the region.
You ought to clean it up once you're done with it.

> +	tuple_unref(tuple);
> +	if (key == NULL)
> +		return luaT_error(L);
> +
> +	struct tuple *ret =
> +		box_tuple_new(box_tuple_format_default(), key, key + key_size);

Please don't use box_ functions here - those are intended to be used
only by external modules. Use tuple_new, tuple_format_runtime directly.
It's especially confusing when you create a key def with key_def_new,
but free it with box_key_def_delete.

Anyway, do we really need to create a tuple here? May be returning a Lua
table would be fine?

> +	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 = luaT_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 = luaT_key_def_check_tuple(L, key_def, 2)) == NULL)
> +		return luaT_error(L);
> +	if ((tuple_b = luaT_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 = luaT_check_key_def(L, 1)) == NULL) {
> +		return luaL_error(L, "Usage: key_def:"
> +				     "compare_with_key(tuple, key)");
> +	}
> +
> +	struct tuple *tuple = luaT_key_def_check_tuple(L, key_def, 2);
> +	if (tuple == NULL)
> +		return luaT_error(L);
> +
> +	size_t key_len;
> +	const char *key = lbox_encode_tuple_on_gc(L, 3, &key_len);

Again, this function allocates a key on the region. Please clean it up.

> +	uint32_t part_count = mp_decode_array(&key);
> +	if (key_validate_parts(key_def, key, part_count, true) != 0) {
> +		tuple_unref(tuple);
> +		return luaT_error(L);
> +	}
> +
> +	int rc = tuple_compare_with_key(tuple, key, part_count, key_def);
> +	tuple_unref(tuple);
> +	lua_pushinteger(L, rc);
> +	return 1;
> +}
> diff --git a/src/box/lua/key_def.h b/src/box/lua/key_def.h
> new file mode 100644
> index 000000000..2cecfaae5
> --- /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_def;
> +
> +/**
> + * Push a new table representing a key_def to a Lua stack.
> + */
> +void
> +luaT_push_key_def(struct lua_State *L, const struct key_def *key_def);
> +
> +/**
> + * Extract a key_def object from a Lua stack.

The comment is insufficient: what happens if the object isn't a key def?

> + */
> +struct key_def *
> +luaT_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/tuple.h b/src/box/tuple.h
> index eed8e1a34..a3ac45574 100644
> --- a/src/box/tuple.h
> +++ b/src/box/tuple.h
> @@ -670,6 +670,34 @@ tuple_field_by_part(struct tuple *tuple, struct key_part *part)
>  				       tuple_field_map(tuple), part);
>  }
>  
> +/**
> + * Check that tuple fields match with given key definition
> + * key_def.
> + * @param key_def Key definition.
> + * @param tuple Tuple to validate.
> + *
> + * @retval 0  The tuple is valid.
> + * @retval -1 The tuple is invalid.
> + */
> +static inline int
> +tuple_validate_parts(struct key_def *key_def, struct tuple *tuple)
> +{

IMO tuple.h isn't a right place for this function:

 - First, it's relatively big and cold so I'd move it to a C file.
 - Second, it's about checking if a tuple conforms to a key_def
   while tuple.[hc] stores only basic struct tuple helpers and
   shouldn't depend on key_def (although sadly it does for now).

The name is somewhat confusing too: we now have tuple_validate and
tuple_validate_parts both of which validate tuple parts...

What about

 - renaming it to tuple_validate_key_parts so as to emphasize that it's
   about key parts, not tuple parts;
 - declaring it in key_def.h while defining it in tuple_extract_key.cc,
   similarly to tuple_key_contains_null, as this function can be used
   for checking if we can extract a specific key from a given tuple

?

> +	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) {
> +			if (key_part_is_nullable(part))
> +				continue;
> +			diag_set(ClientError, ER_TUPLE_KEY_PART_MISSED, idx);
> +			return -1;
> +		}
> +		if (key_part_validate(part->type, field, idx,
> +				      key_part_is_nullable(part)) != 0)
> +			return -1;
> +	}
> +	return 0;
> +}

Re: [tarantool-patches] Re: [PATCH v3 1/1] lua: add key_def lua module

[Kirill Shcherbatov]

Hi! Thank you for review.

> We have ER_FIELD_MISSING, ER_NO_SUCH_FIELD_NO, ER_NO_SUCN_FIELD_NAME.
> Please reuse rather than introducing yet another error code. Also the
> error message should include a path to the missing field.

Replaced with

diag_set(ClientError, ER_FIELD_MISSING,
		tt_sprintf("[%d]%.*s", part->fieldno + TUPLE_INDEX_BASE,
				 part->path_len, part->path));

>> +static uint32_t key_def_type_id = 0;
> 
> Here are the names of other variables storing Lua type ids:
> 
>   CTID_STRUCT_IBUF
>   CTID_STRUCT_PORT_PTR
>   CTID_STRUCT_ITERATOR_REF
>   CTID_STRUCT_ERROR_REF
> 
> Please rename this one to conform.

Done.

> I typed this by mistake:
> 
> 	tarantool> key_def.new({fieldno = 1, type = 'unsigned'})
> 
> and got a valid key def:
> 
> 	---
> 	- []
> 	...
> 
> Can we possibly raise an error in this case?

I've disallowed creating a key_def with part_count = 0 so this is not problem
anymore.

>> +static struct tuple *
>> +luaT_key_def_check_tuple(struct lua_State *L, struct key_def *key_def, int idx)
> 
> Please write a comment explaining briefly what this function does.
> Relevant to all the static functions below.

/**
* Set key_part_def from a table on top of a Lua stack.
* The region argument is used to allocate a JSON path when
* required.
* 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 *part,
struct region *region)

/**
* Check an existent tuple pointer in LUA stack by specified
* index or attemt to construct it by LUA table.
* Increase tuple's reference counter.
* Returns not NULL tuple pointer on success, NULL otherwise.
*/
static struct tuple *
luaT_key_def_check_tuple(struct lua_State *L, struct key_def *key_def, int idx)

/**
* Extract key from tuple by given key definition and return
* tuple representing this key.
* Push the new key tuple as cdata to a LUA stack on success.
* Raise error otherwise.
*/
static int
lbox_key_def_extract_key(struct lua_State *L)

/**
* Compare tuples using the key definition.
* Push 0 if key_fields(tuple_a) == key_fields(tuple_b)
* <0 if key_fields(tuple_a) < key_fields(tuple_b)
* >0 if key_fields(tuple_a) > key_fields(tuple_b)
* integer to a LUA stack on success.
* Raise error otherwise.
*/
static int
lbox_key_def_compare(struct lua_State *L)

/**
* Compare tuple with key using the key definition.
* Push 0 if key_fields(tuple) == parts(key)
* <0 if key_fields(tuple) < parts(key)
* >0 if key_fields(tuple) > parts(key)
* integer to a LUA stack on success.
* Raise error otherwise.
*/
static int
lbox_key_def_compare_with_key(struct lua_State *L)

/**
* Construct and export to LUA a new key definition with a set
* union of key parts from first and second key defs. Parts of
* the new key_def consist of the first key_def's parts and those
* parts of the second key_def that were not among the first
* parts.
* Push the new key_def as cdata to a LUA stack on success.
* Raise error otherwise.
*/
static int
lbox_key_def_merge(struct lua_State *L)

/**
* Push a new table representing a key_def to a Lua stack.
*/
static int
lbox_key_def_to_table(struct lua_State *L)

/**
* 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)


>> +static int
>> +lbox_key_def_gc(struct lua_State *L)
>> +{
>> +	struct key_def *key_def = luaT_check_key_def(L, 1);
>> +	if (key_def == NULL)
>> +		return 0;
> 
> Hmm, how's that even possible?

Nope, replaced with assert.

>> +	struct tuple *ret =
>> +		box_tuple_new(box_tuple_format_default(), key, key + key_size);
> 
> Please don't use box_ functions here - those are intended to be used
> only by external modules. Use tuple_new, tuple_format_runtime directly.
> It's especially confusing when you create a key def with key_def_new,
> but free it with box_key_def_delete.

Done.

> Anyway, do we really need to create a tuple here? May be returning a Lua
> table would be fine?

Discussed to do not implement it because of it's complexity and lack of
expediency.

>> +	uint32_t key_size;
>> +	char *key = tuple_extract_key(tuple, key_def, &key_size);
> 
> tuple_extract_key allocates a key on the region.
> You ought to clean it up once you're done with it.

>> +	size_t key_len;
>> +	const char *key = lbox_encode_tuple_on_gc(L, 3, &key_len);
> 
> Again, this function allocates a key on the region. Please clean it up.

Done.

>> +/**
>> + * Extract a key_def object from a Lua stack.
> 
> The comment is insufficient: what happens if the object isn't a key def?
> 

/**
 * Push a new table representing a key_def to a Lua stack.
 * Table is consists of key_def::parts tables that describe
 * each part correspondingly.
 * The collation and path fields are optional so resulting
 * object doesn't declare them where not necessary.
 */
void
luaT_push_key_def(struct lua_State *L, const struct key_def *key_def);

/**
 * Check key_def pointer in LUA stack by specified index.
 * The value by idx is expected to be key_def's cdata.
 * Returns not NULL tuple pointer on success, NULL otherwise.
 */
struct key_def *
luaT_check_key_def(struct lua_State *L, int idx);

> IMO tuple.h isn't a right place for this function:
> 
>  - First, it's relatively big and cold so I'd move it to a C file.
>  - Second, it's about checking if a tuple conforms to a key_def
>    while tuple.[hc] stores only basic struct tuple helpers and
>    shouldn't depend on key_def (although sadly it does for now).
> 
> The name is somewhat confusing too: we now have tuple_validate and
> tuple_validate_parts both of which validate tuple parts...
> 
> What about
> 
>  - renaming it to tuple_validate_key_parts so as to emphasize that it's
>    about key parts, not tuple parts;
>  - declaring it in key_def.h while defining it in tuple_extract_key.cc,
>    similarly to tuple_key_contains_null, as this function can be used
>    for checking if we can extract a specific key from a given tuple
> 
> ?

Song good. Done.

==========================================================

Needed for #3276.
Fixes #3398.
Fixes #4025.

@TarantoolBot document
Title: lua: key_def module

It is convenient to have tuple compare function into lua-land for the
following case:
 - exporting key from tuple to iterate over secondary non-unique index
   and delete tuples from space
 - support comparing a tuple with a key / a tuple, support merging
   key_defs from Lua
 - factor out key parts parsing code from the tuples merger

A key_def instance has the following methods:
 - :extract_key(tuple)           -> key (as tuple)
    Receives tuple or Lua table. Returns tuple representing extracted
    key.
 - :compare(tuple_a, tuple_b)    -> number
    Receives tuples or Lua tables.
    Returns:
    - a value > 0 when tuple_a > tuple_b,
    - a value == 0 when tuple_a == tuple_b,
    - a value < 0 otherwise.
 - :compare_with_key(tuple, key) -> number
    - a value > 0 when key(tuple) > key,
    - a value == 0 when key(tuple) == key,
    - a value < 0 otherwise.
 - :merge(another_key_def)       -> new key_def instance
    Constructs a new key definition with a set union of key parts
    from first and second key defs
 - :totable()                    -> table
    Dump key_def object as a Lua table (needed to support __serialize
    method)

The root key_def library exports all instance methods directly.

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.

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/key_def.h             |  12 +
 src/box/lua/init.c            |   5 +
 src/box/lua/key_def.c         | 480 ++++++++++++++++++++++++++++++++++
 src/box/lua/key_def.h         |  69 +++++
 src/box/lua/key_def.lua       |  19 ++
 src/box/lua/space.cc          |  34 +--
 src/box/tuple_extract_key.cc  |  22 ++
 test/box-tap/key_def.test.lua | 453 ++++++++++++++++++++++++++++++++
 10 files changed, 1065 insertions(+), 32 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 31600745a..7fbbc7803 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)
 
@@ -140,6 +141,7 @@ add_library(box STATIC
     lua/net_box.c
     lua/xlog.c
     lua/execute.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/key_def.h b/src/box/key_def.h
index 6205c278a..ff0dcbecf 100644
--- a/src/box/key_def.h
+++ b/src/box/key_def.h
@@ -515,6 +515,18 @@ key_part_cmp(const struct key_part *parts1, uint32_t part_count1,
 bool
 tuple_key_contains_null(struct tuple *tuple, struct key_def *def);
 
+/**
+ * Check that tuple fields match with given key definition
+ * key_def.
+ * @param key_def Key definition.
+ * @param tuple Tuple to validate.
+ *
+ * @retval 0  The tuple is valid.
+ * @retval -1 The tuple is invalid.
+ */
+int
+tuple_validate_key_parts(struct key_def *key_def, struct tuple *tuple);
+
 /**
  * Extract key from tuple by given key definition and return
  * buffer allocated on box_txn_alloc with this key. This function
diff --git a/src/box/lua/init.c b/src/box/lua/init.c
index 3ff4f3227..6b8be5096 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/execute.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..0e1236093
--- /dev/null
+++ b/src/box/lua/key_def.c
@@ -0,0 +1,480 @@
+/*
+ * 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 "misc.h"
+#include "tuple.h"
+
+static uint32_t CTID_STRUCT_KEY_DEF_REF = 0;
+
+void
+luaT_push_key_def(struct lua_State *L, const struct key_def *key_def)
+{
+	lua_createtable(L, key_def->part_count, 0);
+	for (uint32_t i = 0; i < key_def->part_count; ++i) {
+		const struct key_part *part = &key_def->parts[i];
+		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");
+		}
+		lua_rawseti(L, -2, i + 1);
+	}
+}
+
+/**
+ * Set key_part_def from a table on top of a Lua stack.
+ * The region argument is used to allocate a JSON path when
+ * required.
+ * 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 *part,
+		      struct region *region)
+{
+	*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(IllegalParams, "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;
+}
+
+/**
+ * Check an existent tuple pointer  in LUA stack by specified
+ * index or attemt to construct it by LUA table.
+ * Increase tuple's reference counter.
+ * Returns not NULL tuple pointer on success, NULL otherwise.
+ */
+static struct tuple *
+luaT_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 || tuple_validate_key_parts(key_def, tuple) != 0)
+		return NULL;
+	tuple_ref(tuple);
+	return tuple;
+}
+
+struct key_def *
+luaT_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 != CTID_STRUCT_KEY_DEF_REF)
+		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 = luaT_check_key_def(L, 1);
+	assert(key_def != NULL);
+	key_def_delete(key_def);
+	return 0;
+}
+
+/**
+ * Extract key from tuple by given key definition and return
+ * tuple representing this key.
+ * Push the new key tuple as cdata to a LUA stack on success.
+ * Raise error otherwise.
+ */
+static int
+lbox_key_def_extract_key(struct lua_State *L)
+{
+	struct key_def *key_def;
+	if (lua_gettop(L) != 2 || (key_def = luaT_check_key_def(L, 1)) == NULL)
+		return luaL_error(L, "Usage: key_def:extract_key(tuple)");
+
+	struct tuple *tuple;
+	if ((tuple = luaT_key_def_check_tuple(L, key_def, 2)) == NULL)
+		return luaT_error(L);
+
+	struct region *region = &fiber()->gc;
+	size_t region_svp = region_used(region);
+	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 =
+		tuple_new(tuple_format_runtime, key, key + key_size);
+	region_truncate(region, region_svp);
+	if (ret == NULL)
+		return luaT_error(L);
+	luaT_pushtuple(L, ret);
+	return 1;
+}
+
+/**
+ * Compare tuples using the key definition.
+ * Push 0  if key_fields(tuple_a) == key_fields(tuple_b)
+ *      <0 if key_fields(tuple_a) < key_fields(tuple_b)
+ *      >0 if key_fields(tuple_a) > key_fields(tuple_b)
+ * integer to a LUA stack on success.
+ * Raise error otherwise.
+ */
+static int
+lbox_key_def_compare(struct lua_State *L)
+{
+	struct key_def *key_def;
+	if (lua_gettop(L) != 3 ||
+	    (key_def = luaT_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 = luaT_key_def_check_tuple(L, key_def, 2)) == NULL)
+		return luaT_error(L);
+	if ((tuple_b = luaT_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;
+}
+
+/**
+ * Compare tuple with key using the key definition.
+ * Push 0  if key_fields(tuple) == parts(key)
+ *      <0 if key_fields(tuple) < parts(key)
+ *      >0 if key_fields(tuple) > parts(key)
+ * integer to a LUA stack on success.
+ * Raise error otherwise.
+ */
+static int
+lbox_key_def_compare_with_key(struct lua_State *L)
+{
+	struct key_def *key_def;
+	if (lua_gettop(L) != 3 ||
+	    (key_def = luaT_check_key_def(L, 1)) == NULL) {
+		return luaL_error(L, "Usage: key_def:"
+				     "compare_with_key(tuple, key)");
+	}
+
+	struct tuple *tuple = luaT_key_def_check_tuple(L, key_def, 2);
+	if (tuple == NULL)
+		return luaT_error(L);
+
+	struct region *region = &fiber()->gc;
+	size_t region_svp = region_used(region);
+	size_t key_len;
+	const char *key = lbox_encode_tuple_on_gc(L, 3, &key_len);
+	uint32_t part_count = mp_decode_array(&key);
+	if (key_validate_parts(key_def, key, part_count, true) != 0) {
+		region_truncate(region, region_svp);
+		tuple_unref(tuple);
+		return luaT_error(L);
+	}
+
+	int rc = tuple_compare_with_key(tuple, key, part_count, key_def);
+	region_truncate(region, region_svp);
+	tuple_unref(tuple);
+	lua_pushinteger(L, rc);
+	return 1;
+}
+
+/**
+ * Construct and export to LUA a new key definition with a set
+ * union of key parts from first and second key defs. Parts of
+ * the new key_def consist of the first key_def's parts and those
+ * parts of the second key_def that were not among the first
+ * parts.
+ * Push the new key_def as cdata to a LUA stack on success.
+ * Raise error otherwise.
+ */
+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 = luaT_check_key_def(L, 1)) == NULL ||
+	    (key_def_b = luaT_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,
+				CTID_STRUCT_KEY_DEF_REF) = new_key_def;
+	lua_pushcfunction(L, lbox_key_def_gc);
+	luaL_setcdatagc(L, -2);
+	return 1;
+}
+
+
+/**
+ * Push a new table representing a key_def to a Lua stack.
+ */
+static int
+lbox_key_def_to_table(struct lua_State *L)
+{
+	struct key_def *key_def;
+	if (lua_gettop(L) != 1 || (key_def = luaT_check_key_def(L, 1)) == NULL)
+		return luaL_error(L, "Usage: key_def:totable()");
+
+	luaT_push_key_def(L, key_def);
+	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);
+	}
+	if (part_count == 0) {
+		diag_set(IllegalParams, "Key definition can only be constructed"
+					" by using at least 1 key_part");
+		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);
+		}
+		lua_pop(L, 1);
+	}
+
+	struct key_def *key_def = key_def_new(parts, part_count);
+	region_truncate(region, region_svp);
+	if (key_def == NULL)
+		return luaT_error(L);
+
+	/*
+	 * Compare and extract key_def methods must work even with
+	 * tuples with omitted (optional) fields. As there is no
+	 * space format which would guarantee certain minimal
+	 * field_count, pass min_field_count = 0 to ensure that
+	 * functions will work correctly in such case.
+	 */
+	key_def_update_optionality(key_def, 0);
+
+	*(struct key_def **) luaL_pushcdata(L,
+				CTID_STRUCT_KEY_DEF_REF) = 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;");
+	CTID_STRUCT_KEY_DEF_REF = luaL_ctypeid(L, "struct key_def&");
+
+	/* Export C functions to Lua. */
+	static const struct luaL_Reg meta[] = {
+		{"new", lbox_key_def_new},
+		{"extract_key", lbox_key_def_extract_key},
+		{"compare", lbox_key_def_compare},
+		{"compare_with_key", lbox_key_def_compare_with_key},
+		{"merge", lbox_key_def_merge},
+		{"totable", lbox_key_def_to_table},
+		{NULL, NULL}
+	};
+	luaL_register_module(L, "key_def", meta);
+	return 1;
+}
diff --git a/src/box/lua/key_def.h b/src/box/lua/key_def.h
new file mode 100644
index 000000000..88b8a0019
--- /dev/null
+++ b/src/box/lua/key_def.h
@@ -0,0 +1,69 @@
+#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_def;
+
+/**
+ * Push a new table representing a key_def to a Lua stack.
+ * Table is consists of key_def::parts tables that describe
+ * each part correspondingly.
+ * The collation and path fields are optional so resulting
+ * object doesn't declare them where not necessary.
+ */
+void
+luaT_push_key_def(struct lua_State *L, const struct key_def *key_def);
+
+/**
+ * Check key_def pointer in LUA stack by specified index.
+ * The value by idx is expected to be key_def's cdata.
+ * Returns not NULL tuple pointer on success, NULL otherwise.
+ */
+struct key_def *
+luaT_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..586005709
--- /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.extract_key,
+    ['compare'] = key_def.compare,
+    ['compare_with_key'] = key_def.compare_with_key,
+    ['merge'] = key_def.merge,
+    ['totable'] = key_def.totable,
+    ['__serialize'] = key_def.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 93269b72b..100da0a79 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,38 +287,7 @@ lbox_fillspace(struct lua_State *L, struct space *space, int i)
 		lua_setfield(L, -2, "name");
 
 		lua_pushstring(L, "parts");
-		lua_newtable(L);
-
-		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");
-			}
-
-			lua_settable(L, -3); /* index[k].parts[j] */
-		}
+		luaT_push_key_def(L, index_def->key_def);
 
 		lua_settable(L, -3); /* space.index[k].parts */
 
diff --git a/src/box/tuple_extract_key.cc b/src/box/tuple_extract_key.cc
index 3c49094b8..d7507bc4e 100644
--- a/src/box/tuple_extract_key.cc
+++ b/src/box/tuple_extract_key.cc
@@ -422,3 +422,25 @@ tuple_key_contains_null(struct tuple *tuple, struct key_def *def)
 	}
 	return false;
 }
+
+int
+tuple_validate_key_parts(struct key_def *key_def, struct tuple *tuple)
+{
+	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) {
+			if (key_part_is_nullable(part))
+				continue;
+			diag_set(ClientError, ER_FIELD_MISSING,
+				 tt_sprintf("[%d]%.*s",
+					    part->fieldno + TUPLE_INDEX_BASE,
+					    part->path_len, part->path));
+			return -1;
+		}
+		if (key_part_validate(part->type, field, idx,
+				      key_part_is_nullable(part)) != 0)
+			return -1;
+	}
+	return 0;
+}
diff --git a/test/box-tap/key_def.test.lua b/test/box-tap/key_def.test.lua
new file mode 100755
index 000000000..c52ff48fe
--- /dev/null
+++ b/test/box-tap/key_def.test.lua
@@ -0,0 +1,453 @@
+#!/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 no key parts',
+        parts = {},
+        exp_err = 'Key definition can only be constructed by using at least 1 key_part',
+    },
+    {
+        "Pass a garbage instead of key parts",
+        parts = {fieldno = 1, type = 'unsigned'},
+        exp_err = 'Key definition can only be constructed by using at least 1 key_part',
+    },
+    {
+        '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 = 999,
+        }},
+        exp_err = coll_not_found(1, 999),
+    },
+    {
+        '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 = 'invalid path',
+    },
+    {
+        '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.
+
+-- Case: extract_key().
+test:test('extract_key()', function(test)
+    test:plan(13)
+
+    local key_def_a = key_def_lib.new({
+        {type = 'unsigned', fieldno = 1},
+    })
+    local key_def_b = key_def_lib.new({
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    })
+    local key_def_c = key_def_lib.new({
+        {type = 'scalar', fieldno = 2},
+        {type = 'scalar', fieldno = 1},
+        {type = 'string', fieldno = 4, is_nullable = true},
+    })
+    local tuple_a = box.tuple.new({1, 1, 22})
+
+    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 = 'Tuple field [2] required by space format is missing'
+    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 = 'Tuple field [2] required by space format is missing'
+    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 = 'Tuple field [2] required by space format is missing'
+    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')
+
+    local key_def = key_def_lib.new({
+        {type = 'number', fieldno = 1, path='a'},
+        {type = 'number', fieldno = 1, path='b'},
+        {type = 'number', fieldno = 1, path='c', is_nullable=true},
+        {type = 'number', fieldno = 3, is_nullable=true},
+    })
+    local ok, err = pcall(key_def.extract_key, key_def,
+                          box.tuple.new({1, 1, 22}))
+    test:is_deeply({ok, tostring(err)},
+                {false, 'Tuple field [1]a required by space format is missing'},
+                'invalid JSON structure')
+    test:is_deeply(key_def:extract_key({{a=1, b=2}, 1}):totable(),
+                   {1, 2, box.NULL, box.NULL},
+                   'tuple with optional parts - case 1')
+    test:is_deeply(key_def:extract_key({{a=1, b=2, c=3}, 1}):totable(),
+                   {1, 2, 3, box.NULL},
+                   'tuple with optional parts - case 2')
+    test:is_deeply(key_def:extract_key({{a=1, b=2}, 1, 3}):totable(),
+                   {1, 2, box.NULL, 3},
+                   'tuple with optional parts - case 3')
+end)
+
+-- Case: compare().
+test:test('compare()', function(test)
+    test:plan(8)
+
+    local key_def_a = key_def_lib.new({
+        {type = 'unsigned', fieldno = 1},
+    })
+    local key_def_b = key_def_lib.new({
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    })
+    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})
+
+    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_def_b = key_def_lib.new({
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    })
+    local tuple_a = box.tuple.new({1, 1, 22})
+
+    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 parts_a = {
+        {type = 'unsigned', fieldno = 1}
+    }
+    local parts_b = {
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    }
+    local key_def_a = key_def_lib.new(parts_a)
+    local key_def_b = key_def_lib.new(parts_b)
+
+    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 parts_a = {
+        {type = 'unsigned', fieldno = 1}
+    }
+    local parts_b = {
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    }
+    local key_def_a = key_def_lib.new(parts_a)
+    local key_def_b = key_def_lib.new(parts_b)
+
+    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 parts_a = {
+        {type = 'unsigned', fieldno = 1}
+    }
+    local parts_b = {
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    }
+    local key_def_a = key_def_lib.new(parts_a)
+    local key_def_b = key_def_lib.new(parts_b)
+
+    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_a = key_def_lib.new({
+        {type = 'unsigned', fieldno = 1},
+    })
+    local key_def_b = key_def_lib.new({
+        {type = 'number', fieldno = 2},
+        {type = 'number', fieldno = 3},
+    })
+    local key_def_c = key_def_lib.new({
+        {type = 'scalar', fieldno = 2},
+        {type = 'scalar', fieldno = 1},
+        {type = 'string', fieldno = 4, is_nullable = true},
+    })
+    local tuple_a = box.tuple.new({1, 1, 22})
+
+    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

Re: [tarantool-patches] Re: [PATCH v3 1/1] lua: add key_def lua module

[Alexander Turenko]

> /**
> * Check an existent tuple pointer in LUA stack by specified

Nit: Lua is not an acronym.

https://stackoverflow.com/a/13615976

> * index or attemt to construct it by LUA table.
> * Increase tuple's reference counter.
> * Returns not NULL tuple pointer on success, NULL otherwise.
> */
> static struct tuple *
> luaT_key_def_check_tuple(struct lua_State *L, struct key_def *key_def, int idx)

Re: [tarantool-patches] Re: [PATCH v3 1/1] lua: add key_def lua module

[Vladimir Davydov]

Pushed to master, thanks!