box: functional and multikey indexes

Tarantool functional and multikey indexes allows you to create
an index based on user-specified function. In a nutshell, this
capability allows you to have case insensitive searches or sorts,
search on complex equations, and extend the SQL language
efficiently by implementing your own functions and operators and
then searching on them.

Feature use hidden space _i_{index_name}_{space_name} containing
tuples of structure [{extractor_format}{pk_format}] and redefines
on_replace trigger for target space to fill this space. Index
object metamethods are monkeypatched to make indirect lookups in
ispace to retrieve required tuple primary key.

Closes #1260

src/box/CMakeLists.txt

@@ -13,6 +13,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/func_idx.lua)
 set(bin_sources)
 bin_source(bin_sources bootstrap.snap bootstrap.h)
 

src/box/alter.cc

@@ -54,6 +54,7 @@
 #include "version.h"
 #include "sequence.h"
 #include "sql.h"
+#include "lua/call.h"
 
 /**
  * chap-sha1 of empty string, i.e.
@@ -279,7 +280,13 @@ index_def_new_from_tuple(struct tuple *tuple, struct space *space)
 				 space->def->fields,
 				 space->def->field_count) != 0)
 		diag_raise();
-	key_def = key_def_new(part_def, part_count);
+	if (opts.func_code != NULL) {
+		struct index *index = space_index(space, 0);
+		assert(index != NULL);
+		key_def = key_def_dup(index->def->key_def);
+	} else {
+		key_def = key_def_new(part_def, part_count);
+	}
 	if (key_def == NULL)
 		diag_raise();
 	struct index_def *index_def =
@@ -1380,6 +1387,45 @@ UpdateSchemaVersion::alter(struct alter_space *alter)
     ++schema_version;
 }
 
+class PrepareFunctionalIndex: public AlterSpaceOp
+{
+public:
+	PrepareFunctionalIndex(struct alter_space * alter, uint32_t iid)
+		:AlterSpaceOp(alter), iid(iid), func_ref(0) {}
+	virtual ~PrepareFunctionalIndex();
+	virtual void prepare(struct alter_space *alter);
+	virtual void alter(struct alter_space *alter);
+	/** New index id. */
+	uint32_t iid;
+	/** Extractor function index. */
+	int32_t func_ref;
+};
+
+PrepareFunctionalIndex::~PrepareFunctionalIndex()
+{
+	if (func_ref != 0)
+		lua_func_delete(func_ref);
+}
+
+void
+PrepareFunctionalIndex::prepare(struct alter_space *alter)
+{
+	struct index *new_index = space_index(alter->new_space, iid);
+	assert(index_is_functional(new_index->def));
+	if (lua_func_new(new_index->def->opts.func_code, &func_ref) != 0)
+		diag_raise();
+}
+
+void
+PrepareFunctionalIndex::alter(struct alter_space *alter)
+{
+	struct index *index = space_index(alter->new_space, iid);
+	lua_func_idx_trigger_set(space_name(alter->new_space),
+				 index->def->name, &index->func_trigger_ref);
+	index->func_ref = func_ref;
+	func_ref = 0;
+}
+
 /* }}} */
 
 /**
@@ -2009,12 +2055,15 @@ on_replace_dd_index(struct trigger * /* trigger */, void *event)
 	}
 	/* Case 2: create an index, if it is simply created. */
 	if (old_index == NULL && new_tuple != NULL) {
+		struct index_def *index_def =
+			index_def_new_from_tuple(new_tuple, old_space);
 		alter_space_move_indexes(alter, 0, iid);
 		CreateIndex *create_index = new CreateIndex(alter);
-		create_index->new_index_def =
-			index_def_new_from_tuple(new_tuple, old_space);
+		create_index->new_index_def = index_def;
 		index_def_update_optionality(create_index->new_index_def,
 					     alter->new_min_field_count);
+		if (index_is_functional(index_def))
+			(void)new PrepareFunctionalIndex(alter, iid);
 	}
 	/* Case 3 and 4: check if we need to rebuild index data. */
 	if (old_index != NULL && new_tuple != NULL) {

src/box/index.cc

@@ -28,6 +28,7 @@
  * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  * SUCH DAMAGE.
  */
+#include "lua/call.h"
 #include "index.h"
 #include "tuple.h"
 #include "say.h"
@@ -493,6 +494,8 @@ index_create(struct index *index, struct engine *engine,
 	index->engine = engine;
 	index->def = def;
 	index->space_cache_version = space_cache_version;
+	index->func_ref = 0;
+	index->func_trigger_ref = 0;
 	return 0;
 }
 
@@ -505,7 +508,13 @@ index_delete(struct index *index)
 	 * the index is primary or secondary.
 	 */
 	struct index_def *def = index->def;
+	int32_t func_ref = index->func_ref;
+	int32_t func_trigger_ref = index->func_trigger_ref;
 	index->vtab->destroy(index);
+	if (func_ref != 0)
+		lua_func_delete(func_ref);
+	if (func_trigger_ref != 0)
+		lua_func_delete(func_trigger_ref);
 	index_def_delete(def);
 }
 

src/box/index.h

@@ -454,6 +454,10 @@ struct index {
 	struct engine *engine;
 	/* Description of a possibly multipart key. */
 	struct index_def *def;
+	/* Functional index extractor routine reference. */
+	int32_t func_ref;
+	/* Functional index trap trigger routine reference. */
+	int32_t func_trigger_ref;
 	/* Space cache version at the time of construction. */
 	uint32_t space_cache_version;
 };

src/box/index_def.c

@@ -48,8 +48,24 @@ const struct index_opts index_opts_default = {
 	/* .lsn                 = */ 0,
 	/* .sql                 = */ NULL,
 	/* .stat                = */ NULL,
+	/* .func_code           = */ NULL,
+	/* .pkey_offset         = */ 0,
 };
 
+static int
+pkey_offset_decode(const char **str, uint32_t len, char *opt, uint32_t errcode,
+		    uint32_t field_no)
+{
+	(void)errcode;
+	(void)field_no;
+	(void)str;
+	(void)opt;
+	*(uint32_t *)opt = len;
+	for (uint32_t i = 0; i < len; i++)
+		mp_next(str);
+	return 0;
+}
+
 const struct opt_def index_opts_reg[] = {
 	OPT_DEF("unique", OPT_BOOL, struct index_opts, is_unique),
 	OPT_DEF("dimension", OPT_INT64, struct index_opts, dimension),
@@ -62,6 +78,9 @@ const struct opt_def index_opts_reg[] = {
 	OPT_DEF("bloom_fpr", OPT_FLOAT, struct index_opts, bloom_fpr),
 	OPT_DEF("lsn", OPT_INT64, struct index_opts, lsn),
 	OPT_DEF("sql", OPT_STRPTR, struct index_opts, sql),
+	OPT_DEF("func_code", OPT_STRPTR, struct index_opts, func_code),
+	OPT_DEF_ARRAY("func_format", struct index_opts, pkey_offset,
+		      pkey_offset_decode),
 	OPT_END,
 };
 
@@ -114,13 +133,23 @@ index_def_new(uint32_t space_id, uint32_t iid, const char *name,
 		if (def->opts.sql == NULL) {
 			diag_set(OutOfMemory, strlen(opts->sql) + 1, "strdup",
 				 "def->opts.sql");
-			index_def_delete(def);
-			return NULL;
+			goto error;
+		}
+	}
+	if (opts->func_code != NULL) {
+		def->opts.func_code = strdup(opts->func_code);
+		if (def->opts.func_code == NULL) {
+			diag_set(OutOfMemory, strlen(opts->func_code) + 1,
+				 "strdup", "def->opts.func_code");
+			goto error;
 		}
 	}
 	/* Statistics are initialized separately. */
 	assert(opts->stat == NULL);
 	return def;
+error:
+	index_def_delete(def);
+	return NULL;
 }
 
 struct index_def *
@@ -153,18 +182,26 @@ index_def_dup(const struct index_def *def)
 		if (dup->opts.sql == NULL) {
 			diag_set(OutOfMemory, strlen(def->opts.sql) + 1,
 				 "strdup", "dup->opts.sql");
-			index_def_delete(dup);
-			return NULL;
+			goto error;
+		}
+	}
+	if (def->opts.func_code != NULL) {
+		dup->opts.func_code = strdup(def->opts.func_code);
+		if (def->opts.func_code == NULL) {
+			diag_set(OutOfMemory, strlen(def->opts.func_code) + 1,
+				 "strdup", "def->opts.func_code");
+			goto error;
 		}
 	}
 	if (def->opts.stat != NULL) {
 		dup->opts.stat = index_stat_dup(def->opts.stat);
-		if (dup->opts.stat == NULL) {
-			index_def_delete(dup);
-			return NULL;
-		}
+		if (dup->opts.stat == NULL)
+			goto error;
 	}
 	return dup;
+error:
+	index_def_delete(dup);
+	return NULL;
 }
 
 size_t
@@ -258,6 +295,14 @@ index_def_cmp(const struct index_def *key1, const struct index_def *key2)
 	if (index_opts_cmp(&key1->opts, &key2->opts))
 		return index_opts_cmp(&key1->opts, &key2->opts);
 
+	int rc = 0;
+	if ((rc = !!index_is_functional(key1) -
+		  !!index_is_functional(key2)) != 0)
+		return rc;
+	else if (index_is_functional(key1) &&
+		 (rc = strcmp(key1->opts.func_code, key2->opts.func_code)) != 0)
+		return rc;
+
 	return key_part_cmp(key1->key_def->parts, key1->key_def->part_count,
 			    key2->key_def->parts, key2->key_def->part_count);
 }
@@ -276,7 +321,8 @@ index_def_is_valid(struct index_def *index_def, const char *space_name)
 			 space_name, "primary key must be unique");
 		return false;
 	}
-	if (index_def->key_def->part_count == 0) {
+	if (index_def->key_def->part_count == 0 &&
+	    !index_is_functional(index_def)) {
 		diag_set(ClientError, ER_MODIFY_INDEX, index_def->name,
 			 space_name, "part count must be positive");
 		return false;

src/box/index_def.h

@@ -167,6 +167,13 @@ struct index_opts {
 	 * filled after running ANALYZE command.
 	 */
 	struct index_stat *stat;
+	/** Extractor function LUA code string. */
+	char *func_code;
+	/**
+	 * Offset of primary key in functional index ispace
+	 * tuples.
+	 */
+	uint32_t pkey_offset;
 };
 
 extern const struct index_opts index_opts_default;
@@ -187,6 +194,7 @@ index_opts_create(struct index_opts *opts)
 static inline void
 index_opts_destroy(struct index_opts *opts)
 {
+	free(opts->func_code);
 	free(opts->sql);
 	free(opts->stat);
 	TRASH(opts);
@@ -329,6 +337,12 @@ index_def_list_add(struct rlist *index_def_list, struct index_def *index_def)
 		rlist_add_tail_entry(index_def_list, index_def, link);
 }
 
+static inline bool
+index_is_functional(const struct index_def *index_def)
+{
+	return index_def->opts.func_code != NULL;
+}
+
 /**
  * Create a new index definition definition.
  *

src/box/lua/call.c

@@ -467,6 +467,49 @@ lbox_module_reload(lua_State *L)
 	return 0;
 }
 
+
+void
+lua_func_idx_trigger_set(const char *space_name, const char *idx_name,
+			 int32_t *trigger_ref)
+{
+	lua_State *L = lua_newthread(tarantool_L);
+	int coro_ref = luaL_ref(tarantool_L, LUA_REGISTRYINDEX);
+	lua_getglobal(L, "func_idx_trigger_set");
+	lua_pushstring(L, space_name);
+	lua_pushstring(L, idx_name);
+	lua_call(L, 2, 1);
+	*trigger_ref = luaL_ref(L, LUA_REGISTRYINDEX);
+	luaL_unref(tarantool_L, LUA_REGISTRYINDEX, coro_ref);
+}
+
+int
+lua_func_new(const char *func_code, int32_t *func_ref)
+{
+	lua_State *L = lua_newthread(tarantool_L);
+	int coro_ref = luaL_ref(tarantool_L, LUA_REGISTRYINDEX);
+	int rc = 0;
+	if ((rc = luaL_loadstring(L, func_code)) != 0) {
+		const char *err_descr =
+			rc == LUA_ERRSYNTAX ?
+			"syntax error during pre-compilation" :
+			rc == LUA_ERRMEM ? "memory allocation error" :
+			"unknow";
+		diag_set(ClientError, ER_CREATE_FUNCTION, "user-defined",
+			 err_descr);
+		return -1;
+	}
+	lua_call(L, 0, 1);
+	*func_ref = luaL_ref(L, LUA_REGISTRYINDEX);
+	luaL_unref(tarantool_L, LUA_REGISTRYINDEX, coro_ref);
+	return 0;
+}
+
+void
+lua_func_delete(uint32_t func_ref)
+{
+	luaL_unref(tarantool_L, LUA_REGISTRYINDEX, func_ref);
+}
+
 static const struct luaL_Reg boxlib_internal[] = {
 	{"call_loadproc",  lbox_call_loadproc},
 	{"sql_create_function",  lbox_sql_create_function},

src/box/lua/call.h

@@ -35,6 +35,8 @@
 extern "C" {
 #endif /* defined(__cplusplus) */
 
+#include <inttypes.h>
+
 struct lua_State;
 
 void
@@ -53,6 +55,16 @@ box_lua_call(struct call_request *request, struct port *port);
 int
 box_lua_eval(struct call_request *request, struct port *port);
 
+void
+lua_func_idx_trigger_set(const char *space_name, const char *idx_name,
+			 int32_t *trigger_ref);
+
+int
+lua_func_new(const char *func_code, int32_t *func_ref);
+
+void
+lua_func_delete(uint32_t func_ref);
+
 #if defined(__cplusplus)
 } /* extern "C" */
 #endif /* defined(__cplusplus) */