From b44e19bbe8829b6dbfd796d5b091a95f08dbaabc Mon Sep 17 00:00:00 2001
From: Nikita Pettik <korablev@tarantool.org>
Date: Wed, 30 May 2018 17:18:24 +0300
Subject: [PATCH] sql: rework VIEW internals

This patch significantly reworks VIEW mechanisms.
Firstly, names resolution for VIEW now occurs during its creation.
In other words, we can't run following code, since name T1 doesn't exist
at the moment of V1 creation:

CREATE VIEW v1 AS SELECT * FROM t1;
CREATE TABLE t1(id PRIMARY KEY);

Now, table t1 must be created before view. As a result, no circularly
defined views are allowed. Moreover, it means that space representing
view is created with appropriate field names, types, collations etc.

Also, introduced view reference counter for each space. It is
incremented for each space participating in select statement.
For instace:

CREATE VIEW v1 AS SELECT * FROM (SELECT * FROM t1, t2);

In this case, view reference counter is increased for spaces T1 and T2.
Similarly, such counter is decremented for all dependent spaces when
VIEW is dropped.  To support such behavior, auxiliary
on_commit triggers are added. However, it is still not enough, since
before dropping space itself, we must drop secondary indexes, clear
_sequence space etc. Such changes can't be rollbacked (due to the lack
of transactional DDL), so before executing DROP routine, special opcode
OP_CheckViewReferences checks view reference counter for space to be
dropped.

Finally, we don't hold struct Select in struct Table or anywhere else
anymore.  Instead, 'CREATE VIEW AS SELECT ...' string is stored in
def->opts.sql.  At compilation time of query on view
(such as 'SELECT * FROM v1;') this string is parsed once and loaded
into struct Select, which in turn is used as before.

Fixed tests where view was created prior to referenced table.

Closes #3429, #3368, #3300
---

Branch: https://github.com/tarantool/tarantool/tree/np/reword-view-internals
Issues:
https://github.com/tarantool/tarantool/issues/3429
https://github.com/tarantool/tarantool/issues/3368
https://github.com/tarantool/tarantool/issues/3300

Notes for the one who will review (Vlad):
Firstly, I don't like code duplication in view on_commit/on_rollback
triggers. But I didn't manage to find any 'satisfactory' solution,
except for adding auxilary function with enum/bool arg which in turn
will determine action on view reference counter.
Secondly, (I guess) including in alter.cc sqliteInt.h header seems to be
total mess, but again - I failed to come up with kind of sinsible
resolution. Is it time for dividing SQL code base into source/header files?
At least, including "select.h" looks less annoying.
Finally, I temporary changed system space in box_space_id_by_name()
from _vspace to _space, in order to find space by name in on_dd_replace trigger.
I have noticed that Kirill Sh. has almost done the same thing, so this
hack is going to be substituted with his patch.

 src/box/alter.cc               |  84 ++++++++++++
 src/box/box.cc                 |   4 +-
 src/box/space_def.c            |   1 +
 src/box/space_def.h            |   2 +
 src/box/sql.h                  |  47 +++++++
 src/box/sql/build.c            | 289 ++++++++++++-----------------------------
 src/box/sql/delete.c           |   6 +-
 src/box/sql/expr.c             |   5 +-
 src/box/sql/fkey.c             |   4 +-
 src/box/sql/insert.c           |   4 +-
 src/box/sql/parse.y            |  19 +--
 src/box/sql/select.c           |  61 +++++++--
 src/box/sql/sqliteInt.h        |  31 ++++-
 src/box/sql/tokenize.c         |  18 +++
 src/box/sql/trigger.c          |   8 +-
 src/box/sql/vdbe.c             |  24 ++++
 test/sql-tap/colname.test.lua  |   4 +-
 test/sql-tap/drop_all.test.lua |   4 +-
 test/sql-tap/fkey2.test.lua    |   1 +
 test/sql-tap/trigger1.test.lua |   1 +
 test/sql-tap/triggerC.test.lua |   1 +
 test/sql-tap/view.test.lua     | 137 ++++++++++++++-----
 test/sql/view.result           |  23 +++-
 test/sql/view.test.lua         |  11 +-
 24 files changed, 502 insertions(+), 287 deletions(-)

diff --git a/src/box/alter.cc b/src/box/alter.cc
index b62f8adea..1fbb54d1a 100644
--- a/src/box/alter.cc
+++ b/src/box/alter.cc
@@ -29,6 +29,7 @@
  * SUCH DAMAGE.
  */
 #include "alter.h"
+#include "box.h"
 #include "schema.h"
 #include "user.h"
 #include "space.h"
@@ -44,6 +45,7 @@
 #include <new> /* for placement new */
 #include <stdio.h> /* snprintf() */
 #include <ctype.h>
+#include <box/sql/sqliteInt.h>
 #include "replication.h" /* for replica_set_id() */
 #include "session.h" /* to fetch the current user. */
 #include "vclock.h" /* VCLOCK_MAX */
@@ -1440,6 +1442,70 @@ alter_space_move_indexes(struct alter_space *alter, uint32_t begin,
 	}
 }
 
+/**
+ * Trigger which is fired on creation of new SQL view.
+ * Its purpose is to increment view reference counters of
+ * dependent spaces.
+ */
+static void
+on_create_view_commit(struct trigger *trigger, void *event)
+{
+	(void) event;
+	struct space *space = (struct space *)trigger->data;
+	struct space_def *def = space->def;
+	struct Select *select;
+	if (sql_view_compile(sql_get(), def->opts.sql, &select) != 0) {
+		diag_log();
+		return;
+	}
+	if (select == NULL)
+		return;
+	sql_select_expand_from_tables(select);
+	for (int i = 0; i < select->pSrc->nSrc; ++i) {
+		const char *space_name = select->pSrc->a[i].zName;
+		if (space_name == NULL)
+			continue;
+		uint32_t space_id = box_space_id_by_name(space_name,
+							 strlen(space_name));
+		assert(space_id != BOX_ID_NIL);
+		struct space *space = space_by_id(space_id);
+		space->def->view_ref_count++;
+	}
+	sql_select_delete(sql_get(), select);
+}
+
+/**
+ * Trigger which is fired on drop of SQL view.
+ * Its purpose is to decrement view reference counters of
+ * dependent spaces.
+ */
+static void
+on_drop_view_commit(struct trigger *trigger, void *event)
+{
+	(void) event;
+	struct space *space = (struct space *)trigger->data;
+	struct space_def *def = space->def;
+	struct Select *select;
+	if (sql_view_compile(sql_get(), def->opts.sql, &select) != 0) {
+		diag_log();
+		return;
+	}
+	if (select == NULL)
+		return;
+	sql_select_expand_from_tables(select);
+	for (int i = 0; i < select->pSrc->nSrc; ++i) {
+		const char *space_name = select->pSrc->a[i].zName;
+		if (space_name == NULL)
+			continue;
+		uint32_t space_id = box_space_id_by_name(space_name,
+							 strlen(space_name));
+		assert(space_id != BOX_ID_NIL);
+		struct space *space = space_by_id(space_id);
+		space->def->view_ref_count--;
+	}
+	sql_select_delete(sql_get(), select);
+}
+
 /**
  * A trigger which is invoked on replace in a data dictionary
  * space _space.
@@ -1545,6 +1611,12 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 		struct trigger *on_commit =
 			txn_alter_trigger_new(on_create_space_commit, space);
 		txn_on_commit(txn, on_commit);
+		if (def->opts.is_view && def->opts.sql != NULL) {
+			struct trigger *on_commit_view =
+				txn_alter_trigger_new(on_create_view_commit,
+						      space);
+			txn_on_commit(txn, on_commit_view);
+		}
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(on_create_space_rollback, space);
 		txn_on_rollback(txn, on_rollback);
@@ -1566,6 +1638,11 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 			tnt_raise(ClientError, ER_DROP_SPACE,
 				  space_name(old_space),
 				  "the space has truncate record");
+		if (old_space->def->view_ref_count > 0) {
+			tnt_raise(ClientError, ER_DROP_SPACE,
+				  space_name(old_space),
+				  "other views depend on this space");
+		}
 		/**
 		 * The space must be deleted from the space
 		 * cache right away to achieve linearisable
@@ -1575,6 +1652,13 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 		struct trigger *on_commit =
 			txn_alter_trigger_new(on_drop_space_commit, space);
 		txn_on_commit(txn, on_commit);
+		if (old_space->def->opts.is_view &&
+		    old_space->def->opts.sql != NULL) {
+			struct trigger *on_commit_view =
+				txn_alter_trigger_new(on_drop_view_commit,
+						      space);
+			txn_on_commit(txn, on_commit_view);
+		}
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(on_drop_space_rollback, space);
 		txn_on_rollback(txn, on_rollback);
diff --git a/src/box/box.cc b/src/box/box.cc
index 02722d6ad..d7e4108f5 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -892,7 +892,7 @@ box_space_id_by_name(const char *name, uint32_t len)
 
 	/* NOTE: error and missing key cases are indistinguishable */
 	box_tuple_t *tuple;
-	if (box_index_get(BOX_VSPACE_ID, 2, begin, end, &tuple) != 0)
+	if (box_index_get(BOX_SPACE_ID, 2, begin, end, &tuple) != 0)
 		return BOX_ID_NIL;
 	if (tuple == NULL)
 		return BOX_ID_NIL;
@@ -919,7 +919,7 @@ box_index_id_by_name(uint32_t space_id, const char *name, uint32_t len)
 
 	/* NOTE: error and missing key cases are indistinguishable */
 	box_tuple_t *tuple;
-	if (box_index_get(BOX_VINDEX_ID, 2, begin, end, &tuple) != 0)
+	if (box_index_get(BOX_INDEX_ID, 2, begin, end, &tuple) != 0)
 		return BOX_ID_NIL;
 	if (tuple == NULL)
 		return BOX_ID_NIL;
diff --git a/src/box/space_def.c b/src/box/space_def.c
index a39978182..21b7d82bb 100644
--- a/src/box/space_def.c
+++ b/src/box/space_def.c
@@ -204,6 +204,7 @@ space_def_new(uint32_t id, uint32_t uid, uint32_t exact_field_count,
 	memcpy(def->engine_name, engine_name, engine_len);
 	def->engine_name[engine_len] = 0;
 
+	def->view_ref_count = 0;
 	def->field_count = field_count;
 	if (field_count == 0) {
 		def->fields = NULL;
diff --git a/src/box/space_def.h b/src/box/space_def.h
index 024e868a9..4ff9c668e 100644
--- a/src/box/space_def.h
+++ b/src/box/space_def.h
@@ -107,6 +107,8 @@ struct space_def {
 	struct field_def *fields;
 	/** Length of @a fields. */
 	uint32_t field_count;
+	/** Number of SQL views which refer to this space. */
+	uint32_t view_ref_count;
 	struct space_opts opts;
 	char name[0];
 };
diff --git a/src/box/sql.h b/src/box/sql.h
index 23021e56b..9a006e5cb 100644
--- a/src/box/sql.h
+++ b/src/box/sql.h
@@ -83,6 +83,19 @@ int
 sql_expr_compile(struct sqlite3 *db, const char *expr, int expr_len,
 		 struct Expr **result);
 
+/**
+ * This routine executes parser on 'CREATE VIEW ...' statement
+ * and loads content of SELECT into internal structs as result.
+ *
+ * @param db Current SQL context.
+ * @param view_stmt String containing 'CREATE VIEW' statement.
+ * @param[out] select Fetched SELECT statement.
+ * @retval 0 on success, -1 otherwise.
+ */
+int
+sql_view_compile(struct sqlite3 *db, const char *view_stmt,
+		 struct Select **select);
+
 /**
  * Store duplicate of a parsed expression into @a parser.
  * @param parser Parser context.
@@ -293,6 +306,40 @@ sql_parser_create(struct Parse *parser, struct sqlite3 *db);
 void
 sql_parser_destroy(struct Parse *parser);
 
+/**
+ * Release memory allocated for given SELECT and all of its
+ * substructures. It accepts NULL pointers.
+ *
+ * @param db Database handler.
+ * @param select Select to be freed.
+ */
+void
+sql_select_delete(struct sqlite3 *db, struct Select *select);
+
+/**
+ * Work the same as sqlite3SrcListAppend(), but before adding to
+ * list provide check on name duplicates: only values with unique
+ * names are appended.
+ *
+ * @param db Database handler.
+ * @param list List of entries.
+ * @param new_name Name of entity to be added.
+ * @retval @list with new element on success, old one otherwise.
+ */
+struct SrcList *
+sql_src_list_append_unique(struct sqlite3 *db, struct SrcList *list,
+			   const char *new_name);
+
+/**
+ * Expand all spaces names from 'FROM' clause, including
+ * ones from subqueries, and add those names to the original
+ * select.
+ *
+ * @param select Select to be expanded.
+ */
+void
+sql_select_expand_from_tables(struct Select *select);
+
 #if defined(__cplusplus)
 } /* extern "C" { */
 #endif
diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index 28e4d7a4d..053cf83b7 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -405,7 +405,6 @@ deleteTable(sqlite3 * db, Table * pTable)
 	sqlite3HashClear(&pTable->idxHash);
 	sqlite3DbFree(db, pTable->aCol);
 	sqlite3DbFree(db, pTable->zColAff);
-	sqlite3SelectDelete(db, pTable->pSelect);
 	assert(pTable->def != NULL);
 	/* Do not delete pTable->def allocated on region. */
 	if (!pTable->def->opts.temporary)
@@ -1849,7 +1848,6 @@ sqlite3EndTable(Parse * pParse,	/* Parse context */
 		p->def->id = SQLITE_PAGENO_TO_SPACEID(p->tnum);
 	}
 
-	assert(p->def->opts.is_view == (p->pSelect != NULL));
 	if (!p->def->opts.is_view) {
 		if ((p->tabFlags & TF_HasPrimaryKey) == 0) {
 			sqlite3ErrorMsg(pParse,
@@ -2009,72 +2007,74 @@ sqlite3EndTable(Parse * pParse,	/* Parse context */
 }
 
 #ifndef SQLITE_OMIT_VIEW
-/*
- * The parser calls this routine in order to create a new VIEW
- */
 void
-sqlite3CreateView(Parse * pParse,	/* The parsing context */
-		  Token * pBegin,	/* The CREATE token that begins the statement */
-		  Token * pName,	/* The token that holds the name of the view */
-		  ExprList * pCNames,	/* Optional list of view column names */
-		  Select * pSelect,	/* A SELECT statement that will become the new view */
-		  int noErr	/* Suppress error messages if VIEW already exists */
-    )
+sql_create_view(struct Parse *parse_context, struct Token *begin,
+		struct Token *name, struct ExprList *aliases,
+		struct Select *select, bool if_exists)
 {
-	Table *p;
-	int n;
-	const char *z;
-	Token sEnd;
-	DbFixer sFix;
-	sqlite3 *db = pParse->db;
-
-	if (pParse->nVar > 0) {
-		sqlite3ErrorMsg(pParse, "parameters are not allowed in views");
+	struct sqlite3 *db = parse_context->db;
+	if (parse_context->nVar > 0) {
+		sqlite3ErrorMsg(parse_context,
+				"parameters are not allowed in views");
 		goto create_view_fail;
 	}
-	sqlite3StartTable(pParse, pName, noErr);
-	p = pParse->pNewTable;
-	if (p == 0 || pParse->nErr)
+	sqlite3StartTable(parse_context, name, if_exists);
+	struct Table *p = parse_context->pNewTable;
+	if (p == NULL || parse_context->nErr)
 		goto create_view_fail;
-	sqlite3FixInit(&sFix, pParse, "view", pName);
-	if (sqlite3FixSelect(&sFix, pSelect))
+	struct Table *sel_tab =	sqlite3ResultSetOfSelect(parse_context, select);
+	if (sel_tab == NULL)
 		goto create_view_fail;
-
-	/* Make a copy of the entire SELECT statement that defines the view.
-	 * This will force all the Expr.token.z values to be dynamically
-	 * allocated rather than point to the input string - which means that
-	 * they will persist after the current sqlite3_exec() call returns.
-	 */
-	p->pSelect = sqlite3SelectDup(db, pSelect, EXPRDUP_REDUCE);
+	if (aliases != NULL) {
+		if ((int)sel_tab->def->field_count != aliases->nExpr) {
+			sqlite3ErrorMsg(parse_context,
+					"expected %d columns for '%s' but got %d",
+					aliases->nExpr, p->def->name,
+					sel_tab->def->field_count);
+			goto create_view_fail;
+		}
+		sqlite3ColumnsFromExprList(parse_context, aliases, p);
+		sqlite3SelectAddColumnTypeAndCollation(parse_context, p,
+						       select);
+	} else {
+		assert(p->aCol == NULL);
+		assert(sel_tab->def->opts.temporary);
+		p->def->fields = sel_tab->def->fields;
+		p->def->field_count = sel_tab->def->field_count;
+		p->aCol = sel_tab->aCol;
+		sel_tab->aCol = NULL;
+		sel_tab->def->fields = NULL;
+		sel_tab->def->field_count = 0;
+	}
 	p->def->opts.is_view = true;
-	p->def->opts.checks = sql_expr_list_dup(db, pCNames, EXPRDUP_REDUCE);
-	if (db->mallocFailed)
-		goto create_view_fail;
-
-	/* Locate the end of the CREATE VIEW statement.  Make sEnd point to
-	 * the end.
+	/*
+	 * Locate the end of the CREATE VIEW statement.
+	 * Make sEnd point to the end.
 	 */
-	sEnd = pParse->sLastToken;
-	assert(sEnd.z[0] != 0);
-	if (sEnd.z[0] != ';') {
-		sEnd.z += sEnd.n;
+	struct Token end = parse_context->sLastToken;
+	assert(end.z[0] != 0);
+	if (end.z[0] != ';') {
+		end.z += end.n;
 	}
-	sEnd.n = 0;
-	n = (int)(sEnd.z - pBegin->z);
+	end.n = 0;
+	int n = (int)(end.z - begin->z);
 	assert(n > 0);
-	z = pBegin->z;
+	const char *z = begin->z;
 	while (sqlite3Isspace(z[n - 1])) {
 		n--;
 	}
-	sEnd.z = &z[n - 1];
-	sEnd.n = 1;
+	end.z = &z[n - 1];
+	end.n = 1;
+	p->def->opts.sql = malloc(n + 1);
+	memcpy(p->def->opts.sql, begin->z, n);
+	p->def->opts.sql[n] = '\0';
 
 	/* Use sqlite3EndTable() to add the view to the Tarantool.  */
-	sqlite3EndTable(pParse, 0, &sEnd, 0);
+	sqlite3EndTable(parse_context, 0, &end, 0);
 
  create_view_fail:
-	sqlite3SelectDelete(db, pSelect);
-	sql_expr_list_delete(db, pCNames);
+	sql_expr_list_delete(db, aliases);
+	sql_select_delete(db, select);
 	return;
 }
 #endif				/* SQLITE_OMIT_VIEW */
@@ -2084,155 +2084,22 @@ sql_view_column_names(struct Parse *parse, struct Table *table)
 {
 	assert(table != NULL);
 	assert(space_is_view(table));
-	/* A positive nCol means the columns names for this view
-	 * are already known.
-	 */
-	if (table->def->field_count > 0)
-		return 0;
-
-	/* A negative nCol is a special marker meaning that we are
-	 * currently trying to compute the column names.  If we
-	 * enter this routine with a negative nCol, it means two
-	 * or more views form a loop, like this:
-	 *
-	 *     CREATE VIEW one AS SELECT * FROM two;
-	 *     CREATE VIEW two AS SELECT * FROM one;
-	 *
-	 * Actually, the error above is now caught prior to
-	 * reaching this point. But the following test is still
-	 * important as it does come up in the following:
-	 *
-	 *     CREATE TABLE main.ex1(a);
-	 *     CREATE TEMP VIEW ex1 AS SELECT a FROM ex1;
-	 *     SELECT * FROM temp.ex1;
-	 */
-	if ((int)table->def->field_count < 0) {
-		sqlite3ErrorMsg(parse, "view %s is circularly defined",
-				table->def->name);
+	struct sqlite3 *db = parse->db;
+	struct Select *select;
+	if (sql_view_compile(db, table->def->opts.sql, &select) != 0) {
+		diag_log();
 		return -1;
 	}
-
-	/* If we get this far, it means we need to compute the
-	 * table names. Note that the call to
-	 * sqlite3ResultSetOfSelect() will expand any "*" elements
-	 * in the results set of the view and will assign cursors
-	 * to the elements of the FROM clause.  But we do not want
-	 * these changes to be permanent.  So the computation is
-	 * done on a copy of the SELECT statement that defines the
-	 * view.
-	 */
-	assert(table->pSelect != NULL);
-	sqlite3 *db = parse->db;
-	struct Select *select = sqlite3SelectDup(db, table->pSelect, 0);
-	if (select == NULL)
-		return -1;
-	int n = parse->nTab;
 	sqlite3SrcListAssignCursors(parse, select->pSrc);
-	table->def->field_count = -1;
-	db->lookaside.bDisable++;
-	struct Table *sel_tab = sqlite3ResultSetOfSelect(parse, select);
-	parse->nTab = n;
-	int rc = 0;
-	/* Get server checks. */
-	ExprList *checks = space_checks_expr_list(table->def->id);
-	if (checks != NULL) {
-		/* CREATE VIEW name(arglist) AS ...
-		 * The names of the columns in the table are taken
-		 * from arglist which is stored in table->pCheck.
-		 * The pCheck field normally holds CHECK
-		 * constraints on an ordinary table, but for a
-		 * VIEW it holds the list of column names.
-		 */
-		struct space_def *old_def = table->def;
-		sqlite3ColumnsFromExprList(parse, checks, table);
-		if (sql_table_def_rebuild(db, table) != 0) {
-			rc = -1;
-		} else {
-			space_def_delete(old_def);
-		}
-		if (db->mallocFailed == 0 && parse->nErr == 0
-		    && (int)table->def->field_count == select->pEList->nExpr) {
-			sqlite3SelectAddColumnTypeAndCollation(parse, table,
-							       select);
-		}
-	} else if (sel_tab != NULL) {
-		/* CREATE VIEW name AS...  without an argument
-		 * list.  Construct the column names from the
-		 * SELECT statement that defines the view.
-		 */
-		assert(table->aCol == NULL);
-		assert(sel_tab->def->opts.temporary);
-		struct space_def *old_def = table->def;
-		struct space_def *new_def =
-			sql_ephemeral_space_def_new(parse,
-						    old_def->name);
-		if (new_def == NULL) {
-			rc = -1;
-		} else {
-			memcpy(new_def, old_def,
-			       sizeof(struct space_def));
-			new_def->dict = NULL;
-			new_def->opts.temporary = true;
-			new_def->fields = sel_tab->def->fields;
-			new_def->field_count =
-				sel_tab->def->field_count;
-			table->def = new_def;
-			if (sql_table_def_rebuild(db, table) != 0)
-				rc = -1;
-		}
-		table->aCol = sel_tab->aCol;
-		sel_tab->aCol = NULL;
-		sel_tab->def = old_def;
-		sel_tab->def->fields = NULL;
-		sel_tab->def->field_count = 0;
-	} else {
-		table->def->field_count = 0;
-		rc = -1;
-	}
-	sqlite3DeleteTable(db, sel_tab);
-	sqlite3SelectDelete(db, select);
-	db->lookaside.bDisable--;
-
-	return rc;
+	return 0;
 }
 
-#ifndef SQLITE_OMIT_VIEW
-/*
- * Clear the column names from every VIEW in database idx.
- */
-static void
-sqliteViewResetAll(sqlite3 * db)
+void
+sql_store_select(struct Parse *parse_context, struct Select *select)
 {
-	HashElem *i;
-	for (i = sqliteHashFirst(&db->pSchema->tblHash); i;
-	     i = sqliteHashNext(i)) {
-		Table *pTab = sqliteHashData(i);
-		assert(pTab->def->opts.is_view == (pTab->pSelect != NULL));
-		if (pTab->def->opts.is_view) {
-			sqlite3DbFree(db, pTab->aCol);
-			struct space_def *old_def = pTab->def;
-			assert(old_def->opts.temporary == false);
-			pTab->def = space_def_new(old_def->id, old_def->uid,
-						  0, old_def->name,
-						  strlen(old_def->name),
-						  old_def->engine_name,
-						  strlen(old_def->engine_name),
-						  &old_def->opts,
-						  NULL, 0);
-			if (pTab->def == NULL) {
-				sqlite3OomFault(db);
-				pTab->def = old_def;
-			} else {
-				space_def_delete(old_def);
-			}
-			pTab->aCol = NULL;
-			pTab->def->field_count = 0;
-		}
-	}
+	Select *select_copy = sqlite3SelectDup(parse_context->db, select, 0);
+	parse_context->parsed_select = select_copy;
 }
-#else
-#define sqliteViewResetAll(A,B)
-#endif				/* SQLITE_OMIT_VIEW */
 
 /**
  * Remove entries from the _sql_stat1 and _sql_stat4
@@ -2283,7 +2150,6 @@ static void
 sql_code_drop_table(struct Parse *parse_context, struct space *space,
 		    bool is_view)
 {
-	struct sqlite3 *db = parse_context->db;
 	struct Vdbe *v = sqlite3GetVdbe(parse_context);
 	assert(v != NULL);
 	/*
@@ -2313,6 +2179,7 @@ sql_code_drop_table(struct Parse *parse_context, struct space *space,
 	int space_id_reg = ++parse_context->nMem;
 	int space_id = space->def->id;
 	sqlite3VdbeAddOp2(v, OP_Integer, space_id, space_id_reg);
+	sqlite3VdbeAddOp1(v, OP_CheckViewReferences, space_id_reg);
 	if (space->sequence != NULL) {
 		/* Delete entry from _space_sequence. */
 		sqlite3VdbeAddOp3(v, OP_MakeRecord, space_id_reg, 1,
@@ -2371,14 +2238,6 @@ sql_code_drop_table(struct Parse *parse_context, struct space *space,
 	VdbeComment((v, "Delete entry from _space"));
 	/* Remove the table entry from SQLite's internal schema. */
 	sqlite3VdbeAddOp4(v, OP_DropTable, 0, 0, 0, space->def->name, 0);
-
-	/*
-	 * Replace to _space/_index will fail if active
-	 * transaction. So, don't pretend, that we are able to
-	 * anything back. To be fixed when transactions
-	 * DDL are enabled.
-	 */
-	sqliteViewResetAll(db);
 }
 
 /**
@@ -2914,7 +2773,7 @@ sql_create_index(struct Parse *parse, struct Token *token,
 	assert(pTab != 0);
 	assert(parse->nErr == 0);
 #ifndef SQLITE_OMIT_VIEW
-	if (pTab->pSelect) {
+	if (pTab->def->opts.is_view) {
 		sqlite3ErrorMsg(parse, "views may not be indexed");
 		goto exit_create_index;
 	}
@@ -3639,6 +3498,22 @@ sqlite3SrcListAppend(sqlite3 * db,	/* Connection to notify of malloc failures */
 	return pList;
 }
 
+struct SrcList *
+sql_src_list_append_unique(struct sqlite3 *db, struct SrcList *list,
+			   const char *new_name)
+{
+	assert(list != NULL);
+	assert(new_name != NULL);
+
+	for (int i = 0; i < list->nSrc; ++i) {
+		const char *name = list->a[i].zName;
+		if (name != NULL && strcmp(new_name, name) == 0)
+			return list;
+	}
+	struct Token token = { new_name, strlen(new_name), 0 };
+	return sqlite3SrcListAppend(db, list, &token);
+}
+
 /*
  * Assign VdbeCursor index numbers to all tables in a SrcList
  */
@@ -3680,7 +3555,7 @@ sqlite3SrcListDelete(sqlite3 * db, SrcList * pList)
 		if (pItem->fg.isTabFunc)
 			sql_expr_list_delete(db, pItem->u1.pFuncArg);
 		sqlite3DeleteTable(db, pItem->pTab);
-		sqlite3SelectDelete(db, pItem->pSelect);
+		sql_select_delete(db, pItem->pSelect);
 		sql_expr_delete(db, pItem->pOn, false);
 		sqlite3IdListDelete(db, pItem->pUsing);
 	}
@@ -3739,7 +3614,7 @@ sqlite3SrcListAppendFromTerm(Parse * pParse,	/* Parsing context */
 	assert(p == 0);
 	sql_expr_delete(db, pOn, false);
 	sqlite3IdListDelete(db, pUsing);
-	sqlite3SelectDelete(db, pSubquery);
+	sql_select_delete(db, pSubquery);
 	return 0;
 }
 
@@ -4141,7 +4016,7 @@ sqlite3WithAdd(Parse * pParse,	/* Parsing context */
 
 	if (db->mallocFailed) {
 		sql_expr_list_delete(db, pArglist);
-		sqlite3SelectDelete(db, pQuery);
+		sql_select_delete(db, pQuery);
 		sqlite3DbFree(db, zName);
 		pNew = pWith;
 	} else {
@@ -4166,7 +4041,7 @@ sqlite3WithDelete(sqlite3 * db, With * pWith)
 		for (i = 0; i < pWith->nCte; i++) {
 			struct Cte *pCte = &pWith->a[i];
 			sql_expr_list_delete(db, pCte->pCols);
-			sqlite3SelectDelete(db, pCte->pSelect);
+			sql_select_delete(db, pCte->pSelect);
 			sqlite3DbFree(db, pCte->zName);
 		}
 		sqlite3DbFree(db, pWith);
diff --git a/src/box/sql/delete.c b/src/box/sql/delete.c
index ddad54b3e..ab371cdb1 100644
--- a/src/box/sql/delete.c
+++ b/src/box/sql/delete.c
@@ -68,7 +68,7 @@ sql_materialize_view(struct Parse *parse, const char *name, struct Expr *where,
 	struct SelectDest dest;
 	sqlite3SelectDestInit(&dest, SRT_EphemTab, cursor);
 	sqlite3Select(parse, select, &dest);
-	sqlite3SelectDelete(db, select);
+	sql_select_delete(db, select);
 }
 
 void
@@ -339,7 +339,7 @@ sql_table_delete_from(struct Parse *parse, struct SrcList *tab_list,
 		if (one_pass != ONEPASS_OFF) {
 			/* OP_Found will use an unpacked key. */
 			assert(key_len == pk_len);
-			assert(pk != NULL || table->pSelect != NULL);
+			assert(pk != NULL || table->def->opts.is_view);
 			sqlite3VdbeAddOp4Int(v, OP_NotFound, tab_cursor,
 					     addr_bypass, reg_key, key_len);
 
@@ -483,7 +483,7 @@ sql_generate_row_delete(struct Parse *parse, struct Table *table,
 	 * of the DELETE statement is to fire the INSTEAD OF
 	 * triggers).
 	 */
-	if (table->pSelect == NULL) {
+	if (!table->def->opts.is_view) {
 		uint8_t p5 = 0;
 		sqlite3VdbeAddOp2(v, OP_Delete, cursor,
 				  (need_update_count ? OPFLAG_NCHANGE : 0));
diff --git a/src/box/sql/expr.c b/src/box/sql/expr.c
index 8866f6fed..a0475fbf3 100644
--- a/src/box/sql/expr.c
+++ b/src/box/sql/expr.c
@@ -953,7 +953,7 @@ sqlite3PExprAddSelect(Parse * pParse, Expr * pExpr, Select * pSelect)
 		sqlite3ExprSetHeightAndFlags(pParse, pExpr);
 	} else {
 		assert(pParse->db->mallocFailed);
-		sqlite3SelectDelete(pParse->db, pSelect);
+		sql_select_delete(pParse->db, pSelect);
 	}
 }
 
@@ -1152,7 +1152,7 @@ sqlite3ExprDeleteNN(sqlite3 * db, Expr * p, bool extern_alloc)
 		if (!extern_alloc)
 			sql_expr_delete(db, p->pRight, extern_alloc);
 		if (ExprHasProperty(p, EP_xIsSelect)) {
-			sqlite3SelectDelete(db, p->x.pSelect);
+			sql_select_delete(db, p->x.pSelect);
 		} else {
 			sql_expr_list_delete(db, p->x.pList);
 		}
@@ -2191,7 +2191,6 @@ isCandidateForInOpt(Expr * pX)
 		return 0;	/* FROM is not a subquery or view */
 	pTab = pSrc->a[0].pTab;
 	assert(pTab != 0);
-	assert(pTab->def->opts.is_view == (pTab->pSelect != NULL));
 	/* FROM clause is not a view */
 	assert(!pTab->def->opts.is_view);
 	pEList = p->pEList;
diff --git a/src/box/sql/fkey.c b/src/box/sql/fkey.c
index 70ebef89f..2644a26f9 100644
--- a/src/box/sql/fkey.c
+++ b/src/box/sql/fkey.c
@@ -744,7 +744,7 @@ fkTriggerDelete(sqlite3 * dbMem, Trigger * p)
 		TriggerStep *pStep = p->step_list;
 		sql_expr_delete(dbMem, pStep->pWhere, false);
 		sql_expr_list_delete(dbMem, pStep->pExprList);
-		sqlite3SelectDelete(dbMem, pStep->pSelect);
+		sql_select_delete(dbMem, pStep->pSelect);
 		sql_expr_delete(dbMem, p->pWhen, false);
 		sqlite3DbFree(dbMem, p);
 	}
@@ -1418,7 +1418,7 @@ fkActionTrigger(Parse * pParse,	/* Parse context */
 		sql_expr_delete(db, pWhere, false);
 		sql_expr_delete(db, pWhen, false);
 		sql_expr_list_delete(db, pList);
-		sqlite3SelectDelete(db, pSelect);
+		sql_select_delete(db, pSelect);
 		if (db->mallocFailed == 1) {
 			fkTriggerDelete(db, pTrigger);
 			return 0;
diff --git a/src/box/sql/insert.c b/src/box/sql/insert.c
index 59c61c703..afde43a64 100644
--- a/src/box/sql/insert.c
+++ b/src/box/sql/insert.c
@@ -371,7 +371,7 @@ sqlite3Insert(Parse * pParse,	/* Parser context */
 	    && pSelect->pPrior == 0) {
 		pList = pSelect->pEList;
 		pSelect->pEList = 0;
-		sqlite3SelectDelete(db, pSelect);
+		sql_select_delete(db, pSelect);
 		pSelect = 0;
 	}
 
@@ -920,7 +920,7 @@ sqlite3Insert(Parse * pParse,	/* Parser context */
  insert_cleanup:
 	sqlite3SrcListDelete(db, pTabList);
 	sql_expr_list_delete(db, pList);
-	sqlite3SelectDelete(db, pSelect);
+	sql_select_delete(db, pSelect);
 	sqlite3IdListDelete(db, pColumn);
 	sqlite3DbFree(db, aRegIdx);
 }
diff --git a/src/box/sql/parse.y b/src/box/sql/parse.y
index 2e5f349c8..fc5d2f0a6 100644
--- a/src/box/sql/parse.y
+++ b/src/box/sql/parse.y
@@ -184,7 +184,7 @@ create_table_args ::= LP columnlist conslist_opt(X) RP(E). {
 }
 create_table_args ::= AS select(S). {
   sqlite3EndTable(pParse,0,0,S);
-  sqlite3SelectDelete(pParse->db, S);
+  sql_select_delete(pParse->db, S);
 }
 columnlist ::= columnlist COMMA columnname carglist.
 columnlist ::= columnname carglist.
@@ -373,7 +373,10 @@ ifexists(A) ::= .            {A = 0;}
 %ifndef SQLITE_OMIT_VIEW
 cmd ::= createkw(X) VIEW ifnotexists(E) nm(Y) eidlist_opt(C)
           AS select(S). {
-  sqlite3CreateView(pParse, &X, &Y, C, S, E);
+  if (!pParse->parse_only)
+    sql_create_view(pParse, &X, &Y, C, S, E);
+  else
+    sql_store_select(pParse, S);
 }
 cmd ::= DROP VIEW ifexists(E) fullname(X). {
   sql_drop_table(pParse, X, 1, E);
@@ -388,15 +391,15 @@ cmd ::= select(X).  {
 	  sqlite3Select(pParse, X, &dest);
   else
 	  sql_expr_extract_select(pParse, X);
-  sqlite3SelectDelete(pParse->db, X);
+  sql_select_delete(pParse->db, X);
 }
 
 %type select {Select*}
-%destructor select {sqlite3SelectDelete(pParse->db, $$);}
+%destructor select {sql_select_delete(pParse->db, $$);}
 %type selectnowith {Select*}
-%destructor selectnowith {sqlite3SelectDelete(pParse->db, $$);}
+%destructor selectnowith {sql_select_delete(pParse->db, $$);}
 %type oneselect {Select*}
-%destructor oneselect {sqlite3SelectDelete(pParse->db, $$);}
+%destructor oneselect {sql_select_delete(pParse->db, $$);}
 
 %include {
   /*
@@ -453,7 +456,7 @@ selectnowith(A) ::= selectnowith(A) multiselect_op(Y) oneselect(Z).  {
     pRhs->selFlags &= ~SF_MultiValue;
     if( Y!=TK_ALL ) pParse->hasCompound = 1;
   }else{
-    sqlite3SelectDelete(pParse->db, pLhs);
+    sql_select_delete(pParse->db, pLhs);
   }
   A = pRhs;
 }
@@ -496,7 +499,7 @@ oneselect(A) ::= SELECT(S) distinct(D) selcollist(W) from(X) where_opt(Y)
 oneselect(A) ::= values(A).
 
 %type values {Select*}
-%destructor values {sqlite3SelectDelete(pParse->db, $$);}
+%destructor values {sql_select_delete(pParse->db, $$);}
 values(A) ::= VALUES LP nexprlist(X) RP. {
   A = sqlite3SelectNew(pParse,X,0,0,0,0,0,SF_Values,0,0);
 }
diff --git a/src/box/sql/select.c b/src/box/sql/select.c
index 2aa35a114..0b64aee74 100644
--- a/src/box/sql/select.c
+++ b/src/box/sql/select.c
@@ -207,11 +207,8 @@ sqlite3SelectSetName(Select * p, const char *zName)
 }
 #endif
 
-/*
- * Delete the given Select structure and all of its substructures.
- */
 void
-sqlite3SelectDelete(sqlite3 * db, Select * p)
+sql_select_delete(sqlite3 *db, Select *p)
 {
 	if (p)
 		clearSelect(db, p, 1);
@@ -228,6 +225,43 @@ findRightmost(Select * p)
 	return p;
 }
 
+/**
+ * This function is an inner call of recursive traverse through
+ * select AST starting from interface function
+ * sql_select_expand_from_tables().
+ *
+ * @param top_select The root of AST.
+ * @param sub_select sub-select of current level recursion.
+ */
+static void
+expand_names_sub_select(struct Select *top_select,
+			struct Select *sub_select)
+{
+	assert(top_select != NULL);
+	assert(sub_select != NULL);
+	for (int i = 0; i < sub_select->pSrc->nSrc; ++i) {
+		if (sub_select->pSrc->a[i].zName == NULL) {
+			expand_names_sub_select(top_select,
+						sub_select->pSrc->a[i].pSelect);
+		} else {
+			top_select->pSrc =
+				sql_src_list_append_unique(sql_get(),
+							   top_select->pSrc,
+							   sub_select->pSrc->a[i].zName);
+		}
+	}
+}
+
+void
+sql_select_expand_from_tables(struct Select *select)
+{
+	assert(select != NULL);
+	for (int i = 0; i < select->pSrc->nSrc; ++i)
+		if (select->pSrc->a[i].zName == NULL)
+			expand_names_sub_select(select,
+						select->pSrc->a[i].pSelect);
+}
+
 /*
  * Given 1 to 3 identifiers preceding the JOIN keyword, determine the
  * type of join.  Return an integer constant that expresses that type
@@ -2830,7 +2864,7 @@ multiSelect(Parse * pParse,	/* Parsing context */
  multi_select_end:
 	pDest->iSdst = dest.iSdst;
 	pDest->nSdst = dest.nSdst;
-	sqlite3SelectDelete(db, pDelete);
+	sql_select_delete(db, pDelete);
 	return rc;
 }
 #endif				/* SQLITE_OMIT_COMPOUND_SELECT */
@@ -3432,7 +3466,7 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 	 * by the calling function
 	 */
 	if (p->pPrior) {
-		sqlite3SelectDelete(db, p->pPrior);
+		sql_select_delete(db, p->pPrior);
 	}
 	p->pPrior = pPrior;
 	pPrior->pNext = p;
@@ -4106,7 +4140,7 @@ flattenSubquery(Parse * pParse,		/* Parsing context */
 	/* Finially, delete what is left of the subquery and return
 	 * success.
 	 */
-	sqlite3SelectDelete(db, pSub1);
+	sql_select_delete(db, pSub1);
 
 #ifdef SELECTTRACE_ENABLED
 	if (sqlite3SelectTrace & 0x100) {
@@ -4738,13 +4772,16 @@ selectExpander(Walker * pWalker, Select * p)
 			}
 #if !defined(SQLITE_OMIT_VIEW)
 			if (space_is_view(pTab)) {
-				if (sql_view_column_names(pParse, pTab) != 0)
+				struct Select *select;
+				if (sql_view_compile(db, pTab->def->opts.sql,
+						     &select) != 0) {
+					diag_log();
 					return WRC_Abort;
+				}
+				sqlite3SrcListAssignCursors(pParse,
+							    select->pSrc);
 				assert(pFrom->pSelect == 0);
-				assert(pTab->def->opts.is_view ==
-				       (pTab->pSelect != NULL));
-				pFrom->pSelect =
-				    sqlite3SelectDup(db, pTab->pSelect, 0);
+				pFrom->pSelect = select;
 				sqlite3SelectSetName(pFrom->pSelect,
 						     pTab->def->name);
 				int columns = pTab->def->field_count;
diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
index 01351a183..781f5d0a0 100644
--- a/src/box/sql/sqliteInt.h
+++ b/src/box/sql/sqliteInt.h
@@ -1913,7 +1913,6 @@ struct Column {
 struct Table {
 	Column *aCol;		/* Information about each column */
 	Index *pIndex;		/* List of SQL indexes on this table. */
-	Select *pSelect;	/* NULL for tables.  Points to definition if a view. */
 	FKey *pFKey;		/* Linked list of all foreign keys in this table */
 	char *zColAff;		/* String defining the affinity of each column */
 	/*   ... also used as column name list in a VIEW */
@@ -2688,9 +2687,7 @@ struct Select {
 	LogEst nSelectRow;	/* Estimated number of result rows */
 	u32 selFlags;		/* Various SF_* values */
 	int iLimit, iOffset;	/* Memory registers holding LIMIT & OFFSET counters */
-#ifdef SELECTTRACE_ENABLED
 	char zSelName[12];	/* Symbolic name of this SELECT use for debugging */
-#endif
 	int addrOpenEphm[2];	/* OP_OpenEphem opcodes related to this select */
 	SrcList *pSrc;		/* The FROM clause */
 	Expr *pWhere;		/* The WHERE clause */
@@ -2962,6 +2959,8 @@ struct Parse {
 	bool parse_only;
 	/** If parse_only is set to true, store parsed expression. */
 	struct Expr *parsed_expr;
+	/** If parse_only is set to true, store parsed SELECT. */
+	struct Select *parsed_select;
 };
 
 /*
@@ -3570,7 +3569,20 @@ int sqlite3ParseUri(const char *, const char *, unsigned int *,
 int sqlite3FaultSim(int);
 #endif
 
-void sqlite3CreateView(Parse *, Token *, Token *, ExprList *, Select *, int);
+/**
+ * The parser calls this routine in order to create a new VIEW.
+ *
+ * @param parse_context Current parsing context.
+ * @param begin The CREATE token that begins the statement.
+ * @param name The token that holds the name of the view.
+ * @param aliases Optional list of view column names.
+ * @param select A SELECT statement that will become the new view.
+ * @param if_exists Suppress error messages if VIEW already exists.
+ */
+void
+sql_create_view(struct Parse *parse_context, struct Token *begin,
+		struct Token *name, struct ExprList *aliases,
+		struct Select *select, bool is_exists);
 
 /**
  * The Table structure pTable is really a VIEW.  Fill in the names
@@ -3585,6 +3597,16 @@ void sqlite3CreateView(Parse *, Token *, Token *, ExprList *, Select *, int);
 int
 sql_view_column_names(struct Parse *parse, struct Table *table);
 
+/**
+ * Store duplicate of SELECT into parsing context.
+ * This routine is called during parsing.
+ *
+ * @param parse_context Current parsing context.
+ * @param select Select to be stored.
+ */
+void
+sql_store_select(struct Parse *parse_context, struct Select *select);
+
 void
 sql_drop_table(struct Parse *, struct SrcList *, bool, bool);
 void sqlite3DeleteTable(sqlite3 *, Table *);
@@ -3656,7 +3678,6 @@ sql_drop_index(struct Parse *, struct SrcList *, struct Token *, bool);
 int sqlite3Select(Parse *, Select *, SelectDest *);
 Select *sqlite3SelectNew(Parse *, ExprList *, SrcList *, Expr *, ExprList *,
 			 Expr *, ExprList *, u32, Expr *, Expr *);
-void sqlite3SelectDelete(sqlite3 *, Select *);
 
 /**
  * While a SrcList can in general represent multiple tables and
diff --git a/src/box/sql/tokenize.c b/src/box/sql/tokenize.c
index 42c70a255..25d8dc59b 100644
--- a/src/box/sql/tokenize.c
+++ b/src/box/sql/tokenize.c
@@ -566,3 +566,21 @@ sql_expr_compile(sqlite3 *db, const char *expr, int expr_len,
 	sql_parser_destroy(&parser);
 	return 0;
 }
+
+int
+sql_view_compile(struct sqlite3 *db, const char *view_stmt,
+		 struct Select **select)
+{
+	assert(select != NULL);
+	struct Parse parser;
+	sql_parser_create(&parser, db);
+	parser.parse_only = true;
+	char *unused;
+	if (sqlite3RunParser(&parser, view_stmt, &unused) != SQLITE_OK) {
+		diag_set(ClientError, ER_SQL_EXECUTE, view_stmt);
+		return -1;
+	}
+	*select = parser.parsed_select;
+	sql_parser_destroy(&parser);
+	return 0;
+}
diff --git a/src/box/sql/trigger.c b/src/box/sql/trigger.c
index ea3521133..32f40044f 100644
--- a/src/box/sql/trigger.c
+++ b/src/box/sql/trigger.c
@@ -54,7 +54,7 @@ sqlite3DeleteTriggerStep(sqlite3 * db, TriggerStep * pTriggerStep)
 
 		sql_expr_delete(db, pTmp->pWhere, false);
 		sql_expr_list_delete(db, pTmp->pExprList);
-		sqlite3SelectDelete(db, pTmp->pSelect);
+		sql_select_delete(db, pTmp->pSelect);
 		sqlite3IdListDelete(db, pTmp->pIdList);
 
 		sqlite3DbFree(db, pTmp);
@@ -346,7 +346,7 @@ sqlite3TriggerSelectStep(sqlite3 * db, Select * pSelect)
 	TriggerStep *pTriggerStep =
 	    sqlite3DbMallocZero(db, sizeof(TriggerStep));
 	if (pTriggerStep == 0) {
-		sqlite3SelectDelete(db, pSelect);
+		sql_select_delete(db, pSelect);
 		return 0;
 	}
 	pTriggerStep->op = TK_SELECT;
@@ -412,7 +412,7 @@ sqlite3TriggerInsertStep(sqlite3 * db,	/* The database connection */
 	} else {
 		sqlite3IdListDelete(db, pColumn);
 	}
-	sqlite3SelectDelete(db, pSelect);
+	sql_select_delete(db, pSelect);
 
 	return pTriggerStep;
 }
@@ -758,7 +758,7 @@ codeTriggerProgram(Parse * pParse,	/* The parser context */
 				    sqlite3SelectDup(db, pStep->pSelect, 0);
 				sqlite3SelectDestInit(&sDest, SRT_Discard, 0);
 				sqlite3Select(pParse, pSelect, &sDest);
-				sqlite3SelectDelete(db, pSelect);
+			sql_select_delete(db, pSelect);
 				break;
 			}
 		}
diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
index 3fe5875ca..5b5cf835d 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -2983,6 +2983,30 @@ case OP_FkCheckCommit: {
 	break;
 }
 
+/* Opcode: CheckViewReferences P1 * * * *
+ * Synopsis: r[P1] = space id
+ *
+ * Check that space to be dropped doesn't have any view
+ * references. This opcode is needed since Tarantool lacks
+ * DDL transaction. On the other hand, to drop space, we must
+ * firstly drop secondary indexes from _index system space,
+ * clear _truncate table etc.
+ */
+case OP_CheckViewReferences: {
+	assert(pOp->p1 > 0);
+	pIn1 = &aMem[pOp->p1];
+	uint32_t space_id = pIn1->u.i;
+	struct space *space = space_by_id(space_id);
+	assert(space != NULL);
+	if (space->def->view_ref_count > 0) {
+		sqlite3VdbeError(p,"Can't drop table %s: other views depend "
+				 "on this space",  space->def->name);
+		rc = SQL_TARANTOOL_ERROR;
+		goto abort_due_to_error;
+	}
+	break;
+}
+
 /* Opcode: TransactionBegin * * * * *
  *
  * Start Tarantool's transaction.
diff --git a/test/sql-tap/colname.test.lua b/test/sql-tap/colname.test.lua
index c96623e4e..c53a1e885 100755
--- a/test/sql-tap/colname.test.lua
+++ b/test/sql-tap/colname.test.lua
@@ -623,14 +623,14 @@ end
 test:do_test(
     "colname-11.0.1",
     function ()
-        test:drop_all_tables()
+        test:drop_all_views()
     end,
     nil)
 
 test:do_test(
     "colname-11.0.2",
     function ()
-        test:drop_all_views()
+        test:drop_all_tables()
     end,
     nil)
 
diff --git a/test/sql-tap/drop_all.test.lua b/test/sql-tap/drop_all.test.lua
index b9abe7594..d4f6c6073 100755
--- a/test/sql-tap/drop_all.test.lua
+++ b/test/sql-tap/drop_all.test.lua
@@ -22,7 +22,7 @@ test:do_test(
 test:do_test(
     prefix.."-1.1",
     function()
-        return #test:drop_all_tables()
+        return #test:drop_all_views()
     end,
     N
 )
@@ -30,7 +30,7 @@ test:do_test(
 test:do_test(
     prefix.."-1.1",
     function()
-        return #test:drop_all_views()
+        return #test:drop_all_tables()
     end,
     N
 )
diff --git a/test/sql-tap/fkey2.test.lua b/test/sql-tap/fkey2.test.lua
index f90c218d4..1f3a81d92 100755
--- a/test/sql-tap/fkey2.test.lua
+++ b/test/sql-tap/fkey2.test.lua
@@ -743,6 +743,7 @@ test:do_catchsql_test(
 test:do_catchsql_test(
     "fkey2-7.3",
     [[
+	DROP VIEW v;
         DROP TABLE IF EXISTS c;
         CREATE TABLE p(a COLLATE binary, b PRIMARY KEY);
         CREATE UNIQUE INDEX idx ON p(a COLLATE "unicode_ci");
diff --git a/test/sql-tap/trigger1.test.lua b/test/sql-tap/trigger1.test.lua
index 40daba4d8..dab3281a3 100755
--- a/test/sql-tap/trigger1.test.lua
+++ b/test/sql-tap/trigger1.test.lua
@@ -460,6 +460,7 @@ test:do_catchsql_test(
 -- }
 test:execsql [[
     CREATE TABLE t2(x int PRIMARY KEY,y);
+    DROP VIEW v1;
     DROP TABLE t1;
     INSERT INTO t2 VALUES(3, 4);
     INSERT INTO t2 VALUES(7, 8);
diff --git a/test/sql-tap/triggerC.test.lua b/test/sql-tap/triggerC.test.lua
index 347007832..e58072e2f 100755
--- a/test/sql-tap/triggerC.test.lua
+++ b/test/sql-tap/triggerC.test.lua
@@ -1097,6 +1097,7 @@ test:do_catchsql_test(
 --
 SQL = [[
   DROP TABLE IF EXISTS t1;
+  DROP VIEW v2;
   DROP TABLE IF EXISTS t2;
   DROP TABLE IF EXISTS t4;
   DROP TABLE IF EXISTS t5;
diff --git a/test/sql-tap/view.test.lua b/test/sql-tap/view.test.lua
index 8ba2044f0..c24e3fb29 100755
--- a/test/sql-tap/view.test.lua
+++ b/test/sql-tap/view.test.lua
@@ -1,6 +1,6 @@
 #!/usr/bin/env tarantool
 test = require("sqltester")
-test:plan(65)
+test:plan(74)
 
 --!./tcltestrunner.lua
 -- 2002 February 26
@@ -125,13 +125,23 @@ test:do_catchsql_test(
     "view-1.6",
     [[
         DROP TABLE t1;
-        SELECT * FROM v1 ORDER BY a;
     ]], {
         -- <view-1.6>
-        1, "no such table: T1"
+        1, "Can't drop table T1: other views depend on this space"
         -- </view-1.6>
     })
 
+test:do_catchsql_test(
+    "view-1.7",
+    [[
+        DROP VIEW v1;
+        DROP TABLE t1;
+    ]], {
+        -- <view-1.7>
+        0
+        -- </view-1.7>
+    })
+
 -- ORIGINAL_TEST
 -- do_test view-1.7 {
 --   execsql {
@@ -143,12 +153,13 @@ test:do_catchsql_test(
 --   }
 -- } {2 3 5 6 8 9}
 test:do_execsql_test(
-    "view-1.7",
+    "view-1.8",
     [[
         CREATE TABLE t1(x primary key,a,b,c);
         INSERT INTO t1 VALUES(1,2,3,4);
         INSERT INTO t1 VALUES(4,5,6,7);
         INSERT INTO t1 VALUES(7,8,9,10);
+        CREATE VIEW v1 AS SELECT a,b FROM t1;
         SELECT * FROM v1 ORDER BY a;
     ]], {
         -- <view-1.7>
@@ -156,24 +167,6 @@ test:do_execsql_test(
         -- </view-1.7>
     })
 
--- MUST_WORK_TEST reopen db
-if (0 > 0)
- then
-    test:do_test(
-        "view-1.8",
-        function()
-            db("close")
-            sqlite3("db", "test.db")
-            return test:execsql [[
-                SELECT * FROM v1 ORDER BY a;
-            ]]
-        end, {
-            -- <view-1.8>
-            2, 3, 5, 6, 8, 9
-            -- </view-1.8>
-        })
-
-end
 test:do_test(
     "view-2.1",
     function()
@@ -1065,10 +1058,9 @@ end
 test:do_execsql_test(
     "view-20.1",
     [[
-        DROP TABLE IF EXISTS t1;
-        DROP VIEW IF EXISTS v1;
-        CREATE TABLE t1(c1 primary key);
-        CREATE VIEW v1 AS SELECT c1 FROM (SELECT t1.c1 FROM t1);
+	DROP VIEW v10;
+        CREATE TABLE t10(c1 primary key);
+        CREATE VIEW v10 AS SELECT c1 FROM (SELECT t10.c1 FROM t10);
     ]], {
         -- <view-20.1>
         
@@ -1078,10 +1070,10 @@ test:do_execsql_test(
 test:do_execsql_test(
     "view-20.1",
     [[
-        DROP TABLE IF EXISTS t1;
-        DROP VIEW IF EXISTS v1;
-        CREATE TABLE t1(c1 primary key);
-        CREATE VIEW v1 AS SELECT c1 FROM (SELECT t1.c1 FROM t1);
+	DROP VIEW IF EXISTS v10;
+        DROP TABLE IF EXISTS t10;
+        CREATE TABLE t10(c1 primary key);
+        CREATE VIEW v10 AS SELECT c1 FROM (SELECT t10.c1 FROM t10);
     ]], {
         -- <view-20.1>
         
@@ -1171,5 +1163,90 @@ if (0 > 0)
 
 end
 
+-- Make sure that VIEW with several internal selects works.
+test:do_catchsql_test(
+    "view-23.1",
+    [[
+        CREATE TABLE t11(a INT PRIMARY KEY);
+        CREATE TABLE t12(b INT PRIMARY KEY);
+        CREATE TABLE t13(c INT PRIMARY KEY);
+        CREATE VIEW v11 AS SELECT * FROM
+            (SELECT a FROM (SELECT a, b FROM t11, t12)),
+            (SELECT * FROM (SELECT a, c FROM t11, t13));
+    ]], {
+        -- <view-23.1>
+        0
+        -- </view-23.1>
+    })
+
+test:do_catchsql_test(
+    "view-23.2",
+    [[
+        DROP TABLE t11;
+    ]], {
+        -- <view-23.2>
+        1, "Can't drop table T11: other views depend on this space"
+        -- </view-23.2>
+    })
+
+test:do_catchsql_test(
+    "view-23.3",
+    [[
+        DROP TABLE t12;
+    ]], {
+        -- <view-23.3>
+        1, "Can't drop table T12: other views depend on this space"
+        -- </view-23.3>
+    })
+
+test:do_catchsql_test(
+    "view-23.4",
+    [[
+        DROP TABLE t13;
+    ]], {
+        -- <view-23.4>
+        1, "Can't drop table T13: other views depend on this space"
+        -- </view-23.4>
+    })
+
+test:do_catchsql_test(
+    "view-23.5",
+    [[
+        DROP VIEW v11;
+    ]], {
+        -- <view-23.5>
+        0
+        -- </view-23.5>
+    })
+
+test:do_catchsql_test(
+    "view-23.6",
+    [[
+        DROP TABLE t11;
+    ]], {
+        -- <view-23.6>
+        0
+        -- </view-23.6>
+    })
+
+test:do_catchsql_test(
+    "view-23.7",
+    [[
+        DROP TABLE t12;
+    ]], {
+        -- <view-23.7>
+        0
+        -- </view-23.7>
+    })
+
+test:do_catchsql_test(
+    "view-23.8",
+    [[
+        DROP TABLE t13;
+    ]], {
+        -- <view-23.8>
+        0
+        -- </view-23.8>
+    })
 
 test:finish_test()
diff --git a/test/sql/view.result b/test/sql/view.result
index 04cb3fae8..50c29b991 100644
--- a/test/sql/view.result
+++ b/test/sql/view.result
@@ -73,7 +73,7 @@ raw_sp = box.space._space:get(sp.id):totable();
 sp:drop();
 ---
 ...
-raw_sp[6].sql = 'fake';
+raw_sp[6].sql = 'SELECT 1';
 ---
 ...
 raw_sp[6].view = true;
@@ -86,13 +86,30 @@ box.space._space:select(sp['id'])[1]['name']
 ---
 - test
 ...
+-- Can't drop space via Lua if at least one view refers to it.
+box.sql.execute('CREATE TABLE t2(id INT PRIMARY KEY);');
+---
+...
+box.sql.execute('CREATE VIEW v2 AS SELECT * FROM t2;');
+---
+...
+box.space.T2:drop();
+---
+- error: 'Can''t drop space ''T2'': other views depend on this space'
+...
+box.sql.execute('DROP VIEW v2;');
+---
+...
+box.sql.execute('DROP TABLE t2;');
+---
+...
 -- Cleanup
 box.space.test:drop();
 ---
 ...
-box.sql.execute("DROP TABLE t1;");
+box.sql.execute("DROP VIEW v1;");
 ---
 ...
-box.sql.execute("DROP VIEW v1;");
+box.sql.execute("DROP TABLE t1;");
 ---
 ...
diff --git a/test/sql/view.test.lua b/test/sql/view.test.lua
index ab2843cb6..7c9819337 100644
--- a/test/sql/view.test.lua
+++ b/test/sql/view.test.lua
@@ -35,12 +35,19 @@ box.schema.create_space('view', {view = true})
 sp = box.schema.create_space('test');
 raw_sp = box.space._space:get(sp.id):totable();
 sp:drop();
-raw_sp[6].sql = 'fake';
+raw_sp[6].sql = 'SELECT 1';
 raw_sp[6].view = true;
 sp = box.space._space:replace(raw_sp);
 box.space._space:select(sp['id'])[1]['name']
 
+-- Can't drop space via Lua if at least one view refers to it.
+box.sql.execute('CREATE TABLE t2(id INT PRIMARY KEY);');
+box.sql.execute('CREATE VIEW v2 AS SELECT * FROM t2;');
+box.space.T2:drop();
+box.sql.execute('DROP VIEW v2;');
+box.sql.execute('DROP TABLE t2;');
+
 -- Cleanup
 box.space.test:drop();
-box.sql.execute("DROP TABLE t1;");
 box.sql.execute("DROP VIEW v1;");
+box.sql.execute("DROP TABLE t1;");
-- 
2.15.1

