sql: move Triggers to server

Introduced sql_triggers field in space structure.
Changed parser logic to do not insert built triggers, just only
to do parsing. All triggers insertions and deletions are operated
via on_replace_dd_trigger now.
Global DB hash has been kept as we still ned to get Trigger AST
by name on trigger deletion.

Resolves #3273.

src/box/alter.cc

@@ -553,6 +553,10 @@ space_swap_triggers(struct space *new_space, struct space *old_space)
 	rlist_swap(&new_space->before_replace, &old_space->before_replace);
 	rlist_swap(&new_space->on_replace, &old_space->on_replace);
 	rlist_swap(&new_space->on_stmt_begin, &old_space->on_stmt_begin);
+	/** Swap SQL Triggers pointer. */
+	struct Trigger *new_value = new_space->sql_triggers;
+	new_space->sql_triggers = old_space->sql_triggers;
+	old_space->sql_triggers = new_value;
 }
 
 /**
@@ -3249,6 +3253,57 @@ lock_before_dd(struct trigger *trigger, void *event)
 	latch_lock(&schema_lock);
 }
 
+/**
+ * Trigger invoked on rollback in the _trigger space.
+ * Since rollback trigger is invoked after insertion to hash and space,
+ * we have to delete it from those structures and release memory.
+ * Vice versa, after deletion of trigger we must return it back to hash and space*.
+ */
+static void
+on_replace_trigger_rollback(struct trigger *trigger, void *event)
+{
+	struct txn_stmt *stmt = txn_last_stmt((struct txn*) event);
+	struct Trigger *old_trigger = (struct Trigger *)trigger->data;
+	struct Trigger *new_trigger;
+
+	if (stmt->old_tuple != NULL && stmt->new_tuple == NULL) {
+		/* Rollback DELETE trigger. */
+		if (sql_trigger_replace(sql_get(),
+					sql_trigger_name(old_trigger),
+					old_trigger, &new_trigger) != 0)
+			panic("Out of memory on insertion into trigger hash");
+		assert(new_trigger == NULL);
+	}  else if (stmt->new_tuple != NULL && stmt->old_tuple == NULL) {
+		/* Rollback INSERT trigger. */
+		int rc = sql_trigger_replace(sql_get(),
+					     sql_trigger_name(old_trigger),
+					     NULL, &new_trigger);
+		(void)rc;
+		assert(rc == 0);
+		assert(new_trigger == old_trigger);
+		sql_trigger_delete(sql_get(), new_trigger);
+	} else {
+		/* Rollback REPLACE trigger. */
+		if (sql_trigger_replace(sql_get(),
+					sql_trigger_name(old_trigger),
+					old_trigger, &new_trigger) != 0)
+			panic("Out of memory on insertion into trigger hash");
+		assert(old_trigger != new_trigger);
+		sql_trigger_delete(sql_get(), new_trigger);
+	}
+}
+
+/**
+ * Trigger invoked on commit in the _trigger space.
+ * Drop useless old sql_trigger AST object if any.
+ */
+static void
+on_replace_trigger_commit(struct trigger *trigger, void * /* event */)
+{
+	struct Trigger *old_trigger = (struct Trigger *)trigger->data;
+	sql_trigger_delete(sql_get(), old_trigger);
+}
+
 /**
  * A trigger invoked on replace in a space containing
  * SQL triggers.
@@ -3258,6 +3313,88 @@ on_replace_dd_trigger(struct trigger * /* trigger */, void *event)
 {
 	struct txn *txn = (struct txn *) event;
 	txn_check_singlestatement_xc(txn, "Space _trigger");
+	struct txn_stmt *stmt = txn_current_stmt(txn);
+	struct tuple *old_tuple = stmt->old_tuple;
+	struct tuple *new_tuple = stmt->new_tuple;
+
+	struct trigger *on_rollback =
+		txn_alter_trigger_new(on_replace_trigger_rollback, NULL);
+	struct trigger *on_commit =
+		txn_alter_trigger_new(on_replace_trigger_commit, NULL);
+
+	if (old_tuple != NULL && new_tuple == NULL) {
+		/* DROP trigger. */
+		uint32_t trigger_name_len;
+		const char *trigger_name_src =
+			tuple_field_str_xc(old_tuple, BOX_TRIGGER_FIELD_NAME,
+					   &trigger_name_len);
+		char *trigger_name =
+			(char *)region_alloc_xc(&fiber()->gc,
+						trigger_name_len + 1);
+		memcpy(trigger_name, trigger_name_src, trigger_name_len);
+		trigger_name[trigger_name_len] = 0;
+
+		struct Trigger *old_trigger;
+		int rc = sql_trigger_replace(sql_get(), trigger_name, NULL,
+					     &old_trigger);
+		(void)rc;
+		assert(rc == 0);
+		assert(old_trigger != NULL);
+
+		on_commit->data = old_trigger;
+		on_rollback->data = old_trigger;
+	} else {
+		/* INSERT, REPLACE trigger. */
+		uint32_t trigger_name_len;
+		const char *trigger_name_src =
+			tuple_field_str_xc(new_tuple, BOX_TRIGGER_FIELD_NAME,
+					   &trigger_name_len);
+
+		const char *space_opts =
+			tuple_field_with_type_xc(new_tuple,
+						 BOX_TRIGGER_FIELD_OPTS,
+						 MP_MAP);
+		struct space_opts opts;
+		struct region *region = &fiber()->gc;
+		space_opts_decode(&opts, space_opts, region);
+		struct Trigger *new_trigger =
+			sql_trigger_compile(sql_get(), opts.sql);
+		if (new_trigger == NULL)
+			diag_raise();
+
+		auto new_trigger_guard = make_scoped_guard([=] {
+		    sql_trigger_delete(sql_get(), new_trigger);
+		});
+
+		const char *trigger_name = sql_trigger_name(new_trigger);
+		if (strlen(trigger_name) != trigger_name_len ||
+		    memcmp(trigger_name_src, trigger_name,
+			   trigger_name_len) != 0) {
+			tnt_raise(ClientError, ER_SQL,
+				  "trigger name does not match extracted "
+				  "from SQL");
+		}
+		uint32_t space_id =
+			tuple_field_u32_xc(new_tuple,
+					   BOX_TRIGGER_FIELD_SPACE_ID);
+		if (space_id != sql_trigger_space_id(new_trigger)) {
+			tnt_raise(ClientError, ER_SQL,
+				  "trigger space_id does not match the value "
+				  "resolved on AST building from SQL");
+		}
+
+		struct Trigger *old_trigger;
+		if (sql_trigger_replace(sql_get(), trigger_name, new_trigger,
+					&old_trigger) != 0)
+			diag_raise();
+
+		on_commit->data = old_trigger;
+		on_rollback->data = new_trigger;
+		new_trigger_guard.is_active = false;
+	}
+
+	txn_on_rollback(txn, on_rollback);
+	txn_on_commit(txn, on_commit);
 }
 
 struct trigger alter_space_on_replace_space = {

src/box/errcode.h

@@ -107,7 +107,7 @@ struct errcode_record {
 	/* 52 */_(ER_FUNCTION_EXISTS,		"Function '%s' already exists") \
 	/* 53 */_(ER_BEFORE_REPLACE_RET,	"Invalid return value of space:before_replace trigger: expected tuple or nil, got %s") \
 	/* 54 */_(ER_FUNCTION_MAX,		"A limit on the total number of functions has been reached: %u") \
-	/* 55 */_(ER_UNUSED4,			"") \
+	/* 55 */_(ER_TRIGGER_EXISTS,		"Trigger '%s' already exists") \
 	/* 56 */_(ER_USER_MAX,			"A limit on the total number of users has been reached: %u") \
 	/* 57 */_(ER_NO_SUCH_ENGINE,		"Space engine '%s' does not exist") \
 	/* 58 */_(ER_RELOAD_CFG,		"Can't set option '%s' dynamically") \

src/box/lua/schema.lua

@@ -502,6 +502,7 @@ box.schema.space.drop = function(space_id, space_name, opts)
     check_param_table(opts, { if_exists = 'boolean' })
     local _space = box.space[box.schema.SPACE_ID]
     local _index = box.space[box.schema.INDEX_ID]
+    local _trigger = box.space[box.schema.TRIGGER_ID]
     local _vindex = box.space[box.schema.VINDEX_ID]
     local _truncate = box.space[box.schema.TRUNCATE_ID]
     local _space_sequence = box.space[box.schema.SPACE_SEQUENCE_ID]
@@ -510,6 +511,9 @@ box.schema.space.drop = function(space_id, space_name, opts)
         -- Delete automatically generated sequence.
         box.schema.sequence.drop(sequence_tuple[2])
     end
+    for _, t in _trigger.index.space_id:pairs({space_id}) do
+        _trigger:delete({t.name})
+    end
     local keys = _vindex:select(space_id)
     for i = #keys, 1, -1 do
         local v = keys[i]

src/box/space.c

@@ -211,6 +211,11 @@ space_delete(struct space *space)
 	trigger_destroy(&space->on_replace);
 	trigger_destroy(&space->on_stmt_begin);
 	space_def_delete(space->def);
+	/*
+	 * SQL Triggers should be deleted with
+	 * on_replace_dd_trigger on deletion from _trigger.
+	 */
+	assert(space->sql_triggers == NULL);
 	space->vtab->destroy(space);
 }
 

src/box/space.h

@@ -146,6 +146,8 @@ struct space {
 	struct rlist on_replace;
 	/** Triggers fired before space statement */
 	struct rlist on_stmt_begin;
+	/** SQL Trigger list. */
+	struct Trigger *sql_triggers;
 	/**
 	 * The number of *enabled* indexes in the space.
 	 *

src/box/sql.c

@@ -1228,9 +1228,6 @@ space_foreach_put_cb(struct space *space, void *udata)
 /* Load database schema from Tarantool. */
 void tarantoolSqlite3LoadSchema(InitData *init)
 {
-	box_iterator_t *it;
-	box_tuple_t *tuple;
-
 	sql_schema_put(
 		init, TARANTOOL_SYS_SCHEMA_NAME,
 		BOX_SCHEMA_ID, 0,
@@ -1299,42 +1296,6 @@ void tarantoolSqlite3LoadSchema(InitData *init)
 		init->rc = SQL_TARANTOOL_ERROR;
 		return;
 	}
-
-	/* Read _trigger */
-	it = box_index_iterator(BOX_TRIGGER_ID, 0, ITER_GE,
-				nil_key, nil_key + sizeof(nil_key));
-
-	if (it == NULL) {
-		init->rc = SQL_TARANTOOL_ITERATOR_FAIL;
-		return;
-	}
-
-	while (box_iterator_next(it, &tuple) == 0 && tuple != NULL) {
-		const char *field, *ptr;
-		char *name, *sql;
-		unsigned len;
-		assert(tuple_field_count(tuple) == 3);
-
-		field = tuple_field(tuple, 0);
-		assert (field != NULL);
-		ptr = mp_decode_str(&field, &len);
-		name = strndup(ptr, len);
-
-		field = tuple_field(tuple, BOX_TRIGGER_FIELD_OPTS);
-		assert (field != NULL);
-		mp_decode_array(&field);
-		ptr = mp_decode_str(&field, &len);
-		assert (strncmp(ptr, "sql", 3) == 0);
-
-		ptr = mp_decode_str(&field, &len);
-		sql = strndup(ptr, len);
-
-		sql_schema_put(init, name, 0, 0, sql);
-
-		free(name);
-		free(sql);
-	}
-	box_iterator_free(it);
 }
 
 /*********************************************************************

src/box/sql.h

@@ -94,6 +94,17 @@ sql_expr_compile(struct sqlite3 *db, const char *expr, int expr_len);
 struct Select *
 sql_view_compile(struct sqlite3 *db, const char *view_stmt);
 
+/**
+ * Perform parsing of provided SQL request and construct trigger AST.
+ * @param db SQL context handle.
+ * @param sql request to parse.
+ *
+ * @retval NULL on error
+ * @retval not NULL Trigger AST pointer on success.
+ */
+struct Trigger *
+sql_trigger_compile(struct sqlite3 *db, const char *sql);
+
 /**
  * Free AST pointed by trigger.
  * @param db SQL handle.
@@ -102,6 +113,45 @@ sql_view_compile(struct sqlite3 *db, const char *view_stmt);
 void
 sql_trigger_delete(struct sqlite3 *db, struct Trigger *trigger);
 
+/**
+ * Get server triggers list by space_id.
+ * @param space_id valid Space ID.
+ *
+ * @retval trigger AST list.
+ */
+struct Trigger *
+space_trigger_list(uint32_t space_id);
+
+/**
+ * Perform replace trigger in SQL internals with new AST object.
+ * @param db SQL handle.
+ * @param name a name of the trigger.
+ * @param trigger AST object to insert.
+ * @param[out] old_trigger Old object if exists.
+ *
+ * @retval 0 on success.
+ * @retval -1 on error.
+ */
+int
+sql_trigger_replace(struct sqlite3 *db, const char *name,
+		    struct Trigger *trigger, struct Trigger **old_trigger);
+
+/**
+ * Get trigger name by trigger AST object.
+ * @param trigger AST object.
+ * @return trigger name string.
+ */
+const char *
+sql_trigger_name(struct Trigger *trigger);
+
+/**
+ * Get space_id of the space that trigger has been built for.
+ * @param trigger AST object.
+ * @return space identifier.
+ */
+uint32_t
+sql_trigger_space_id(struct Trigger *trigger);
+
 /**
  * Store duplicate of a parsed expression into @a parser.
  * @param parser Parser context.

src/box/sql/build.c

@@ -2112,16 +2112,14 @@ sql_code_drop_table(struct Parse *parse_context, struct space *space,
 	/*
 	 * Drop all triggers associated with the table being
 	 * dropped. Code is generated to remove entries from
-	 * _trigger. OP_DropTrigger will remove it from internal
-	 * SQL structures.
+	 * _trigger. on_replace_dd_trigger will remove it from
+	 * internal SQL structures.
 	 *
 	 * Do not account triggers deletion - they will be
 	 * accounted in DELETE from _space below.
 	 */
 	parse_context->nested++;
-	Table *table = sqlite3HashFind(&parse_context->db->pSchema->tblHash,
-				       space->def->name);
-	struct Trigger *trigger = table->pTrigger;
+	struct Trigger *trigger = space->sql_triggers;
 	while (trigger != NULL) {
 		sqlite3DropTriggerPtr(parse_context, trigger);
 		trigger = trigger->pNext;

src/box/sql/fkey.c

@@ -1429,8 +1429,6 @@ fkActionTrigger(Parse * pParse,	/* Parse context */
 			pStep->op = TK_UPDATE;
 		}
 		pStep->pTrig = pTrigger;
-		pTrigger->pSchema = pTab->pSchema;
-		pTrigger->pTabSchema = pTab->pSchema;
 		pFKey->apTrigger[iAction] = pTrigger;
 		pTrigger->op = (pChanges ? TK_UPDATE : TK_DELETE);
 	}

src/box/sql/insert.c

@@ -1748,9 +1748,9 @@ xferOptimization(Parse * pParse,	/* Parser context */
 		 */
 		return 0;
 	}
-	if (pDest->pTrigger) {
-		return 0;	/* tab1 must not have triggers */
-	}
+	/* The pDest must not have triggers. */
+	if (space_trigger_list(pDest->def->id) != NULL)
+		return 0;
 	if (onError == ON_CONFLICT_ACTION_DEFAULT) {
 		if (pDest->iPKey >= 0)
 			onError = pDest->keyConf;

src/box/sql/prepare.c

@@ -461,6 +461,9 @@ sql_parser_destroy(Parse *parser)
 	case AST_TYPE_EXPR:
 		sql_expr_delete(db, parser->parsed_ast.expr, false);
 		break;
+	case AST_TYPE_TRIGGER:
+		sql_trigger_delete(db, parser->parsed_ast.trigger);
+		break;
 	default:
 		assert(parser->parsed_ast_type == AST_TYPE_UNDEFINED);
 	}