From 35c3192c2be1c5f2406a1d9df91f06d0464bda97 Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@tarantool.org>
Date: Tue, 17 Dec 2013 17:38:43 +0400
Subject: [PATCH] Add new files (broken out box_lua.cc).

---
 src/box/lua/index.cc | 405 +++++++++++++++++++++++++++
 src/box/lua/index.h  |  36 +++
 src/box/lua/tuple.cc | 637 +++++++++++++++++++++++++++++++++++++++++++
 src/box/lua/tuple.h  |  55 ++++
 4 files changed, 1133 insertions(+)
 create mode 100644 src/box/lua/index.cc
 create mode 100644 src/box/lua/index.h
 create mode 100644 src/box/lua/tuple.cc
 create mode 100644 src/box/lua/tuple.h

diff --git a/src/box/lua/index.cc b/src/box/lua/index.cc
new file mode 100644
index 0000000000..d5f70d085a
--- /dev/null
+++ b/src/box/lua/index.cc
@@ -0,0 +1,405 @@
+/*
+ * 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/lua/index.h"
+#include "lua/utils.h"
+#include "box/index.h"
+#include "box/space.h"
+#include "box/schema.h"
+#include "box/lua/tuple.h"
+#include "fiber.h"
+#include "tbuf.h"
+
+/** {{{ box.index Lua library: access to spaces and indexes
+ */
+
+static const char *indexlib_name = "box.index";
+static const char *iteratorlib_name = "box.index.iterator";
+
+/* Index userdata. */
+struct lbox_index
+{
+	Index *index;
+	/* space id. */
+	uint32_t id;
+	/* index id. */
+	uint32_t iid;
+	/* space cache version at the time of push. */
+	int sc_version;
+};
+
+static struct iterator *
+lbox_checkiterator(struct lua_State *L, int i)
+{
+	struct iterator **it = (struct iterator **)
+			luaL_checkudata(L, i, iteratorlib_name);
+	assert(it != NULL);
+	return *it;
+}
+
+static void
+lbox_pushiterator(struct lua_State *L, Index *index,
+		  struct iterator *it, enum iterator_type type,
+		  const char *key, size_t key_size, uint32_t part_count)
+{
+	struct lbox_iterator_udata {
+		struct iterator *it;
+		char key[];
+	};
+
+	struct lbox_iterator_udata *udata = (struct lbox_iterator_udata *)
+		lua_newuserdata(L, sizeof(*udata) + key_size);
+	luaL_getmetatable(L, iteratorlib_name);
+	lua_setmetatable(L, -2);
+
+	udata->it = it;
+	if (key) {
+		memcpy(udata->key, key, key_size);
+		key = udata->key;
+	}
+	key_validate(index->key_def, type, key, part_count);
+	index->initIterator(it, type, key, part_count);
+}
+
+static int
+lbox_iterator_gc(struct lua_State *L)
+{
+	struct iterator *it = lbox_checkiterator(L, -1);
+	it->free(it);
+	return 0;
+}
+
+static Index *
+lua_checkindex(struct lua_State *L, int i)
+{
+	struct lbox_index *index =
+		(struct lbox_index *) luaL_checkudata(L, i, indexlib_name);
+	assert(index != NULL);
+	if (index->sc_version != sc_version) {
+		index->index = index_find(space_find(index->id), index->iid);
+		index->sc_version = sc_version;
+	}
+	return index->index;
+}
+
+static int
+lbox_index_bind(struct lua_State *L)
+{
+	uint32_t id = (uint32_t) luaL_checkint(L, 1); /* get space id */
+	uint32_t iid = (uint32_t) luaL_checkint(L, 2); /* get index id in */
+	/* locate the appropriate index */
+	struct space *space = space_find(id);
+	Index *i = index_find(space, iid);
+
+	/* create a userdata object */
+	struct lbox_index *index = (struct lbox_index *)
+		lua_newuserdata(L, sizeof(struct lbox_index));
+	index->id = id;
+	index->iid = iid;
+	index->sc_version = sc_version;
+	index->index = i;
+	/* set userdata object metatable to indexlib */
+	luaL_getmetatable(L, indexlib_name);
+	lua_setmetatable(L, -2);
+
+	return 1;
+}
+
+static int
+lbox_index_tostring(struct lua_State *L)
+{
+	Index *index = lua_checkindex(L, 1);
+	lua_pushfstring(L, " index %d", (int) index_id(index));
+	return 1;
+}
+
+static int
+lbox_index_len(struct lua_State *L)
+{
+	Index *index = lua_checkindex(L, 1);
+	lua_pushinteger(L, index->size());
+	return 1;
+}
+
+static int
+lbox_index_part_count(struct lua_State *L)
+{
+	Index *index = lua_checkindex(L, 1);
+	lua_pushinteger(L, index->key_def->part_count);
+	return 1;
+}
+
+static int
+lbox_index_min(struct lua_State *L)
+{
+	Index *index = lua_checkindex(L, 1);
+	lbox_pushtuple(L, index->min());
+	return 1;
+}
+
+static int
+lbox_index_max(struct lua_State *L)
+{
+	Index *index = lua_checkindex(L, 1);
+	lbox_pushtuple(L, index->max());
+	return 1;
+}
+
+static int
+lbox_index_random(struct lua_State *L)
+{
+	if (lua_gettop(L) != 2 || lua_isnil(L, 2))
+		luaL_error(L, "Usage: index:random((uint32) rnd)");
+
+	Index *index = lua_checkindex(L, 1);
+	uint32_t rnd = lua_tointeger(L, 2);
+	lbox_pushtuple(L, index->random(rnd));
+	return 1;
+}
+
+
+/*
+ * Lua iterator over a Taratnool/Box index.
+ *
+ *	(iteration_state, tuple) = index.next(index, [params])
+ *
+ * When [params] are absent or nil
+ * returns a pointer to a new ALL iterator and
+ * to the first tuple (or nil, if the index is
+ * empty).
+ *
+ * When [params] is a userdata,
+ * i.e. we're inside an iteration loop, retrieves
+ * the next tuple from the iterator.
+ *
+ * Otherwise, [params] can be used to seed
+ * a new iterator with iterator type and
+ * type-specific arguments. For exaple,
+ * for GE iterator, a list of Lua scalars
+ * cann follow the box.index.GE: this will
+ * start iteration from the offset specified by
+ * the given (multipart) key.
+ *
+ * @return Returns an iterator object, either created
+ *         or taken from Lua stack.
+ */
+
+static inline struct iterator *
+lbox_create_iterator(struct lua_State *L)
+{
+	Index *index = lua_checkindex(L, 1);
+	int argc = lua_gettop(L);
+
+	/* Create a new iterator. */
+	enum iterator_type type = ITER_ALL;
+	uint32_t key_part_count = 0;
+	const char *key = NULL;
+	size_t key_size = 0;
+
+	if (argc > 1 && lua_type(L, 2) != LUA_TNIL) {
+		type = (enum iterator_type) luaL_checkint(L, 2);
+		if (type < ITER_ALL || type >= iterator_type_MAX)
+			luaL_error(L, "unknown iterator type: %d", type);
+	}
+
+	RegionGuard region_guard(&fiber->gc);
+
+	/* What else do we have on the stack? */
+	if (argc > 2 && (lua_type(L, 3) != LUA_TNIL)) {
+		/* Single or multi- part key. */
+		struct tbuf *b = tbuf_new(&fiber->gc);
+		luamp_encodestack(L, b, 3, argc);
+		key = b->data;
+		assert(b->size > 0);
+		if (unlikely(mp_typeof(*key) != MP_ARRAY))
+			tnt_raise(ClientError, ER_TUPLE_NOT_ARRAY);
+		key_part_count = mp_decode_array(&key);
+		key_size = b->data + b->size - key;
+	}
+
+	struct iterator *it = index->allocIterator();
+	lbox_pushiterator(L, index, it, type, key, key_size, key_part_count);
+	return it;
+}
+
+/**
+ * Lua-style next() function, for use in pairs().
+ * @example:
+ * for k, v in box.space[0].index[0].idx.next, box.space[0].index[0].idx, nil do
+ *	print(v)
+ * end
+ */
+static int
+lbox_index_next(struct lua_State *L)
+{
+	int argc = lua_gettop(L);
+	struct iterator *it = NULL;
+	if (argc == 2 && lua_type(L, 2) == LUA_TUSERDATA) {
+		/*
+		 * Apart from the index itself, we have only one
+		 * other argument, and it's a userdata: must be
+		 * iteration state created before.
+		 */
+		it = lbox_checkiterator(L, 2);
+	} else {
+		it = lbox_create_iterator(L);
+	}
+	struct tuple *tuple = it->next(it);
+	/* If tuple is NULL, pushes nil as end indicator. */
+	lbox_pushtuple(L, tuple);
+	return tuple ? 2 : 1;
+}
+
+/** iterator() closure function. */
+static int
+lbox_index_iterator_closure(struct lua_State *L)
+{
+	/* Extract closure arguments. */
+	struct iterator *it = lbox_checkiterator(L, lua_upvalueindex(1));
+
+	struct tuple *tuple = it->next(it);
+
+	/* If tuple is NULL, push nil as end indicator. */
+	lbox_pushtuple(L, tuple);
+	return 1;
+}
+
+/**
+ * @brief Create iterator closure over a Taratnool/Box index.
+ * @example lua it = box.space[0].index[0]:iterator(box.index.GE, 1);
+ *   print(it(), it()).
+ * @param L lua stack
+ * @see http://www.lua.org/pil/7.1.html
+ * @return number of return values put on the stack
+ */
+static int
+lbox_index_iterator(struct lua_State *L)
+{
+	/* Create iterator and push it onto the stack. */
+	(void) lbox_create_iterator(L);
+	lua_pushcclosure(L, &lbox_index_iterator_closure, 1);
+	return 1;
+}
+
+
+/**
+ * Lua index subtree count function.
+ * Iterate over an index, count the number of tuples which equal the
+ * provided search criteria. The argument can either point to a
+ * tuple, a key, or one or more key parts. Returns the number of matched
+ * tuples.
+ */
+static int
+lbox_index_count(struct lua_State *L)
+{
+	Index *index = lua_checkindex(L, 1);
+	int argc = lua_gettop(L);
+	if (argc == 0)
+		luaL_error(L, "index.count(): one or more arguments expected");
+
+	/* preparing single or multi-part key */
+	if (argc == 1 || (argc == 2 && lua_type(L, 2) == LUA_TNIL)) {
+		/* Nothing */
+		/* Return index size */
+		lua_pushnumber(L, index->size());
+		return 1;
+	}
+
+	RegionGuard region_guard(&fiber->gc);
+	struct tbuf *b = tbuf_new(&fiber->gc);
+
+	/* Single or multi- part key. */
+	luamp_encodestack(L, b, 2, argc);
+
+	const char *key = b->data;
+	if (unlikely(mp_typeof(*key) != MP_ARRAY))
+		tnt_raise(ClientError, ER_TUPLE_NOT_ARRAY);
+	uint32_t part_count = mp_decode_array(&key);
+	key_validate(index->key_def, ITER_EQ, key, part_count);
+
+	/* Prepare index iterator */
+	struct iterator *it = index->position();
+	index->initIterator(it, ITER_EQ, key, part_count);
+	/* Iterate over the index and count tuples. */
+	struct tuple *tuple;
+	uint32_t count = 0;
+	while ((tuple = it->next(it)) != NULL)
+		count++;
+
+	/* Return subtree size */
+	lua_pushnumber(L, count);
+	return 1;
+}
+
+static const struct luaL_reg lbox_index_meta[] = {
+	{"__tostring", lbox_index_tostring},
+	{"__len", lbox_index_len},
+	{"part_count", lbox_index_part_count},
+	{"min", lbox_index_min},
+	{"max", lbox_index_max},
+	{"random", lbox_index_random},
+	{"next", lbox_index_next},
+	{"iterator", lbox_index_iterator},
+	{"count", lbox_index_count},
+	{NULL, NULL}
+};
+
+static const struct luaL_reg indexlib [] = {
+	{"bind", lbox_index_bind},
+	{NULL, NULL}
+};
+
+static const struct luaL_reg lbox_iterator_meta[] = {
+	{"__gc", lbox_iterator_gc},
+	{NULL, NULL}
+};
+
+
+static void
+box_index_init_iterator_types(struct lua_State *L, int idx)
+{
+	for (int i = 0; i < iterator_type_MAX; i++) {
+		assert(strncmp(iterator_type_strs[i], "ITER_", 5) == 0);
+		lua_pushnumber(L, i);
+		/* cut ITER_ prefix from enum name */
+		lua_setfield(L, idx, iterator_type_strs[i] + 5);
+	}
+}
+
+/* }}} */
+
+void
+box_lua_index_init(struct lua_State *L)
+{
+	/* box.index */
+	luaL_register_type(L, indexlib_name, lbox_index_meta);
+	luaL_register(L, "box.index", indexlib);
+	box_index_init_iterator_types(L, -2);
+	lua_pop(L, 1);
+	luaL_register_type(L, iteratorlib_name, lbox_iterator_meta);
+}
diff --git a/src/box/lua/index.h b/src/box/lua/index.h
new file mode 100644
index 0000000000..a16fba724a
--- /dev/null
+++ b/src/box/lua/index.h
@@ -0,0 +1,36 @@
+#ifndef INCLUDES_TARANTOOL_BOX_LUA_INDEX_H
+#define INCLUDES_TARANTOOL_BOX_LUA_INDEX_H
+/*
+ * 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.
+ */
+struct lua_State;
+
+void
+box_lua_index_init(struct lua_State *L);
+
+#endif /* INCLUDES_TARANTOOL_BOX_LUA_INDEX_H */
diff --git a/src/box/lua/tuple.cc b/src/box/lua/tuple.cc
new file mode 100644
index 0000000000..e241e5c67f
--- /dev/null
+++ b/src/box/lua/tuple.cc
@@ -0,0 +1,637 @@
+/*
+ * 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/lua/tuple.h"
+#include "box/tuple.h"
+#include "box/tuple_update.h"
+#include "fiber.h"
+#include "tbuf.h"
+#include "lua/utils.h"
+#include "lua/msgpack.h"
+#include "third_party/lua-yaml/lyaml.h"
+
+/** {{{ box.tuple Lua library
+ *
+ * To avoid extra copying between Lua memory and garbage-collected
+ * tuple memory, provide a Lua userdata object 'box.tuple'.  This
+ * object refers to a tuple instance in the slab allocator, and
+ * allows accessing it using Lua primitives (array subscription,
+ * iteration, etc.). When Lua object is garbage-collected,
+ * tuple reference counter in the slab allocator is decreased,
+ * allowing the tuple to be eventually garbage collected in
+ * the slab allocator.
+ */
+
+static const char *tuplelib_name = "box.tuple";
+static const char *tuple_iteratorlib_name = "box.tuple.iterator";
+static int tuple_totable_mt_ref = 0; /* a precreated metable for totable() */
+
+
+static inline struct tuple *
+lua_checktuple(struct lua_State *L, int narg)
+{
+	struct tuple *t = *(struct tuple **) luaL_checkudata(L, narg, tuplelib_name);
+	assert(t->refs);
+	return t;
+}
+
+struct tuple *
+lua_istuple(struct lua_State *L, int narg)
+{
+	if (lua_getmetatable(L, narg) == 0)
+		return NULL;
+	luaL_getmetatable(L, tuplelib_name);
+	struct tuple *tuple = 0;
+	if (lua_equal(L, -1, -2))
+		tuple = *(struct tuple **) lua_touserdata(L, narg);
+	lua_pop(L, 2);
+	return tuple;
+}
+
+static int
+lbox_tuple_new(lua_State *L)
+{
+	int argc = lua_gettop(L);
+	if (unlikely(argc < 1)) {
+		lua_newtable(L); /* create an empty tuple */
+		++argc;
+	}
+	struct tuple *tuple = lua_totuple(L, 1, argc);
+	lbox_pushtuple(L, tuple);
+	return 1;
+}
+
+static int
+lbox_tuple_gc(struct lua_State *L)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	tuple_ref(tuple, -1);
+	return 0;
+}
+
+static int
+lbox_tuple_len(struct lua_State *L)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	lua_pushnumber(L, tuple_arity(tuple));
+	return 1;
+}
+
+static int
+lbox_tuple_slice(struct lua_State *L)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	int argc = lua_gettop(L) - 1;
+	uint32_t start, end;
+	int offset;
+
+	/*
+	 * Prepare the range. The second argument is optional.
+	 * If the end is beyond tuple size, adjust it.
+	 * If no arguments, or start > end, return an error.
+	 */
+	if (argc == 0 || argc > 2)
+		luaL_error(L, "tuple.slice(): bad arguments");
+
+	uint32_t arity = tuple_arity(tuple);
+	offset = lua_tointeger(L, 2);
+	if (offset >= 0 && offset < arity) {
+		start = offset;
+	} else if (offset < 0 && -offset <= arity) {
+		start = offset + arity;
+	} else {
+		return luaL_error(L, "tuple.slice(): start >= field count");
+	}
+
+	if (argc == 2) {
+		offset = lua_tointeger(L, 3);
+		if (offset > 0 && offset <= arity) {
+			end = offset;
+		} else if (offset < 0 && -offset < arity) {
+			end = offset + arity;
+		} else {
+			return luaL_error(L, "tuple.slice(): end > field count");
+		}
+	} else {
+		end = arity;
+	}
+	if (end <= start)
+		return luaL_error(L, "tuple.slice(): start must be less than end");
+
+	struct tuple_iterator it;
+	tuple_rewind(&it, tuple);
+	const char *field;
+
+	assert(start < arity);
+	uint32_t field_no = start;
+	field = tuple_seek(&it, start);
+	while (field && field_no < end) {
+		luamp_decode(L, &field);
+		++field_no;
+		field = tuple_next(&it);
+	}
+	assert(field_no == end);
+	return end - start;
+}
+
+/* A MsgPack extensions handler that supports tuples */
+static void
+luamp_encode_extension_box(struct lua_State *L, int idx, struct tbuf *b)
+{
+	if (lua_type(L, idx) == LUA_TUSERDATA &&
+			lua_istuple(L, idx)) {
+		struct tuple *tuple = lua_checktuple(L, idx);
+		tuple_to_tbuf(tuple, b);
+		return;
+	}
+
+	tnt_raise(ClientError, ER_PROC_RET,
+		  lua_typename(L, lua_type(L, idx)));
+}
+
+/*
+ * A luamp_encode wrapper to support old Tarantool 1.5 API.
+ * Will be removed after API change.
+ */
+int
+luamp_encodestack(struct lua_State *L, struct tbuf *b, int first, int last)
+{
+	if (first == last && (lua_istable(L, first) ||
+	    (lua_isuserdata(L, first) && lua_istuple(L, first)))) {
+		/* New format */
+		luamp_encode(L, b, first);
+		return 1;
+	} else {
+		/* Backward-compatible format */
+		/* sic: if arg_count is 0, first > last */
+		luamp_encode_array(b, last + 1 - first);
+		for (int k = first; k <= last; ++k) {
+			luamp_encode(L, b, k);
+		}
+		return last + 1 - first;
+	}
+}
+
+static void *
+tuple_update_region_alloc(void *alloc_ctx, size_t size)
+{
+	return region_alloc((struct region *) alloc_ctx, size);
+}
+
+/**
+ * Tuple transforming function.
+ *
+ * Remove the fields designated by 'offset' and 'len' from an tuple,
+ * and replace them with the elements of supplied data fields,
+ * if any.
+ *
+ * Function returns newly allocated tuple.
+ * It does not change any parent tuple data.
+ */
+static int
+lbox_tuple_transform(struct lua_State *L)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	int argc = lua_gettop(L);
+	if (argc < 3)
+		luaL_error(L, "tuple.transform(): bad arguments");
+	lua_Integer offset = lua_tointeger(L, 2);  /* Can be negative and can be > INT_MAX */
+	lua_Integer field_count = lua_tointeger(L, 3);
+
+	uint32_t arity = tuple_arity(tuple);
+	/* validate offset and len */
+	if (offset < 0) {
+		if (-offset > arity)
+			luaL_error(L, "tuple.transform(): offset is out of bound");
+		offset += arity;
+	} else if (offset > arity) {
+		offset = arity;
+	}
+	if (field_count < 0)
+		luaL_error(L, "tuple.transform(): len is negative");
+	if (field_count > arity - offset)
+		field_count = arity - offset;
+
+	assert(offset + field_count <= arity);
+
+	/*
+	 * Calculate the number of operations and length of UPDATE expression
+	 */
+	uint32_t op_cnt = 0;
+	if (offset < arity && field_count > 0)
+		op_cnt++;
+	if (argc > 3)
+		op_cnt += argc - 3;
+
+	if (op_cnt == 0) {
+		/* tuple_update() does not accept an empty operation list. */
+		lbox_pushtuple(L, tuple);
+		return 1;
+	}
+
+	RegionGuard region_guard(&fiber->gc);
+
+	/*
+	 * Prepare UPDATE expression
+	 */
+	struct tbuf *b = tbuf_new(&fiber->gc);
+	tbuf_append(b, (char *) &op_cnt, sizeof(op_cnt));
+	if (field_count > 0) {
+		tbuf_ensure(b, sizeof(uint32_t) + 1 + 9);
+
+		/* offset */
+		char *data = pack_u32(b->data + b->size, offset);
+
+		/* operation */
+		*data++ = UPDATE_OP_DELETE;
+
+		assert(data <= b->data + b->capacity);
+		b->size = data - b->data;
+
+		/* field: count */
+		luamp_encode_uint(b, field_count);
+	}
+
+	for (int i = argc ; i > 3; i--) {
+		tbuf_ensure(b, sizeof(uint32_t) + 1 + 10);
+
+		/* offset */
+		char *data = pack_u32(b->data + b->size, offset);
+
+		/* operation */
+		*data++ = UPDATE_OP_INSERT;
+
+		assert(data <= b->data + b->capacity);
+		b->size = data - b->data;
+
+		/* field */
+		luamp_encode(L, b, i);
+	}
+
+	/* Execute tuple_update */
+	struct tuple *new_tuple = tuple_update(tuple_format_ber,
+					       tuple_update_region_alloc,
+					       &fiber->gc,
+					       tuple, tbuf_str(b), tbuf_end(b));
+	lbox_pushtuple(L, new_tuple);
+	return 1;
+}
+
+/*
+ * Tuple find function.
+ *
+ * Find each or one tuple field according to the specified key.
+ *
+ * Function returns indexes of the tuple fields that match
+ * key criteria.
+ *
+ */
+static int
+lbox_tuple_find_do(struct lua_State *L, bool all)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	int argc = lua_gettop(L);
+	size_t offset = 0;
+	switch (argc - 1) {
+	case 1: break;
+	case 2:
+		offset = lua_tointeger(L, 2);
+		break;
+	default:
+		luaL_error(L, "tuple.find(): bad arguments");
+	}
+
+	int top = lua_gettop(L);
+	int idx = offset;
+
+	struct luaL_field arg;
+	luaL_checkfield(L, 2, &arg);
+	struct tuple_iterator it;
+	tuple_rewind(&it, tuple);
+	const char *field = tuple_seek(&it, idx);
+	for (; field; field = tuple_next(&it), idx++) {
+		bool found = false;
+		const char *f = field;
+		if (arg.type != mp_typeof(*field))
+			continue;
+
+		switch (arg.type) {
+		case MP_UINT:
+			found = (arg.ival == mp_decode_uint(&f));
+			break;
+		case MP_INT:
+			found = (arg.ival == mp_decode_int(&f));
+			break;
+		case MP_BOOL:
+			found = (arg.bval == mp_decode_bool(&f));
+			break;
+		case MP_DOUBLE:
+			found = (arg.bval == mp_decode_double(&f));
+			break;
+		case MP_STR:
+		{
+			uint32_t len1 = 0;
+			const char *s1 = mp_decode_str(&f, &len1);
+			size_t len2 = arg.sval.len;
+			const char *s2 = arg.sval.data;
+			found = (len1 == len2) && (memcmp(s1, s2, len1) == 0);
+			break;
+		}
+		default:
+			break;
+		}
+		if (found) {
+			lua_pushinteger(L, idx);
+			if (!all)
+				break;
+		}
+	}
+	return lua_gettop(L) - top;
+}
+
+static int
+lbox_tuple_find(struct lua_State *L)
+{
+	return lbox_tuple_find_do(L, false);
+}
+
+static int
+lbox_tuple_findall(struct lua_State *L)
+{
+	return lbox_tuple_find_do(L, true);
+}
+
+static int
+lbox_tuple_unpack(struct lua_State *L)
+{
+	int argc = lua_gettop(L);
+	(void) argc;
+	struct tuple *tuple = lua_checktuple(L, 1);
+
+	struct tuple_iterator it;
+	tuple_rewind(&it, tuple);
+	const char *field;
+	while ((field = tuple_next(&it)))
+		luamp_decode(L, &field);
+
+	assert(lua_gettop(L) == argc + tuple_arity(tuple));
+	(void) argc;
+	return tuple_arity(tuple);
+}
+
+static int
+lbox_tuple_totable(struct lua_State *L)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	lua_newtable(L);
+	int index = 1;
+
+	struct tuple_iterator it;
+	tuple_rewind(&it, tuple);
+	const char *field;
+	while ((field = tuple_next(&it))) {
+		lua_pushnumber(L, index++);
+		luamp_decode(L, &field);
+		lua_rawset(L, -3);
+	}
+
+	/* Hint serializer */
+	assert(tuple_totable_mt_ref != 0);
+	lua_rawgeti(L, LUA_REGISTRYINDEX, tuple_totable_mt_ref);
+	lua_setmetatable(L, -2);
+
+	return 1;
+}
+
+/**
+ * Implementation of tuple __index metamethod.
+ *
+ * Provides operator [] access to individual fields for integer
+ * indexes, as well as searches and invokes metatable methods
+ * for strings.
+ */
+static int
+lbox_tuple_index(struct lua_State *L)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	/* For integer indexes, implement [] operator */
+	if (lua_isnumber(L, 2)) {
+		int i = luaL_checkint(L, 2);
+		const char *field = tuple_field(tuple, i);
+		if (field == NULL) {
+			const char *data = tuple->data;
+			luaL_error(L, "%s: index %d is out of bounds (0..%d)",
+				   tuplelib_name, i, mp_decode_array(&data));
+		}
+		luamp_decode(L, &field);
+		return 1;
+	}
+
+	/* If we got a string, try to find a method for it. */
+	const char *sz = luaL_checkstring(L, 2);
+	lua_getmetatable(L, 1);
+	lua_getfield(L, -1, sz);
+	return 1;
+}
+
+static int
+lbox_tuple_tostring(struct lua_State *L)
+{
+	/*
+	 * The method does next things:
+	 * 1. Calls :unpack
+	 * 2. Serializes the result using yaml
+	 * 3. Strips start and end of yaml document symbols
+	 */
+
+	/* unpack */
+	lbox_tuple_totable(L);
+
+	/* serialize */
+	lua_replace(L, 1);
+	yamlL_encode(L);
+
+	/* strip yaml tags */
+	size_t len;
+	const char *str = lua_tolstring(L, -1, &len);
+	assert(strlen(str) == len);
+	const char *s = index(str, '[');
+	const char *e = rindex(str, ']');
+	assert(s != NULL && e != NULL && s + 1 <= e);
+	lua_pushlstring(L, s, e - s + 1);
+	return 1;
+}
+
+void
+lbox_pushtuple(struct lua_State *L, struct tuple *tuple)
+{
+	if (tuple) {
+		struct tuple **ptr = (struct tuple **)
+				lua_newuserdata(L, sizeof(*ptr));
+		luaL_getmetatable(L, tuplelib_name);
+		lua_setmetatable(L, -2);
+		*ptr = tuple;
+		tuple_ref(tuple, 1);
+	} else {
+		lua_pushnil(L);
+	}
+}
+
+/**
+ * Sequential access to tuple fields. Since tuple is a list-like
+ * structure, iterating over tuple fields is faster
+ * than accessing fields using an index.
+ */
+static int
+lbox_tuple_next(struct lua_State *L)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	int argc = lua_gettop(L) - 1;
+
+	struct tuple_iterator *it = NULL;
+	if (argc == 0 || (argc == 1 && lua_type(L, 2) == LUA_TNIL)) {
+		it = (struct tuple_iterator *) lua_newuserdata(L, sizeof(*it));
+		assert(it != NULL);
+		luaL_getmetatable(L, tuple_iteratorlib_name);
+		lua_setmetatable(L, -2);
+		tuple_rewind(it, tuple);
+	} else if (argc == 1 && lua_type(L, 2) == LUA_TUSERDATA) {
+		it = (struct tuple_iterator *)
+			luaL_checkudata(L, 2, tuple_iteratorlib_name);
+		assert(it != NULL);
+		lua_pushvalue(L, 2);
+	} else {
+		return luaL_error(L, "tuple.next(): bad arguments");
+	}
+
+	const char *field = tuple_next(it);
+	if (field == NULL) {
+		lua_pop(L, 1);
+		lua_pushnil(L);
+		return 1;
+	}
+
+	luamp_decode(L, &field);
+	return 2;
+}
+
+/** Iterator over tuple fields. Adapt lbox_tuple_next
+ * to Lua iteration conventions.
+ */
+static int
+lbox_tuple_pairs(struct lua_State *L)
+{
+	lua_pushcfunction(L, lbox_tuple_next);
+	lua_pushvalue(L, -2); /* tuple */
+	lua_pushnil(L);
+	return 3;
+}
+
+
+/** tuple:bsize()
+ *
+ */
+static int
+lbox_tuple_bsize(struct lua_State *L)
+{
+	struct tuple *tuple = lua_checktuple(L, 1);
+	lua_pushnumber(L, tuple->bsize);
+	return 1;
+}
+
+static const struct luaL_reg lbox_tuple_meta[] = {
+	{"__gc", lbox_tuple_gc},
+	{"__len", lbox_tuple_len},
+	{"__index", lbox_tuple_index},
+	{"__tostring", lbox_tuple_tostring},
+	{"next", lbox_tuple_next},
+	{"pairs", lbox_tuple_pairs},
+	{"slice", lbox_tuple_slice},
+	{"transform", lbox_tuple_transform},
+	{"find", lbox_tuple_find},
+	{"findall", lbox_tuple_findall},
+	{"unpack", lbox_tuple_unpack},
+	{"totable", lbox_tuple_totable},
+	{"bsize", lbox_tuple_bsize},
+	{NULL, NULL}
+};
+
+static const struct luaL_reg lbox_tuplelib[] = {
+	{"new", lbox_tuple_new},
+	{NULL, NULL}
+};
+
+static const struct luaL_reg lbox_tuple_iterator_meta[] = {
+	{NULL, NULL}
+};
+
+
+struct tuple*
+lua_totuple(struct lua_State *L, int first, int last)
+{
+	RegionGuard region_guard(&fiber->gc);
+	struct tbuf *b = tbuf_new(&fiber->gc);
+	try {
+		luamp_encodestack(L, b, first, last);
+	} catch (const Exception &e) {
+		throw;
+	} catch (...) {
+		tnt_raise(ClientError, ER_PROC_LUA, lua_tostring(L, -1));
+	}
+	const char *data = b->data;
+	if (unlikely(mp_typeof(*data) != MP_ARRAY))
+		tnt_raise(ClientError, ER_TUPLE_NOT_ARRAY);
+	struct tuple *tuple = tuple_new(tuple_format_ber, &data, tbuf_end(b));
+	return tuple;
+}
+
+/* }}} */
+
+void
+box_lua_tuple_init(struct lua_State *L)
+{
+	luaL_register_type(L, tuplelib_name, lbox_tuple_meta);
+	luaL_register_type(L, tuple_iteratorlib_name,
+			   lbox_tuple_iterator_meta);
+	luaL_register(L, tuplelib_name, lbox_tuplelib);
+	lua_pop(L, 1);
+
+	luamp_set_encode_extension(luamp_encode_extension_box);
+
+	/* Precreate a metatable for tuple_unpack */
+	lua_newtable(L);
+	lua_pushstring(L, "_serializer_compact");
+	lua_pushboolean(L, true);
+	lua_settable(L, -3);
+	lua_pushstring(L, "_serializer_type");
+	lua_pushstring(L, "array");
+	lua_settable(L, -3);
+	tuple_totable_mt_ref = luaL_ref(L, LUA_REGISTRYINDEX);
+	assert(tuple_totable_mt_ref != 0);
+}
+
diff --git a/src/box/lua/tuple.h b/src/box/lua/tuple.h
new file mode 100644
index 0000000000..bdcb840840
--- /dev/null
+++ b/src/box/lua/tuple.h
@@ -0,0 +1,55 @@
+#ifndef INCLUDES_TARANTOOL_MOD_BOX_LUA_TUPLE_H
+#define INCLUDES_TARANTOOL_MOD_BOX_LUA_TUPLE_H
+/*
+ * 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.
+ */
+struct lua_State;
+struct txn;
+struct tuple;
+struct tbuf;
+
+/**
+ * Push tuple on lua stack
+ */
+void
+lbox_pushtuple(struct lua_State *L, struct tuple *tuple);
+
+
+struct tuple *lua_istuple(struct lua_State *L, int narg);
+
+struct tuple*
+lua_totuple(struct lua_State *L, int first, int last);
+
+int
+luamp_encodestack(struct lua_State *L, struct tbuf *b,
+		  int first, int last);
+
+void
+box_lua_tuple_init(struct lua_State *L);
+
+#endif /* INCLUDES_TARANTOOL_MOD_BOX_LUA_TUPLE_H */
-- 
GitLab