part 1 (ver 2, postreview fix) of WL#2787

view definer information syntax/storage/replication
fixed SOURCE field of .frm

mysql-test/r/func_in.result

@@ -209,7 +209,7 @@ a
 CREATE VIEW v1 AS SELECT * FROM t1 WHERE a NOT IN (45);
 SHOW CREATE VIEW v1;
 View	Create View
-v1	CREATE ALGORITHM=UNDEFINED VIEW `test`.`v1` AS select `test`.`t1`.`a` AS `a` from `test`.`t1` where (`test`.`t1`.`a` <> 45)
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `test`.`v1` AS select `test`.`t1`.`a` AS `a` from `test`.`t1` where (`test`.`t1`.`a` <> 45)
 SELECT * FROM v1;
 a
 44

mysql-test/r/lowercase_view.result

@@ -7,7 +7,7 @@ create table TaB (Field int);
 create view ViE as select * from TAb;
 show create table VIe;
 View	Create View
-vie	CREATE ALGORITHM=UNDEFINED VIEW `mysqltest`.`vie` AS select `mysqltest`.`tab`.`Field` AS `Field` from `mysqltest`.`tab`
+vie	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `mysqltest`.`vie` AS select `mysqltest`.`tab`.`Field` AS `Field` from `mysqltest`.`tab`
 drop database MySQLTest;
 use test;
 create table t1Aa (col1 int);
@@ -119,7 +119,7 @@ create table t1Aa (col1 int);
 create view v1Aa as select col1 from t1Aa as AaA;
 show create view v1AA;
 View	Create View
-v1aa	CREATE ALGORITHM=UNDEFINED VIEW `test`.`v1aa` AS select `aaa`.`col1` AS `col1` from `test`.`t1aa` `AaA`
+v1aa	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `test`.`v1aa` AS select `aaa`.`col1` AS `col1` from `test`.`t1aa` `AaA`
 drop view v1AA;
 select Aaa.col1 from t1Aa as AaA;
 col1
@@ -128,6 +128,6 @@ drop view v1AA;
 create view v1Aa as select AaA.col1 from t1Aa as AaA;
 show create view v1AA;
 View	Create View
-v1aa	CREATE ALGORITHM=UNDEFINED VIEW `test`.`v1aa` AS select `aaa`.`col1` AS `col1` from `test`.`t1aa` `AaA`
+v1aa	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `test`.`v1aa` AS select `aaa`.`col1` AS `col1` from `test`.`t1aa` `AaA`
 drop view v1AA;
 drop table t1Aa;

mysql-test/r/mysqldump.result

@@ -1549,7 +1549,7 @@ DROP TABLE IF EXISTS `v1`;
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1*/;
 /*!50001 DROP TABLE IF EXISTS `v1`*/;
 /*!50001 DROP VIEW IF EXISTS `v1`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED VIEW `test`.`v1` AS select `test`.`t1`.`a` AS `a` from `test`.`t1`*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `test`.`v1` AS select `test`.`t1`.`a` AS `a` from `test`.`t1`*/;
 
 /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
 /*!40014 SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS */;
@@ -1601,7 +1601,7 @@ DROP TABLE IF EXISTS `v2`;
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1*/;
 /*!50001 DROP TABLE IF EXISTS `v2`*/;
 /*!50001 DROP VIEW IF EXISTS `v2`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED VIEW `mysqldump_test_db`.`v2` AS select `mysqldump_test_db`.`t2`.`a` AS `a` from `mysqldump_test_db`.`t2` where (`mysqldump_test_db`.`t2`.`a` like _latin1'a%') WITH CASCADED CHECK OPTION*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `mysqldump_test_db`.`v2` AS select `mysqldump_test_db`.`t2`.`a` AS `a` from `mysqldump_test_db`.`t2` where (`mysqldump_test_db`.`t2`.`a` like _latin1'a%') WITH CASCADED CHECK OPTION*/;
 
 /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
 /*!40014 SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS */;
@@ -1665,7 +1665,7 @@ v2	VIEW
 v3	VIEW
 show create view v1;
 View	Create View
-v1	CREATE ALGORITHM=UNDEFINED VIEW `test`.`v1` AS select `v3`.`a` AS `a`,`v3`.`b` AS `b`,`v3`.`c` AS `c` from `test`.`v3` where (`v3`.`b` in (1,2,3,4,5,6,7))
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `test`.`v1` AS select `v3`.`a` AS `a`,`v3`.`b` AS `b`,`v3`.`c` AS `c` from `test`.`v3` where (`v3`.`b` in (1,2,3,4,5,6,7))
 select * from v1;
 a	b	c
 1	2	one

mysql-test/r/rpl_view.result

@@ -6,6 +6,7 @@ drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
 start slave;
 drop table if exists t1,v1;
 drop view if exists t1,v1;
+reset master;
 create table t1 (a int);
 insert into t1 values (1);
 create view v1 as select a from t1;
@@ -42,3 +43,15 @@ drop view v1;
 select * from v1 order by a;
 ERROR 42S02: Table 'test.v1' doesn't exist
 drop table t1;
+show binlog events;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000001	#	Format_desc	2	#	Server ver: 5.0.13-beta-debug-log, Binlog ver: 4
+slave-bin.000001	#	Query	1	#	use `test`; create table t1 (a int)
+slave-bin.000001	#	Query	1	#	use `test`; insert into t1 values (1)
+slave-bin.000001	#	Query	1	#	use `test`; CREATE ALGORITHM=UNDEFINED DEFINER=root@localhost SQL SECURITY DEFINER VIEW v1 AS select a from t1
+slave-bin.000001	#	Query	1	#	use `test`; insert into v1 values (2)
+slave-bin.000001	#	Query	1	#	use `test`; update v1 set a=3 where a=1
+slave-bin.000001	#	Query	1	#	use `test`; delete from v1 where a=2
+slave-bin.000001	#	Query	1	#	use `test`; ALTER ALGORITHM=UNDEFINED DEFINER=root@localhost SQL SECURITY DEFINER VIEW v1 AS select a as b from t1
+slave-bin.000001	#	Query	1	#	use `test`; drop view v1
+slave-bin.000001	#	Query	1	#	use `test`; drop table t1

mysql-test/r/skip_grants.result

@@ -4,7 +4,7 @@ drop procedure if exists f1;
 use test;
 create table t1 (field1 INT);
 CREATE VIEW v1 AS SELECT field1 FROM t1;
-drop view v1;
+ERROR HY000: View definer is not fully qualified
 drop table t1;
 create procedure f1() select 1;
 drop procedure f1;

mysql-test/r/sql_mode.result

@@ -449,11 +449,11 @@ create table t2 (a int);
 create view v1 as select a from t1;
 show create view v1;
 View	Create View
-v1	CREATE ALGORITHM=UNDEFINED VIEW `test`.`v1` AS select `test`.`t1`.`a` AS `a` from `test`.`t1`
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `test`.`v1` AS select `test`.`t1`.`a` AS `a` from `test`.`t1`
 SET @@SQL_MODE='ANSI_QUOTES';
 show create view v1;
 View	Create View
-v1	CREATE ALGORITHM=UNDEFINED VIEW "test"."v1" AS select "test"."t1"."a" AS "a" from "test"."t1"
+v1	CREATE ALGORITHM=UNDEFINED DEFINER="root"@"localhost" SQL SECURITY DEFINER VIEW "test"."v1" AS select "test"."t1"."a" AS "a" from "test"."t1"
 create view v2 as select a from t2 where a in (select a from v1);
 drop view v2, v1;
 drop table t1, t2;

mysql-test/r/temp_table.result

@@ -109,7 +109,7 @@ t1	CREATE TEMPORARY TABLE `t1` (
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1
 show create view t1;
 View	Create View
-t1	CREATE ALGORITHM=UNDEFINED VIEW `test`.`t1` AS select _latin1'This is view' AS `A`
+t1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `test`.`t1` AS select _latin1'This is view' AS `A`
 drop view t1;
 select * from t1;
 A

mysql-test/r/view_grant.result

@@ -12,6 +12,8 @@ create table mysqltest.t1 (a int, b int);
 create table mysqltest.t2 (a int, b int);
 grant select on mysqltest.t1 to mysqltest_1@localhost;
 grant create view,select on test.* to mysqltest_1@localhost;
+create definer=root@localhost view v1 as select * from mysqltest.t1;
+ERROR HY000: You need the SUPER privilege for creation view with root@localhost definer
 create view v1 as select * from mysqltest.t1;
 alter view v1 as select * from mysqltest.t1;
 ERROR 42000: DROP command denied to user 'mysqltest_1'@'localhost' for table 'v1'
@@ -21,6 +23,9 @@ create view mysqltest.v2  as select * from mysqltest.t1;
 ERROR 42000: CREATE VIEW command denied to user 'mysqltest_1'@'localhost' for table 'v2'
 create view v2 as select * from mysqltest.t2;
 ERROR 42000: ANY command denied to user 'mysqltest_1'@'localhost' for table 't2'
+show create view v1;
+View	Create View
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`mysqltest_1`@`localhost` SQL SECURITY DEFINER VIEW `test`.`v1` AS select `mysqltest`.`t1`.`a` AS `a`,`mysqltest`.`t1`.`b` AS `b` from `mysqltest`.`t1`
 grant create view,drop,select on test.* to mysqltest_1@localhost;
 use test;
 alter view v1 as select * from mysqltest.t1;
@@ -120,27 +125,27 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 1	PRIMARY	t1	system	NULL	NULL	NULL	NULL	0	const row not found
 show create view mysqltest.v1;
 View	Create View
-v1	CREATE ALGORITHM=UNDEFINED VIEW `mysqltest`.`v1` AS select (`mysqltest`.`t1`.`a` + 1) AS `c`,(`mysqltest`.`t1`.`b` + 1) AS `d` from `mysqltest`.`t1`
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `mysqltest`.`v1` AS select (`mysqltest`.`t1`.`a` + 1) AS `c`,(`mysqltest`.`t1`.`b` + 1) AS `d` from `mysqltest`.`t1`
 explain select c from mysqltest.v2;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 1	PRIMARY	<derived2>	system	NULL	NULL	NULL	NULL	0	const row not found
 2	DERIVED	NULL	NULL	NULL	NULL	NULL	NULL	NULL	no matching row in const table
 show create view mysqltest.v2;
 View	Create View
-v2	CREATE ALGORITHM=TEMPTABLE VIEW `mysqltest`.`v2` AS select (`mysqltest`.`t1`.`a` + 1) AS `c`,(`mysqltest`.`t1`.`b` + 1) AS `d` from `mysqltest`.`t1`
+v2	CREATE ALGORITHM=TEMPTABLE DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `mysqltest`.`v2` AS select (`mysqltest`.`t1`.`a` + 1) AS `c`,(`mysqltest`.`t1`.`b` + 1) AS `d` from `mysqltest`.`t1`
 explain select c from mysqltest.v3;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 1	PRIMARY	t2	system	NULL	NULL	NULL	NULL	0	const row not found
 show create view mysqltest.v3;
 View	Create View
-v3	CREATE ALGORITHM=UNDEFINED VIEW `mysqltest`.`v3` AS select (`mysqltest`.`t2`.`a` + 1) AS `c`,(`mysqltest`.`t2`.`b` + 1) AS `d` from `mysqltest`.`t2`
+v3	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `mysqltest`.`v3` AS select (`mysqltest`.`t2`.`a` + 1) AS `c`,(`mysqltest`.`t2`.`b` + 1) AS `d` from `mysqltest`.`t2`
 explain select c from mysqltest.v4;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 1	PRIMARY	<derived2>	system	NULL	NULL	NULL	NULL	0	const row not found
 2	DERIVED	NULL	NULL	NULL	NULL	NULL	NULL	NULL	no matching row in const table
 show create view mysqltest.v4;
 View	Create View
-v4	CREATE ALGORITHM=TEMPTABLE VIEW `mysqltest`.`v4` AS select (`mysqltest`.`t2`.`a` + 1) AS `c`,(`mysqltest`.`t2`.`b` + 1) AS `d` from `mysqltest`.`t2`
+v4	CREATE ALGORITHM=TEMPTABLE DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `mysqltest`.`v4` AS select (`mysqltest`.`t2`.`a` + 1) AS `c`,(`mysqltest`.`t2`.`b` + 1) AS `d` from `mysqltest`.`t2`
 revoke all privileges on mysqltest.* from mysqltest_1@localhost;
 delete from mysql.user where user='mysqltest_1';
 drop database mysqltest;

mysql-test/t/rpl_view.test

@@ -3,6 +3,7 @@ source include/master-slave.inc;
 drop table if exists t1,v1;
 drop view if exists t1,v1;
 sync_slave_with_master;
+reset master;
 --enable_warnings
 
 #
@@ -42,3 +43,5 @@ select * from v1 order by a;
 connection master;
 drop table t1;
 sync_slave_with_master;
+--replace_column 2 # 5 #
+show binlog events;

mysql-test/t/skip_grants.test

@@ -9,9 +9,8 @@ use test;
 # test that we can create VIEW if privileges check switched off
 #
 create table t1 (field1 INT);
+-- error ER_NO_VIEW_USER
 CREATE VIEW v1 AS SELECT field1 FROM t1;
-
-drop view v1;
 drop table t1;
 
 #

mysql-test/t/view.test

@@ -2059,3 +2059,12 @@ order by users_names;
 
 drop view v1, v2;
 drop table t1, t2;
+
+#
+# DEFINER information check
+#
+-- error ER_NO_VIEW_USER
+create definer=some_user@__% sql security invoker view v1 as select 1;
+create definer=some_user@localhost sql security invoker view v1 as select 1;
+show create view v1;
+drop view v1;

mysql-test/t/view_grant.test

@@ -24,6 +24,8 @@ grant create view,select on test.* to mysqltest_1@localhost;
 connect (user1,localhost,mysqltest_1,,test);
 connection user1;
 
+-- error ER_VIEW_OTHER_USER
+create definer=root@localhost view v1 as select * from mysqltest.t1;
 create view v1 as select * from mysqltest.t1;
 # try to modify view without DROP privilege on it
 -- error 1142
@@ -38,6 +40,9 @@ create view mysqltest.v2  as select * from mysqltest.t1;
 create view v2 as select * from mysqltest.t2;
 
 connection root;
+# check view definer information
+show create view v1;
+
 grant create view,drop,select on test.* to mysqltest_1@localhost;
 
 connection user1;

sql/mysql_priv.h

@@ -379,6 +379,10 @@ void debug_sync_point(const char* lock_name, uint lock_timeout);
 #define SHOW_LOG_STATUS_FREE "FREE"
 #define SHOW_LOG_STATUS_INUSE "IN USE"
 
+struct st_table_list;
+class String;
+void view_store_options(THD *thd, st_table_list *table, String *buff);
+
 /* Options to add_table_to_list() */
 #define TL_OPTION_UPDATING	1
 #define TL_OPTION_FORCE_INDEX	2
@@ -511,6 +515,8 @@ bool delete_precheck(THD *thd, TABLE_LIST *tables);
 bool insert_precheck(THD *thd, TABLE_LIST *tables);
 bool create_table_precheck(THD *thd, TABLE_LIST *tables,
                            TABLE_LIST *create_table);
+bool default_view_definer(THD *thd, st_lex_user *definer);
+
 
 enum enum_mysql_completiontype {
   ROLLBACK_RELEASE=-2, ROLLBACK=1,  ROLLBACK_AND_CHAIN=7,

sql/share/errmsg.txt

@@ -5403,3 +5403,11 @@ ER_VIEW_PREVENT_UPDATE
         eng "The definition of table '%-.64s' prevents operation %s on table '%-.64s'."
 ER_PS_NO_RECURSION
         eng "The prepared statement contains a stored routine call that refers to that same statement. It's not allowed to execute a prepared statement in such a recursive manner"
+ER_NO_VIEW_USER
+        eng "View definer is not fully qualified"
+ER_VIEW_FRM_NO_USER
+        eng "View %-.64s.%-.64s has not definer information (old table format). Current user is used as definer. Please recreate view!"
+ER_VIEW_OTHER_USER
+        eng "You need the SUPER privilege for creation view with %-.64s@%-.64s definer"
+ER_NO_SUCH_USER
+        eng "There is not %-.64s@%-.64s registered"

sql/sql_acl.cc

@@ -1456,6 +1456,29 @@ end:
 }
 
 
+/*
+  Find user in ACL
+
+  SYNOPSIS
+    is_acl_user()
+    host                 host name
+    user                 user name
+
+  RETURN
+   FALSE  user not fond
+   TRUE   there are such user
+*/
+
+bool is_acl_user(const char *host, const char *user)
+{
+  bool res;
+  VOID(pthread_mutex_lock(&acl_cache->lock));
+  res= find_acl_user(host, user, TRUE);
+  VOID(pthread_mutex_unlock(&acl_cache->lock));
+  return res;
+}
+
+
 /*
   Find first entry that matches the current user
 */

sql/sql_acl.h

@@ -229,7 +229,7 @@ bool sp_grant_privileges(THD *thd, const char *sp_db, const char *sp_name,
                          bool is_proc);
 bool check_routine_level_acl(THD *thd, const char *db, const char *name,
                              bool is_proc);
-
+bool is_acl_user(const char *host, const char *user);
 #ifdef NO_EMBEDDED_ACCESS_CHECKS
 #define check_grant(A,B,C,D,E,F) 0
 #define check_grant_db(A,B) 0

sql/sql_lex.h

@@ -733,6 +733,8 @@ typedef struct st_lex
   TABLE_LIST **query_tables_last;
   /* store original leaf_tables for INSERT SELECT and PS/SP */
   TABLE_LIST *leaf_tables_insert;
+  st_lex_user *create_view_definer;
+  char *create_view_select_start;
 
   List<key_part_spec> col_list;
   List<key_part_spec> ref_list;
@@ -853,6 +855,10 @@ typedef struct st_lex
     rexecuton
   */
   bool empty_field_list_on_rset;
+  /*
+    view created to be run from definer (standard behaviour)
+  */
+  bool create_view_suid;
   /* Characterstics of trigger being created */
   st_trg_chistics trg_chistics;
   /*

sql/sql_parse.cc

@@ -4451,8 +4451,29 @@ end_with_restore_list:
       if (!(res= mysql_create_view(thd, thd->lex->create_view_mode)) &&
           mysql_bin_log.is_open())
       {
+        String buff;
+        LEX_STRING command[3]=
+          {{STRING_WITH_LEN("CREATE ")},
+           {STRING_WITH_LEN("ALTER ")},
+           {STRING_WITH_LEN("CREATE OR REPLACE ")}};
         thd->clear_error();
-        Query_log_event qinfo(thd, thd->query, thd->query_length, 0, FALSE);
+
+        buff.append(command[thd->lex->create_view_mode].str,
+                    command[thd->lex->create_view_mode].length);
+        view_store_options(thd, first_table, &buff);
+        buff.append("VIEW ", 5);
+        if (!first_table->current_db_used)
+        {
+          append_identifier(thd, &buff, first_table->db,
+                            first_table->db_length);
+          buff.append('.');
+        }
+        append_identifier(thd, &buff, first_table->table_name,
+                          first_table->table_name_length);
+        buff.append(" AS ", 4);
+        buff.append(first_table->source.str, first_table->source.length);
+
+        Query_log_event qinfo(thd, buff.ptr(), buff.length(), 0, FALSE);
         mysql_bin_log.write(&qinfo);
       }
       break;
@@ -6009,12 +6030,14 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
   {
     ptr->db= thd->db;
     ptr->db_length= thd->db_length;
+    ptr->current_db_used= 1;
   }
   else
   {
     /* The following can't be "" as we may do 'casedn_str()' on it */
     ptr->db= empty_c_string;
     ptr->db_length= 0;
+    ptr->current_db_used= 1;
   }
   if (thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
     ptr->db= thd->strdup(ptr->db);
@@ -7294,3 +7317,34 @@ Item *negate_expression(THD *thd, Item *expr)
     return negated;
   return new Item_func_not(expr);
 }
+
+
+/*
+  Assign as view definer current user
+
+  SYNOPSIS
+    default_definer()
+    thd                  thread handler
+    definer              structure where it should be assigned
+
+  RETURN
+    FALSE   OK
+    TRUE    Error
+*/
+
+bool default_view_definer(THD *thd, st_lex_user *definer)
+{
+  definer->user.str= thd->priv_user;
+  definer->user.length= strlen(thd->priv_user);
+  if (*thd->priv_host != 0)
+  {
+    definer->host.str= thd->priv_host;
+    definer->host.length= strlen(thd->priv_host);
+  }
+  else
+  {
+    my_error(ER_NO_VIEW_USER, MYF(0));
+    return TRUE;
+  }
+  return FALSE;
+}

sql/sql_show.cc

@@ -42,7 +42,7 @@ static TYPELIB grant_types = { sizeof(grant_names)/sizeof(char **),
 static int
 store_create_info(THD *thd, TABLE_LIST *table_list, String *packet);
 static int
-view_store_create_info(THD *thd, TABLE_LIST *table, String *packet);
+view_store_create_info(THD *thd, TABLE_LIST *table, String *buff);
 static bool schema_table_store_record(THD *thd, TABLE *table);
 
 
@@ -1045,6 +1045,34 @@ store_create_info(THD *thd, TABLE_LIST *table_list, String *packet)
   DBUG_RETURN(0);
 }
 
+void
+view_store_options(THD *thd, TABLE_LIST *table, String *buff)
+{
+  buff->append("ALGORITHM=", 10);
+  switch ((int8)table->algorithm) {
+  case VIEW_ALGORITHM_UNDEFINED:
+    buff->append("UNDEFINED ", 10);
+    break;
+  case VIEW_ALGORITHM_TMPTABLE:
+    buff->append("TEMPTABLE ", 10);
+    break;
+  case VIEW_ALGORITHM_MERGE:
+    buff->append("MERGE ", 6);
+    break;
+  default:
+    DBUG_ASSERT(0); // never should happen
+  }
+  buff->append("DEFINER=", 8);
+  append_identifier(thd, buff,
+                    table->definer.user.str, table->definer.user.length);
+  buff->append('@');
+  append_identifier(thd, buff,
+                    table->definer.host.str, table->definer.host.length);
+  if (table->view_suid)
+    buff->append(" SQL SECURITY DEFINER ", 22);
+  else
+    buff->append(" SQL SECURITY INVOKER ", 22);
+}
 
 static int
 view_store_create_info(THD *thd, TABLE_LIST *table, String *buff)
@@ -1058,21 +1086,7 @@ view_store_create_info(THD *thd, TABLE_LIST *table, String *buff)
   buff->append("CREATE ", 7);
   if (!foreign_db_mode)
   {
-    buff->append("ALGORITHM=", 10);
-    switch((int8)table->algorithm)
-    {
-    case VIEW_ALGORITHM_UNDEFINED:
-      buff->append("UNDEFINED ", 10);
-      break;
-    case VIEW_ALGORITHM_TMPTABLE:
-      buff->append("TEMPTABLE ", 10);
-      break;
-    case VIEW_ALGORITHM_MERGE:
-      buff->append("MERGE ", 6);
-      break;
-    default:
-	DBUG_ASSERT(0); // never should happen
-    }
+    view_store_options(thd, table, buff);
   }
   buff->append("VIEW ", 5);
   append_identifier(thd, buff, table->view_db.str, table->view_db.length);

sql/sql_view.cc

@@ -209,6 +209,36 @@ bool mysql_create_view(THD *thd,
     sp_cache_invalidate();
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
+  /*
+    check definer of view:
+      - same as current user
+      - current user has SUPER_ACL
+  */
+  if (strcmp(lex->create_view_definer->user.str, thd->priv_user) != 0 ||
+      my_strcasecmp(system_charset_info,
+                    lex->create_view_definer->host.str,
+                    thd->priv_host) != 0)
+  {
+    if (!(thd->master_access & SUPER_ACL))
+    {
+      my_error(ER_VIEW_OTHER_USER, MYF(0), lex->create_view_definer->user.str,
+               lex->create_view_definer->host.str);
+      res= TRUE;
+      goto err;
+    }
+    else
+    {
+      if (!is_acl_user(lex->create_view_definer->host.str,
+                       lex->create_view_definer->user.str))
+      {
+        push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
+                            ER_NO_SUCH_USER,
+                            ER(ER_NO_SUCH_USER),
+                            lex->create_view_definer->user.str,
+                            lex->create_view_definer->host.str);
+      }
+    }
+  }
   /*
     Privilege check for view creation:
     - user has CREATE VIEW privilege on view table
@@ -369,6 +399,7 @@ bool mysql_create_view(THD *thd,
     if (lex->view_list.elements != select_lex->item_list.elements)
     {
       my_message(ER_VIEW_WRONG_LIST, ER(ER_VIEW_WRONG_LIST), MYF(0));
+      res= TRUE;
       goto err;
     }
     while ((item= it++, name= nm++))
@@ -447,9 +478,9 @@ err:
 
 
 /* index of revision number in following table */
-static const int revision_number_position= 5;
+static const int revision_number_position= 8;
 /* index of last required parameter for making view */
-static const int required_view_parameters= 7;
+static const int required_view_parameters= 10;
 
 /*
   table of VIEW .frm field descriptors
@@ -458,23 +489,41 @@ static const int required_view_parameters= 7;
   parse()
 */
 static File_option view_parameters[]=
-{{{(char*) "query", 5},		offsetof(TABLE_LIST, query),
+{{{(char*) STRING_WITH_LEN("query")},
+  offsetof(TABLE_LIST, query),
+  FILE_OPTIONS_STRING},
+ {{(char*) STRING_WITH_LEN("md5")},
+  offsetof(TABLE_LIST, md5),
+  FILE_OPTIONS_STRING},
+ {{(char*) STRING_WITH_LEN("updatable")},
+  offsetof(TABLE_LIST, updatable_view),
+  FILE_OPTIONS_ULONGLONG},
+ {{(char*) STRING_WITH_LEN("algorithm")},
+  offsetof(TABLE_LIST, algorithm),
+  FILE_OPTIONS_ULONGLONG},
+ {{(char*) STRING_WITH_LEN("definer_user")},
+  offsetof(TABLE_LIST, definer.user),
   FILE_OPTIONS_STRING},
- {{(char*) "md5", 3},		offsetof(TABLE_LIST, md5),
+ {{(char*) STRING_WITH_LEN("definer_host")},
+  offsetof(TABLE_LIST, definer.host),
   FILE_OPTIONS_STRING},
- {{(char*) "updatable", 9},	offsetof(TABLE_LIST, updatable_view),
+ {{(char*) STRING_WITH_LEN("suid")},
+  offsetof(TABLE_LIST, view_suid),
   FILE_OPTIONS_ULONGLONG},
- {{(char*) "algorithm", 9},	offsetof(TABLE_LIST, algorithm),
+ {{(char*) STRING_WITH_LEN("with_check_option")},
+  offsetof(TABLE_LIST, with_check),
   FILE_OPTIONS_ULONGLONG},
- {{(char*) "with_check_option", 17}, offsetof(TABLE_LIST, with_check),
-   FILE_OPTIONS_ULONGLONG},
- {{(char*) "revision", 8},	offsetof(TABLE_LIST, revision),
+ {{(char*) STRING_WITH_LEN("revision")},
+  offsetof(TABLE_LIST, revision),
   FILE_OPTIONS_REV},
- {{(char*) "timestamp", 9},	offsetof(TABLE_LIST, timestamp),
+ {{(char*) STRING_WITH_LEN("timestamp")},
+  offsetof(TABLE_LIST, timestamp),
   FILE_OPTIONS_TIMESTAMP},
- {{(char*)"create-version", 14},offsetof(TABLE_LIST, file_version),
+ {{(char*)STRING_WITH_LEN("create-version")},
+  offsetof(TABLE_LIST, file_version),
   FILE_OPTIONS_ULONGLONG},
- {{(char*) "source", 6},	offsetof(TABLE_LIST, source),
+ {{(char*) STRING_WITH_LEN("source")},
+  offsetof(TABLE_LIST, source),
   FILE_OPTIONS_ESTRING},
  {{NullS, 0},			0,
   FILE_OPTIONS_STRING}
@@ -587,8 +636,9 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
   /* fill structure */
   view->query.str= (char*)str.ptr();
   view->query.length= str.length()-1; // we do not need last \0
-  view->source.str= thd->query;
-  view->source.length= thd->query_length;
+  view->source.str= thd->lex->create_view_select_start;
+  view->source.length= (thd->query_length -
+                        (thd->lex->create_view_select_start - thd->query));
   view->file_version= 1;
   view->calc_md5(md5);
   view->md5.str= md5;
@@ -602,6 +652,9 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
     lex->create_view_algorithm= VIEW_ALGORITHM_UNDEFINED;
   }
   view->algorithm= lex->create_view_algorithm;
+  view->definer.user= lex->create_view_definer->user;
+  view->definer.host= lex->create_view_definer->host;
+  view->view_suid= lex->create_view_suid;
   view->with_check= lex->create_view_check;
   if ((view->updatable_view= (can_be_merged &&
                               view->algorithm != VIEW_ALGORITHM_TMPTABLE)))
@@ -709,6 +762,11 @@ mysql_make_view(File_parser *parser, TABLE_LIST *table)
   /* init timestamp */
   if (!table->timestamp.str)
     table->timestamp.str= table->timestamp_buffer;
+  /* prepare default values for old format */
+  table->view_suid= 1;
+  table->definer.user.str= table->definer.host.str= 0;
+  table->definer.user.length= table->definer.host.length= 0;
+
   /*
     TODO: when VIEWs will be stored in cache, table mem_root should
     be used here
@@ -717,6 +775,21 @@ mysql_make_view(File_parser *parser, TABLE_LIST *table)
                     required_view_parameters))
     goto err;
 
+  /*
+    check old format view .frm
+  */
+  if (!table->definer.user.str)
+  {
+    DBUG_ASSERT(!table->definer.host.str &&
+                !table->definer.user.length &&
+                !table->definer.host.length);
+    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                        ER_VIEW_FRM_NO_USER, ER(ER_VIEW_FRM_NO_USER),
+                        table->db, table->table_name);
+    if (default_view_definer(thd, &table->definer))
+      goto err;
+  }
+
   /*
     Save VIEW parameters, which will be wiped out by derived table
     processing
@@ -1162,7 +1235,7 @@ bool check_key_in_view(THD *thd, TABLE_LIST *view)
     {
       if (!fld->item->fixed && fld->item->fix_fields(thd, &fld->item))
       {
-        thd->set_query_id= save_set_query_id;        
+        thd->set_query_id= save_set_query_id;
         return TRUE;
       }
     }

sql/sql_yacc.yy

@@ -823,10 +823,11 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 	precision subselect_start opt_and charset
 	subselect_end select_var_list select_var_list_init help opt_len
 	opt_extended_describe
-        prepare prepare_src execute deallocate 
+        prepare prepare_src execute deallocate
 	statement sp_suid opt_view_list view_list or_replace algorithm
 	sp_c_chistics sp_a_chistics sp_chistic sp_c_chistic xa
         load_data opt_field_or_var_spec fields_or_vars opt_load_data_set_spec
+        view_user view_suid
 END_OF_INPUT
 
 %type <NONE> call sp_proc_stmts sp_proc_stmts1 sp_proc_stmt
@@ -1254,16 +1255,16 @@ create:
 	      YYTHD->client_capabilities |= CLIENT_MULTI_QUERIES;
 	    sp->restore_thd_mem_root(YYTHD);
 	  }
-	| CREATE or_replace algorithm VIEW_SYM table_ident
+	| CREATE or_replace algorithm view_user view_suid VIEW_SYM table_ident
 	  {
 	    THD *thd= YYTHD;
 	    LEX *lex= thd->lex;
 	    lex->sql_command= SQLCOM_CREATE_VIEW;
 	    /* first table in list is target VIEW name */
-	    if (!lex->select_lex.add_table_to_list(thd, $5, NULL, 0))
+	    if (!lex->select_lex.add_table_to_list(thd, $7, NULL, 0))
               YYABORT;
 	  }
-	  opt_view_list AS select_init check_option
+	  opt_view_list AS select_view_init check_option
 	  {}
         | CREATE TRIGGER_SYM sp_name trg_action_time trg_event 
           ON table_ident FOR_SYM EACH_SYM ROW_SYM
@@ -3418,16 +3419,16 @@ alter:
 	    lex->sql_command= SQLCOM_ALTER_FUNCTION;
 	    lex->spname= $3;
 	  }
-	| ALTER algorithm VIEW_SYM table_ident
+	| ALTER algorithm view_user view_suid VIEW_SYM table_ident
 	  {
 	    THD *thd= YYTHD;
 	    LEX *lex= thd->lex;
 	    lex->sql_command= SQLCOM_CREATE_VIEW;
 	    lex->create_view_mode= VIEW_ALTER;
 	    /* first table in list is target VIEW name */
-	    lex->select_lex.add_table_to_list(thd, $4, NULL, 0);
+	    lex->select_lex.add_table_to_list(thd, $6, NULL, 0);
 	  }
-	  opt_view_list AS select_init check_option
+	  opt_view_list AS select_view_init check_option
 	  {}
 	;
 
@@ -3995,6 +3996,18 @@ select_init:
 	|
 	'(' select_paren ')' union_opt;
 
+select_view_init:
+        SELECT_SYM remember_name select_init2
+          {
+            Lex->create_view_select_start= $2;
+          }
+        |
+        '(' remember_name select_paren ')' union_opt
+          {
+            Lex->create_view_select_start= $2;
+          }
+        ;
+
 select_paren:
 	SELECT_SYM select_part2
 	  {
@@ -8885,6 +8898,53 @@ algorithm:
 	| ALGORITHM_SYM EQ TEMPTABLE_SYM
 	  { Lex->create_view_algorithm= VIEW_ALGORITHM_TMPTABLE; }
 	;
+
+view_user:
+        /* empty */
+          {
+            THD *thd= YYTHD;
+            if (!(thd->lex->create_view_definer=
+                  (LEX_USER*) thd->alloc(sizeof(st_lex_user))))
+              YYABORT;
+            if (default_view_definer(thd, thd->lex->create_view_definer))
+              YYABORT;
+          }
+        | CURRENT_USER optional_braces
+          {
+            THD *thd= YYTHD;
+            if (!(thd->lex->create_view_definer=
+                  (LEX_USER*) thd->alloc(sizeof(st_lex_user))))
+              YYABORT;
+            if (default_view_definer(thd, thd->lex->create_view_definer))
+              YYABORT;
+          }
+	| DEFINER_SYM EQ ident_or_text '@' ident_or_text
+	  {
+	    THD *thd= YYTHD;
+            st_lex_user *view_user;
+	    if (!(thd->lex->create_view_definer= view_user=
+                  (LEX_USER*) thd->alloc(sizeof(st_lex_user))))
+	      YYABORT;
+	    view_user->user = $3; view_user->host=$5;
+            if (strchr(view_user->host.str, wild_many) ||
+                strchr(view_user->host.str, wild_one))
+            {
+              my_error(ER_NO_VIEW_USER, MYF(0));
+              YYABORT;
+            }
+	  }
+        ;
+
+view_suid:
+        /* empty */
+	  { Lex->create_view_suid= TRUE; }
+        |
+	  SQL_SYM SECURITY_SYM DEFINER_SYM
+	  { Lex->create_view_suid= TRUE; }
+	| SQL_SYM SECURITY_SYM INVOKER_SYM
+	  { Lex->create_view_suid= FALSE; }
+	;
+
 check_option:
         /* empty */
           { Lex->create_view_check= VIEW_CHECK_NONE; }

sql/table.h

@@ -539,10 +539,12 @@ typedef struct st_table_list
   LEX_STRING	view_db;		/* saved view database */
   LEX_STRING	view_name;		/* saved view name */
   LEX_STRING	timestamp;		/* GMT time stamp of last operation */
+  st_lex_user   definer;                /* definer of view */
   ulonglong	file_version;		/* version of file's field set */
   ulonglong     updatable_view;         /* VIEW can be updated */
   ulonglong	revision;		/* revision control number */
   ulonglong	algorithm;		/* 0 any, 1 tmp tables , 2 merging */
+  ulonglong     view_suid;              /* view is suid (TRUE dy default) */
   ulonglong     with_check;             /* WITH CHECK OPTION */
   /*
     effective value of WITH CHECK OPTION (differ for temporary table
@@ -578,6 +580,8 @@ typedef struct st_table_list
   bool          multitable_view;        /* TRUE iff this is multitable view */
   /* view where processed */
   bool          where_processed;
+  /* db part was not defined in table definition */
+  bool          current_db_used;
   /* FRMTYPE_ERROR if any type is acceptable */
   enum frm_type_enum required_type;
   char		timestamp_buffer[20];	/* buffer for timestamp (19+1) */