WL#2818 (Add creator to the trigger definition for privilege

checks on trigger activation)


mysql-test/r/information_schema.result:
  Update result file: a new column DEFINER has been added to
  INFORMATION_SCHEMA.TRIGGERS.
mysql-test/r/mysqldump.result:
  Update result file: a new column DEFINER has been added to
  INFORMATION_SCHEMA.TRIGGERS.
mysql-test/r/rpl_ddl.result:
  Update result file: a new column DEFINER has been added to
  INFORMATION_SCHEMA.TRIGGERS.
mysql-test/r/rpl_sp.result:
  Update result file: a new clause DEFINER has been added to
  CREATE TRIGGER statement.
mysql-test/r/rpl_trigger.result:
  Results for new test cases were added.
mysql-test/r/skip_grants.result:
  Error message has been changed.
mysql-test/r/trigger.result:
  Added DEFINER column.
mysql-test/r/view.result:
  Error messages have been changed.
mysql-test/r/view_grant.result:
  Error messages have been changed.
mysql-test/t/mysqldump.test:
  Drop created procedure to not affect further tests.
mysql-test/t/rpl_trigger.test:
  Add tests for new column in information schema.
mysql-test/t/skip_grants.test:
  Error tag has been renamed.
mysql-test/t/view.test:
  Error tag has been renamed.
mysql-test/t/view_grant.test:
  Error tag has been changed.
sql/item_func.cc:
  Fix typo in comments.
sql/mysql_priv.h:
  A try to minimize copy&paste:
    - introduce operations to be used from sql_yacc.yy;
    - introduce an operation to be used from trigger and
      view processing code.
sql/share/errmsg.txt:
  - Rename ER_NO_VIEW_USER to ER_MALFORMED_DEFINER in order to
    be shared for view and trigger implementations;
  - Fix a typo;
  - Add a new error code for trigger warning.
sql/sp.cc:
  set_info() was split into set_info() and set_definer().
sql/sp_head.cc:
  set_info() was split into set_info() and set_definer().
sql/sp_head.h:
  set_info() was split into set_info() and set_definer().
sql/sql_acl.cc:
  Add a new check: exit from the cycle if the table is NULL.
sql/sql_lex.h:
  - Rename create_view_definer to definer, since it is used for views
    and triggers;
  - Change st_lex_user to LEX_USER, since st_lex_user is a structure.
    So, formally, it should be "struct st_lex_user", which is longer
    than just LEX_USER;
  - Add trigger_definition_begin.
sql/sql_parse.cc:
  - Add a new check: exit from the cycle if the table is NULL;
  - Implement definer-related functions.
sql/sql_show.cc:
  Add DEFINER column.
sql/sql_trigger.cc:
  Add DEFINER support for triggers.
sql/sql_trigger.h:
  Add DEFINER support for triggers.
sql/sql_view.cc:
  Rename create_view_definer to definer.
sql/sql_yacc.yy:
  Add support for DEFINER-clause in CREATE TRIGGER statement.
  
  Since CREATE TRIGGER and CREATE VIEW can be similar at the start,
  yacc is unable to distinguish between them. So, had to modify both
  statements in order to make it parsable by yacc.
mysql-test/r/trigger-compat.result:
  Result file for triggers backward compatibility test.
mysql-test/r/trigger-grant.result:
  Result file of the test for WL#2818.
mysql-test/t/trigger-compat.test:
  Triggers backward compatibility test: check that the server
  still can load triggers w/o definer attribute and modify
  tables with such triggers (add a new trigger, etc).
mysql-test/t/trigger-grant.test:
  Test for WL#2818 -- check that DEFINER support in triggers
  works properly

14 files changed, 683 insertions, 253 deletions

diff --git a/sql/item_func.cc b/sql/item_func.cc
index f467981540b..f5d5cac7e17 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -4894,7 +4894,7 @@ Item_func_sp::tmp_table_field(TABLE *t_arg)
 
 
 /*
-  Find the function and chack access rigths to the function
+  Find the function and check access rights to the function
 
   SYNOPSIS
     find_and_check_access()
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index b7bd34c613f..feede38c286 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -522,8 +522,9 @@ 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(Security_context *sctx, st_lex_user *definer);
 
+bool get_default_definer(THD *thd, LEX_USER *definer);
+LEX_USER *create_definer(THD *thd, LEX_STRING *user_name, LEX_STRING *host_name);
 
 enum enum_mysql_completiontype {
   ROLLBACK_RELEASE=-2, ROLLBACK=1,  ROLLBACK_AND_CHAIN=7,
@@ -847,6 +848,10 @@ bool mysqld_show_column_types(THD *thd);
 bool mysqld_help (THD *thd, const char *text);
 void calc_sum_of_all_status(STATUS_VAR *to);
 
+void append_definer(THD *thd, String *buffer, const LEX_STRING *definer_user,
+                    const LEX_STRING *definer_host);
+
+
 /* information schema */
 extern LEX_STRING information_schema_name;
 LEX_STRING *make_lex_string(THD *thd, LEX_STRING *lex_str,
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index f85bda90e81..b35a31081a3 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5405,14 +5405,14 @@ 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_SP_CANT_SET_AUTOCOMMIT
 	eng "Not allowed to set autocommit from a stored function or trigger"
-ER_NO_VIEW_USER
-        eng "View definer is not fully qualified"
+ER_MALFORMED_DEFINER
+	eng "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"
+	eng "You need the SUPER privilege for creation view with %-.64s@%-.64s definer"
 ER_NO_SUCH_USER
-        eng "There is not %-.64s@%-.64s registered"
+        eng "There is no '%-.64s'@'%-.64s' registered"
 ER_FORBID_SCHEMA_CHANGE
 	eng "Changing schema from '%-.64s' to '%-.64s' is not allowed."
 ER_ROW_IS_REFERENCED_2 23000
@@ -5421,3 +5421,5 @@ ER_NO_REFERENCED_ROW_2 23000
 	eng "Cannot add or update a child row: a foreign key constraint fails (%.192s)"
 ER_SP_BAD_VAR_SHADOW 42000
 	eng "Variable '%-.64s' must be quoted with `...`, or renamed"
+ER_TRG_NO_DEFINER
+  eng "No definer attribute for trigger '%-.64s'.'%-.64s'. The trigger will be activated under the authorization of the invoker, which may have insufficient privileges. Please recreate the trigger."
diff --git a/sql/sp.cc b/sql/sp.cc
index 8386c5d58a2..8d6e093c70d 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -441,8 +441,8 @@ db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
       if (dbchanged && (ret= mysql_change_db(thd, olddb, 1)))
 	goto done;
       *sphp= thd->lex->sphead;
-      (*sphp)->set_info((char *)definer, (uint)strlen(definer),
-			created, modified, &chistics, sql_mode);
+      (*sphp)->set_definer((char*) definer, (uint) strlen(definer));
+      (*sphp)->set_info(created, modified, &chistics, sql_mode);
       (*sphp)->optimize();
     }
     thd->lex->sql_command= oldcmd;
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index abc66ce0b21..08a189165b5 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -1569,21 +1569,9 @@ sp_head::check_backpatch(THD *thd)
 }
 
 void
-sp_head::set_info(char *definer, uint definerlen,
-		  longlong created, longlong modified,
+sp_head::set_info(longlong created, longlong modified,
 		  st_sp_chistics *chistics, ulong sql_mode)
 {
-  char *p= strchr(definer, '@');
-  uint len;
-
-  if (! p)
-    p= definer;		// Weird...
-  len= p-definer;
-  m_definer_user.str= strmake_root(mem_root, definer, len);
-  m_definer_user.length= len;
-  len= definerlen-len-1;
-  m_definer_host.str= strmake_root(mem_root, p+1, len);
-  m_definer_host.length= len;
   m_created= created;
   m_modified= modified;
   m_chistics= (st_sp_chistics *) memdup_root(mem_root, (char*) chistics,
@@ -1597,6 +1585,34 @@ sp_head::set_info(char *definer, uint definerlen,
   m_sql_mode= sql_mode;
 }
 
+
+void
+sp_head::set_definer(char *definer, uint definerlen)
+{
+  char *p= strrchr(definer, '@');
+
+  if (!p)
+  {
+    m_definer_user.str= strmake_root(mem_root, "", 0);
+    m_definer_user.length= 0;
+    
+    m_definer_host.str= strmake_root(mem_root, "", 0);
+    m_definer_host.length= 0;
+  }
+  else
+  {
+    const uint user_name_len= p - definer;
+    const uint host_name_len= definerlen - user_name_len - 1;
+
+    m_definer_user.str= strmake_root(mem_root, definer, user_name_len);
+    m_definer_user.length= user_name_len;
+
+    m_definer_host.str= strmake_root(mem_root, p + 1, host_name_len);
+    m_definer_host.length= host_name_len;
+  }
+}
+
+
 void
 sp_head::reset_thd_mem_root(THD *thd)
 {
diff --git a/sql/sp_head.h b/sql/sp_head.h
index ed0f3987e01..d1a122fd410 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -251,10 +251,11 @@ public:
 
   Field *make_field(uint max_length, const char *name, TABLE *dummy);
 
-  void set_info(char *definer, uint definerlen,
-		longlong created, longlong modified,
+  void set_info(longlong created, longlong modified,
 		st_sp_chistics *chistics, ulong sql_mode);
 
+  void set_definer(char *definer, uint definerlen);
+
   void reset_thd_mem_root(THD *thd);
 
   void restore_thd_mem_root(THD *thd);
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 2b5945b74af..bc8b9ba2efb 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3532,7 +3532,7 @@ bool check_grant(THD *thd, ulong want_access, TABLE_LIST *tables,
     of other queries). For simple queries first_not_own_table is 0.
   */
   for (i= 0, table= tables;
-       table != first_not_own_table && i < number;
+       table && table != first_not_own_table && i < number;
        table= table->next_global, i++)
   {
     /* Remove SHOW_VIEW_ACL, because it will be checked during making view */
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index a8bee9bb59b..0e836b6e9b9 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -737,10 +737,15 @@ 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_start;
   char *create_view_select_start;
 
+  /*
+    The definer of the object being created (view, trigger, stored routine).
+    I.e. the value of DEFINER clause.
+  */
+  LEX_USER *definer;
+
   List<key_part_spec> col_list;
   List<key_part_spec> ref_list;
   List<String>	      interval_list;
@@ -888,6 +893,14 @@ typedef struct st_lex
   SQL_LIST trg_table_fields;
 
   /*
+    trigger_definition_begin points to the beginning of the word "TRIGGER" in
+    CREATE TRIGGER statement. This is used to add possibly omitted DEFINER
+    clause to the trigger definition statement before dumping it to the
+    binlog. 
+  */
+  const char *trigger_definition_begin;
+
+  /*
     If non-0 then indicates that query requires prelocking and points to
     next_global member of last own element in query table list (i.e. last
     table which was not added to it as part of preparation to prelocking).
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 4bbca55f6ba..96e7393e4d6 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5062,7 +5062,7 @@ check_table_access(THD *thd, ulong want_access,TABLE_LIST *tables,
     the given table list refers to the list for prelocking (contains tables
     of other queries). For simple queries first_not_own_table is 0.
   */
-  for (; tables != first_not_own_table; tables= tables->next_global)
+  for (; tables && tables != first_not_own_table; tables= tables->next_global)
   {
     if (tables->schema_table && 
         (want_access & ~(SELECT_ACL | EXTRA_ACL | FILE_ACL)))
@@ -7466,32 +7466,81 @@ Item *negate_expression(THD *thd, Item *expr)
   return new Item_func_not(expr);
 }
 
+/*
+  Set the specified definer to the default value, which is the current user in
+  the thread. Also check that the current user satisfies to the definers
+  requirements.
+ 
+  SYNOPSIS
+    get_default_definer()
+    thd       [in] thread handler
+    definer   [out] definer
+ 
+  RETURN
+    error status, that is:
+      - FALSE -- on success;
+      - TRUE -- on error (current user can not be a definer).
+*/
+ 
+bool get_default_definer(THD *thd, LEX_USER *definer)
+{
+  /* Check that current user has non-empty host name. */
+
+  const Security_context *sctx= thd->security_ctx;
+
+  if (sctx->priv_host[0] == 0)
+  {
+    my_error(ER_MALFORMED_DEFINER, MYF(0));
+    return TRUE;
+  }
+
+  /* Fill in. */
+
+  definer->user.str= (char *) sctx->priv_user;
+  definer->user.length= strlen(definer->user.str);
+
+  definer->host.str= (char *) sctx->priv_host;
+  definer->host.length= strlen(definer->host.str);
+
+  return FALSE;
+}
+
 
 /*
-  Assign as view definer current user
+  Create definer with the given user and host names. Also check that the user
+  and host names satisfy definers requirements.
 
   SYNOPSIS
-    default_view_definer()
-    sctx		current security context
-    definer             structure where it should be assigned
+    create_definer()
+    thd         [in] thread handler
+    user_name   [in] user name
+    host_name   [in] host name
 
   RETURN
-    FALSE   OK
-    TRUE    Error
+    On success, return a valid pointer to the created and initialized
+    LEX_STRING, which contains definer information.
+    On error, return 0.
 */
 
-bool default_view_definer(Security_context *sctx, st_lex_user *definer)
+LEX_USER *create_definer(THD *thd, LEX_STRING *user_name, LEX_STRING *host_name)
 {
-  definer->user.str= sctx->priv_user;
-  definer->user.length= strlen(sctx->priv_user);
+  LEX_USER *definer;
+
+  /* Check that specified host name is valid. */
 
-  if (!*sctx->priv_host)
+  if (host_name->length == 0)
   {
-    my_error(ER_NO_VIEW_USER, MYF(0));
-    return TRUE;
+    my_error(ER_MALFORMED_DEFINER, MYF(0));
+    return 0;
   }
 
-  definer->host.str= sctx->priv_host;
-  definer->host.length= strlen(sctx->priv_host);
-  return FALSE;
+  /* Create and initialize. */
+
+  if (! (definer= (LEX_USER*) thd->alloc(sizeof (LEX_USER))))
+    return 0;
+
+  definer->user= *user_name;
+  definer->host= *host_name;
+
+  return definer;
 }
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index b4b24e0b6be..0e42c45b273 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1060,18 +1060,36 @@ view_store_options(THD *thd, TABLE_LIST *table, String *buff)
   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);
+  append_definer(thd, buff, &table->definer.user, &table->definer.host);
   if (table->view_suid)
-    buff->append(" SQL SECURITY DEFINER ", 22);
+    buff->append("SQL SECURITY DEFINER ", 21);
   else
-    buff->append(" SQL SECURITY INVOKER ", 22);
+    buff->append("SQL SECURITY INVOKER ", 21);
 }
 
+
+/*
+  Append DEFINER clause to the given buffer.
+  
+  SYNOPSIS
+    append_definer()
+    thd           [in] thread handle
+    buffer        [inout] buffer to hold DEFINER clause
+    definer_user  [in] user name part of definer
+    definer_host  [in] host name part of definer
+*/
+
+void append_definer(THD *thd, String *buffer, const LEX_STRING *definer_user,
+                    const LEX_STRING *definer_host)
+{
+  buffer->append(STRING_WITH_LEN("DEFINER="));
+  append_identifier(thd, buffer, definer_user->str, definer_user->length);
+  buffer->append('@');
+  append_identifier(thd, buffer, definer_host->str, definer_host->length);
+  buffer->append(' ');
+}
+
+
 static int
 view_store_create_info(THD *thd, TABLE_LIST *table, String *buff)
 {
@@ -3094,7 +3112,8 @@ static bool store_trigger(THD *thd, TABLE *table, const char *db,
                           enum trg_event_type event,
                           enum trg_action_time_type timing,
                           LEX_STRING *trigger_stmt,
-                          ulong sql_mode)
+                          ulong sql_mode,
+                          LEX_STRING *definer_buffer)
 {
   CHARSET_INFO *cs= system_charset_info;
   byte *sql_mode_str;
@@ -3119,6 +3138,7 @@ static bool store_trigger(THD *thd, TABLE *table, const char *db,
                                                        sql_mode,
                                                        &sql_mode_len);
   table->field[17]->store((const char*)sql_mode_str, sql_mode_len, cs);
+  table->field[18]->store((const char *)definer_buffer->str, definer_buffer->length, cs);
   return schema_table_store_record(thd, table);
 }
 
@@ -3152,15 +3172,21 @@ static int get_schema_triggers_record(THD *thd, struct st_table_list *tables,
         LEX_STRING trigger_name;
         LEX_STRING trigger_stmt;
         ulong sql_mode;
+        char definer_holder[HOSTNAME_LENGTH + USERNAME_LENGTH + 2];
+        LEX_STRING definer_buffer;
+        definer_buffer.str= definer_holder;
         if (triggers->get_trigger_info(thd, (enum trg_event_type) event,
                                        (enum trg_action_time_type)timing,
                                        &trigger_name, &trigger_stmt,
-                                       &sql_mode))
+                                       &sql_mode,
+                                       &definer_buffer))
           continue;
+
         if (store_trigger(thd, table, base_name, file_name, &trigger_name,
                          (enum trg_event_type) event,
                          (enum trg_action_time_type) timing, &trigger_stmt,
-                         sql_mode))
+                         sql_mode,
+                         &definer_buffer))
           DBUG_RETURN(1);
       }
     }
@@ -4064,6 +4090,7 @@ ST_FIELD_INFO triggers_fields_info[]=
   {"ACTION_REFERENCE_NEW_ROW", 3, MYSQL_TYPE_STRING, 0, 0, 0},
   {"CREATED", 0, MYSQL_TYPE_TIMESTAMP, 0, 1, "Created"},
   {"SQL_MODE", 65535, MYSQL_TYPE_STRING, 0, 0, "sql_mode"},
+  {"DEFINER", 65535, MYSQL_TYPE_STRING, 0, 0, "Definer"},
   {0, 0, MYSQL_TYPE_STRING, 0, 0, 0}
 };
 
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index dbad8dcffb5..3ab389d0519 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -32,15 +32,36 @@ const char * const triggers_file_ext= ".TRG";
 */
 static File_option triggers_file_parameters[]=
 {
-  {{(char*)"triggers", 8},
+  {
+    { (char *) STRING_WITH_LEN("triggers") },
     offsetof(class Table_triggers_list, definitions_list),
-    FILE_OPTIONS_STRLIST},
-  {{(char*)"sql_modes", 13},
+    FILE_OPTIONS_STRLIST
+  },
+  {
+    /*
+      FIXME: Length specified for "sql_modes" key is erroneous, problem caused
+      by this are reported as BUG#14090 and should be fixed ASAP.
+    */
+    { (char *) "sql_modes", 13 },
     offsetof(class Table_triggers_list, definition_modes_list),
-    FILE_OPTIONS_ULLLIST},
-  {{0, 0}, 0, FILE_OPTIONS_STRING}
+    FILE_OPTIONS_ULLLIST
+  },
+  {
+    { (char *) STRING_WITH_LEN("definers") },
+    offsetof(class Table_triggers_list, definers_list),
+    FILE_OPTIONS_STRLIST
+  },
+  { { 0, 0 }, 0, FILE_OPTIONS_STRING }
 };
 
+/*
+  This must be kept up to date whenever a new option is added to the list
+  above, as it specifies the number of required parameters of the trigger in
+  .trg file.
+*/
+
+static const int TRG_NUM_REQUIRED_PARAMETERS= 4;
+static const int TRG_MAX_VERSIONS= 3;
 
 /*
   Structure representing contents of .TRN file which are used to support
@@ -58,9 +79,16 @@ const char * const trigname_file_ext= ".TRN";
 
 static File_option trigname_file_parameters[]=
 {
-  {{(char*)"trigger_table", 15}, offsetof(struct st_trigname, trigger_table),
-   FILE_OPTIONS_ESTRING},
-  {{0, 0}, 0, FILE_OPTIONS_STRING}
+  {
+    /*
+      FIXME: Length specified for "trigger_table" key is erroneous, problem
+      caused by this are reported as BUG#14090 and should be fixed ASAP.
+    */
+    { (char *) "trigger_table", 15 },
+    offsetof(struct st_trigname, trigger_table),
+   FILE_OPTIONS_ESTRING
+  },
+  { { 0, 0 }, 0, FILE_OPTIONS_STRING }
 };
 
 
@@ -104,6 +132,9 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
 {
   TABLE *table;
   bool result= TRUE;
+  LEX_STRING definer_user;
+  LEX_STRING definer_host;
+
   DBUG_ENTER("mysql_create_or_drop_trigger");
 
   /*
@@ -184,7 +215,7 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
   }
 
   result= (create ?
-           table->triggers->create_trigger(thd, tables):
+           table->triggers->create_trigger(thd, tables, &definer_user, &definer_host):
            table->triggers->drop_trigger(thd, tables));
 
 end:
@@ -192,17 +223,30 @@ end:
   start_waiting_global_read_lock(thd);
 
   if (!result)
+  {
+    if (mysql_bin_log.is_open())
     {
-      if (mysql_bin_log.is_open())
+      thd->clear_error();
+
+      String log_query(thd->query, thd->query_length, system_charset_info);
+
+      if (create)
       {
-	thd->clear_error();
-	/* Such a statement can always go directly to binlog, no trans cache */
-	Query_log_event qinfo(thd, thd->query, thd->query_length, 0, FALSE);
-	mysql_bin_log.write(&qinfo);
+        log_query.set((char *) 0, 0, system_charset_info); /* reset log_query */
+
+        log_query.append("CREATE ");
+        append_definer(thd, &log_query, &definer_user, &definer_host);
+        log_query.append(thd->lex->trigger_definition_begin);
       }
-      send_ok(thd);
+
+      /* Such a statement can always go directly to binlog, no trans cache. */
+      Query_log_event qinfo(thd, log_query.ptr(), log_query.length(), 0, FALSE);
+      mysql_bin_log.write(&qinfo);
     }
 
+    send_ok(thd);
+  }
+
   DBUG_RETURN(result);
 }
 
@@ -212,15 +256,26 @@ end:
 
   SYNOPSIS
     create_trigger()
-      thd    - current thread context (including trigger definition in LEX)
-      tables - table list containing one open table for which trigger is
-               created.
+      thd          - current thread context (including trigger definition in
+                     LEX)
+      tables       - table list containing one open table for which the
+                     trigger is created.
+      definer_user - [out] after a call it points to 0-terminated string,
+                     which contains user name part of the actual trigger
+                     definer. The caller is responsible to provide memory for
+                     storing LEX_STRING object.
+      definer_host - [out] after a call it points to 0-terminated string,
+                     which contains host name part of the actual trigger
+                     definer. The caller is responsible to provide memory for
+                     storing LEX_STRING object.
 
   RETURN VALUE
     False - success
     True  - error
 */
-bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables)
+bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
+                                         LEX_STRING *definer_user,
+                                         LEX_STRING *definer_host)
 {
   LEX *lex= thd->lex;
   TABLE *table= tables->table;
@@ -229,6 +284,8 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables)
   LEX_STRING dir, file, trigname_file;
   LEX_STRING *trg_def, *name;
   ulonglong *trg_sql_mode;
+  char trg_definer_holder[HOSTNAME_LENGTH + USERNAME_LENGTH + 2];
+  LEX_STRING *trg_definer;
   Item_trigger_field *trg_field;
   struct st_trigname trigname;
 
@@ -250,6 +307,31 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables)
   }
 
   /*
+    Definer attribute of the Lex instance is always set in sql_yacc.yy when
+    trigger is created.
+  */
+
+  DBUG_ASSERT(lex->definer);
+
+  /*
+    If the specified definer differs from the current user, we should check
+    that the current user has SUPER privilege (in order to create trigger
+    under another user one must have SUPER privilege).
+  */
+  
+  if (strcmp(lex->definer->user.str, thd->security_ctx->priv_user) ||
+      my_strcasecmp(system_charset_info,
+                    lex->definer->host.str,
+                    thd->security_ctx->priv_host))
+  {
+    if (check_global_access(thd, SUPER_ACL))
+    {
+      my_error(ER_SPECIFIC_ACCESS_DENIED_ERROR, MYF(0), "SUPER");
+      return TRUE;
+    }
+  }
+
+  /*
     Let us check if all references to fields in old/new versions of row in
     this trigger are ok.
 
@@ -318,15 +400,39 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables)
       definitions_list.push_back(trg_def, &table->mem_root) ||
       !(trg_sql_mode= (ulonglong*)alloc_root(&table->mem_root,
                                              sizeof(ulonglong))) ||
-      definition_modes_list.push_back(trg_sql_mode, &table->mem_root))
+      definition_modes_list.push_back(trg_sql_mode, &table->mem_root) ||
+      !(trg_definer= (LEX_STRING*) alloc_root(&table->mem_root,
+                                              sizeof(LEX_STRING))) ||
+      definers_list.push_back(trg_definer, &table->mem_root))
     goto err_with_cleanup;
 
   trg_def->str= thd->query;
   trg_def->length= thd->query_length;
   *trg_sql_mode= thd->variables.sql_mode;
 
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+  if (!is_acl_user(lex->definer->host.str,
+      lex->definer->user.str))
+  {
+    push_warning_printf(thd,
+                        MYSQL_ERROR::WARN_LEVEL_NOTE,
+                        ER_NO_SUCH_USER,
+                        ER(ER_NO_SUCH_USER),
+                        lex->definer->user.str,
+                        lex->definer->host.str);
+  }
+#endif /* NO_EMBEDDED_ACCESS_CHECKS */
+
+  *definer_user= lex->definer->user;
+  *definer_host= lex->definer->host;
+
+  trg_definer->str= trg_definer_holder;
+  trg_definer->length= strxmov(trg_definer->str, definer_user->str, "@",
+                               definer_host->str, NullS) - trg_definer->str;
+
   if (!sql_create_definition_file(&dir, &file, &triggers_file_type,
-                                  (gptr)this, triggers_file_parameters, 3))
+                                  (gptr)this, triggers_file_parameters,
+                                  TRG_MAX_VERSIONS))
     return 0;
 
 err_with_cleanup:
@@ -403,12 +509,14 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables)
   List_iterator_fast<LEX_STRING> it_name(names_list);
   List_iterator<LEX_STRING>      it_def(definitions_list);
   List_iterator<ulonglong>       it_mod(definition_modes_list);
+  List_iterator<LEX_STRING>      it_definer(definers_list);
   char path[FN_REFLEN];
 
   while ((name= it_name++))
   {
     it_def++;
     it_mod++;
+    it_definer++;
 
     if (my_strcasecmp(table_alias_charset, lex->spname->m_name.str,
                       name->str) == 0)
@@ -419,6 +527,7 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables)
       */
       it_def.remove();
       it_mod.remove();
+      it_definer.remove();
 
       if (definitions_list.is_empty())
       {
@@ -446,7 +555,7 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables)
 
         if (sql_create_definition_file(&dir, &file, &triggers_file_type,
                                        (gptr)this, triggers_file_parameters,
-                                       3))
+                                       TRG_MAX_VERSIONS))
           return 1;
       }
 
@@ -568,7 +677,7 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
     DBUG_RETURN(0);
 
   /*
-    File exists so we got to load triggers
+    File exists so we got to load triggers.
     FIXME: A lot of things to do here e.g. how about other funcs and being
     more paranoical ?
   */
@@ -584,13 +693,16 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
         DBUG_RETURN(1);
 
       /*
-        We don't have sql_modes in old versions of .TRG file, so we should
-        initialize list for safety.
+        We don't have the following attributes in old versions of .TRG file, so
+        we should initialize the list for safety:
+          - sql_modes;
+          - definers;
       */
       triggers->definition_modes_list.empty();
+      triggers->definers_list.empty();
 
       if (parser->parse((gptr)triggers, &table->mem_root,
-                        triggers_file_parameters, 2))
+                        triggers_file_parameters, TRG_NUM_REQUIRED_PARAMETERS))
         DBUG_RETURN(1);
 
       List_iterator_fast<LEX_STRING> it(triggers->definitions_list);
@@ -612,7 +724,7 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
           DBUG_RETURN(1); // EOM
         }
         *trg_sql_mode= global_system_variables.sql_mode;
-        while ((trg_create_str= it++))
+        while (it++)
         {
           if (triggers->definition_modes_list.push_back(trg_sql_mode,
                                                         &table->mem_root))
@@ -623,8 +735,43 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
         it.rewind();
       }
 
+      if (triggers->definers_list.is_empty() &&
+          !triggers->definitions_list.is_empty())
+      {
+        /*
+          It is old file format => we should fill list of definers.
+
+          If there is no definer information, we should not switch context to
+          definer when checking privileges. I.e. privileges for such triggers
+          are checked for "invoker" rather than for "definer".
+        */
+
+        LEX_STRING *trg_definer;
+
+        if (! (trg_definer= (LEX_STRING*)alloc_root(&table->mem_root,
+                                                    sizeof(LEX_STRING))))
+          DBUG_RETURN(1); // EOM
+
+        trg_definer->str= "";
+        trg_definer->length= 0;
+
+        while (it++)
+        {
+          if (triggers->definers_list.push_back(trg_definer,
+                                                &table->mem_root))
+          {
+            DBUG_RETURN(1); // EOM
+          }
+        }
+
+        it.rewind();
+      }
+
       DBUG_ASSERT(triggers->definition_modes_list.elements ==
                   triggers->definitions_list.elements);
+      DBUG_ASSERT(triggers->definers_list.elements ==
+                  triggers->definitions_list.elements);
+
       table->triggers= triggers;
 
       /*
@@ -647,6 +794,8 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
 
       char *trg_name_buff;
       List_iterator_fast<ulonglong> itm(triggers->definition_modes_list);
+      List_iterator_fast<LEX_STRING> it_definer(triggers->
+                                                definers_list);
       LEX *old_lex= thd->lex, lex;
       ulong save_sql_mode= thd->variables.sql_mode;
 
@@ -659,22 +808,55 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
       while ((trg_create_str= it++))
       {
         trg_sql_mode= itm++;
+        LEX_STRING *trg_definer= it_definer++;
         thd->variables.sql_mode= (ulong)*trg_sql_mode;
         lex_start(thd, (uchar*)trg_create_str->str, trg_create_str->length);
 
         if (yyparse((void *)thd) || thd->is_fatal_error)
         {
           /*
-            Free lex associated resources
+            Free lex associated resources.
             QQ: Do we really need all this stuff here ?
           */
           delete lex.sphead;
           goto err_with_lex_cleanup;
         }
 
-        lex.sphead->m_sql_mode= *trg_sql_mode;
+        lex.sphead->set_info(0, 0, &lex.sp_chistics, *trg_sql_mode);
+
         triggers->bodies[lex.trg_chistics.event]
                              [lex.trg_chistics.action_time]= lex.sphead;
+
+        if (!trg_definer->length)
+        {
+          /*
+            This trigger was created/imported from the previous version of
+            MySQL, which does not support triggers definers. We should emit
+            warning here.
+          */
+
+          push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                              ER_TRG_NO_DEFINER, ER(ER_TRG_NO_DEFINER),
+                              (const char*) db,
+                              (const char*) lex.sphead->m_name.str);
+
+          /*
+            Set definer to the '' to correct displaying in the information
+            schema.
+          */
+
+          lex.sphead->set_definer("", 0);
+
+          /*
+            Triggers without definer information are executed under the
+            authorization of the invoker.
+          */
+
+          lex.sphead->m_chistics->suid= SP_IS_NOT_SUID;
+        }
+        else
+          lex.sphead->set_definer(trg_definer->str, trg_definer->length);
+
         if (triggers->names_list.push_back(&lex.sphead->m_name,
                                            &table->mem_root))
             goto err_with_lex_cleanup;
@@ -701,6 +883,10 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
              trg_field= trg_field->next_trg_field)
           trg_field->setup_field(thd, table);
 
+        triggers->m_spec_var_used[lex.trg_chistics.event]
+          [lex.trg_chistics.action_time]=
+          lex.trg_table_fields.first ? TRUE : FALSE;
+
         lex_end(&lex);
       }
       thd->db= save_db.str;
@@ -744,6 +930,9 @@ err_with_lex_cleanup:
       name      - returns name of trigger
       stmt      - returns statement of trigger
       sql_mode  - returns sql_mode of trigger
+      definer_user - returns definer/creator of trigger. The caller is
+                  responsible to allocate enough space for storing definer
+                  information.
 
   RETURN VALUE
     False - success
@@ -754,7 +943,8 @@ bool Table_triggers_list::get_trigger_info(THD *thd, trg_event_type event,
                                            trg_action_time_type time_type,
                                            LEX_STRING *trigger_name,
                                            LEX_STRING *trigger_stmt,
-                                           ulong *sql_mode)
+                                           ulong *sql_mode,
+                                           LEX_STRING *definer)
 {
   sp_head *body;
   DBUG_ENTER("get_trigger_info");
@@ -763,6 +953,18 @@ bool Table_triggers_list::get_trigger_info(THD *thd, trg_event_type event,
     *trigger_name= body->m_name;
     *trigger_stmt= body->m_body;
     *sql_mode= body->m_sql_mode;
+
+    if (body->m_chistics->suid == SP_IS_NOT_SUID)
+    {
+      definer->str[0]= 0;
+      definer->length= 0;
+    }
+    else
+    {
+      definer->length= strxmov(definer->str, body->m_definer_user.str, "@",
+                               body->m_definer_host.str, NullS) - definer->str;
+    }
+
     DBUG_RETURN(0);
   }
   DBUG_RETURN(1);
@@ -898,8 +1100,9 @@ bool Table_triggers_list::process_triggers(THD *thd, trg_event_type event,
                                            bool old_row_is_record1)
 {
   int res= 0;
+  sp_head *sp_trigger= bodies[event][time_type];
 
-  if (bodies[event][time_type])
+  if (sp_trigger)
   {
     Sub_statement_state statement_state;
 
@@ -914,14 +1117,54 @@ bool Table_triggers_list::process_triggers(THD *thd, trg_event_type event,
       old_field= table->field;
     }
 
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+    Security_context *save_ctx;
+
+    if (sp_change_security_context(thd, sp_trigger, &save_ctx))
+      return TRUE;
+
     /*
-      FIXME: We should juggle with security context here (because trigger
-      should be invoked with creator rights).
+      NOTE: TRIGGER_ACL should be used below.
     */
 
+    if (check_global_access(thd, SUPER_ACL))
+    {
+      sp_restore_security_context(thd, save_ctx);
+      return TRUE;
+    }
+
+    /*
+      If the trigger uses special variables (NEW/OLD), check that we have
+      SELECT and UPDATE privileges on the subject table.
+    */
+    
+    if (is_special_var_used(event, time_type))
+    {
+      TABLE_LIST table_list;
+      bzero((char *) &table_list, sizeof (table_list));
+      table_list.db= (char *) table->s->db;
+      table_list.db_length= strlen(table_list.db);
+      table_list.table_name= (char *) table->s->table_name;
+      table_list.table_name_length= strlen(table_list.table_name);
+      table_list.alias= (char *) table->alias;
+      table_list.table= table;
+
+      if (check_table_access(thd, SELECT_ACL | UPDATE_ACL, &table_list, 0))
+      {
+        sp_restore_security_context(thd, save_ctx);
+        return TRUE;
+      }
+    }
+    
+#endif // NO_EMBEDDED_ACCESS_CHECKS
+
     thd->reset_sub_statement_state(&statement_state, SUB_STMT_TRIGGER);
-    res= bodies[event][time_type]->execute_function(thd, 0, 0, 0);
+    res= sp_trigger->execute_function(thd, 0, 0, 0);
     thd->restore_sub_statement_state(&statement_state);
+
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+    sp_restore_security_context(thd, save_ctx);
+#endif // NO_EMBEDDED_ACCESS_CHECKS
   }
 
   return res;
diff --git a/sql/sql_trigger.h b/sql/sql_trigger.h
index c1d1f8d0e9e..6be42d7b868 100644
--- a/sql/sql_trigger.h
+++ b/sql/sql_trigger.h
@@ -55,6 +55,12 @@ class Table_triggers_list: public Sql_alloc
   */
   LEX_STRING        sroutines_key;
 
+  /*
+    is_special_var_used specifies whether trigger body contains special
+    variables (NEW/OLD).
+  */
+  bool m_spec_var_used[TRG_EVENT_MAX][TRG_ACTION_MAX];
+
 public:
   /*
     Field responsible for storing triggers definitions in file.
@@ -66,6 +72,8 @@ public:
   */
   List<ulonglong> definition_modes_list;
 
+  List<LEX_STRING>  definers_list;
+
   Table_triggers_list(TABLE *table_arg):
     record1_field(0), table(table_arg)
   {
@@ -73,7 +81,9 @@ public:
   }
   ~Table_triggers_list();
 
-  bool create_trigger(THD *thd, TABLE_LIST *table);
+  bool create_trigger(THD *thd, TABLE_LIST *table,
+                      LEX_STRING *definer_user,
+                      LEX_STRING *definer_host);
   bool drop_trigger(THD *thd, TABLE_LIST *table);
   bool process_triggers(THD *thd, trg_event_type event,
                         trg_action_time_type time_type,
@@ -81,7 +91,8 @@ public:
   bool get_trigger_info(THD *thd, trg_event_type event,
                         trg_action_time_type time_type,
                         LEX_STRING *trigger_name, LEX_STRING *trigger_stmt,
-                        ulong *sql_mode);
+                        ulong *sql_mode,
+                        LEX_STRING *definer);
 
   static bool check_n_load(THD *thd, const char *db, const char *table_name,
                            TABLE *table, bool names_only);
@@ -98,6 +109,11 @@ public:
     return test(bodies[TRG_EVENT_UPDATE][TRG_ACTION_BEFORE]);
   }
 
+  inline bool is_special_var_used(int event, int action_time) const
+  {
+    return m_spec_var_used[event][action_time];
+  }
+
   void set_table(TABLE *new_table);
 
   friend class Item_trigger_field;
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 71c5d198b27..b642d24b30d 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -214,29 +214,28 @@ bool mysql_create_view(THD *thd,
       - same as current user
       - current user has SUPER_ACL
   */
-  if (strcmp(lex->create_view_definer->user.str,
+  if (strcmp(lex->definer->user.str,
              thd->security_ctx->priv_user) != 0 ||
       my_strcasecmp(system_charset_info,
-                    lex->create_view_definer->host.str,
+                    lex->definer->host.str,
                     thd->security_ctx->priv_host) != 0)
   {
     if (!(thd->security_ctx->master_access & SUPER_ACL))
     {
-      my_error(ER_VIEW_OTHER_USER, MYF(0), lex->create_view_definer->user.str,
-               lex->create_view_definer->host.str);
+      my_error(ER_SPECIFIC_ACCESS_DENIED_ERROR, MYF(0), "SUPER");
       res= TRUE;
       goto err;
     }
     else
     {
-      if (!is_acl_user(lex->create_view_definer->host.str,
-                       lex->create_view_definer->user.str))
+      if (!is_acl_user(lex->definer->host.str,
+                       lex->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);
+                            lex->definer->user.str,
+                            lex->definer->host.str);
       }
     }
   }
@@ -658,8 +657,8 @@ 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->definer.user= lex->definer->user;
+  view->definer.host= lex->definer->host;
   view->view_suid= lex->create_view_suid;
   view->with_check= lex->create_view_check;
   if ((view->updatable_view= (can_be_merged &&
@@ -807,7 +806,7 @@ bool mysql_make_view(THD *thd, File_parser *parser, TABLE_LIST *table)
     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->security_ctx, &table->definer))
+    if (get_default_definer(thd, &table->definer))
       goto err;
   }
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index abc440df1c5..55002def5e9 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -776,7 +776,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 
 %type <symbol> FUNC_ARG0 FUNC_ARG1 FUNC_ARG2 FUNC_ARG3 keyword keyword_sp
 
-%type <lex_user> user grant_user
+%type <lex_user> user grant_user get_definer
 
 %type <charset>
 	opt_collate
@@ -827,10 +827,12 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 	subselect_end select_var_list select_var_list_init help opt_len
 	opt_extended_describe
         prepare prepare_src execute deallocate
-	statement sp_suid opt_view_list view_list or_replace algorithm
+	statement sp_suid
 	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
+        definer view_replace_or_algorithm view_replace view_algorithm_opt
+        view_algorithm view_or_trigger_tail view_suid view_tail view_list_opt
+        view_list view_select view_check_option trigger_tail
 END_OF_INPUT
 
 %type <NONE> call sp_proc_stmts sp_proc_stmts1 sp_proc_stmt
@@ -1258,80 +1260,14 @@ create:
 	      YYTHD->client_capabilities |= CLIENT_MULTI_QUERIES;
 	    sp->restore_thd_mem_root(YYTHD);
 	  }
-	| CREATE or_replace algorithm view_user view_suid VIEW_SYM table_ident
+	| CREATE
 	  {
-	    THD *thd= YYTHD;
-	    LEX *lex= thd->lex;
-	    lex->sql_command= SQLCOM_CREATE_VIEW;
-            lex->create_view_start= thd->query;
-	    /* first table in list is target VIEW name */
-	    if (!lex->select_lex.add_table_to_list(thd, $7, NULL, 0))
-              YYABORT;
+            Lex->create_view_mode= VIEW_CREATE_NEW;
+            Lex->create_view_algorithm= VIEW_ALGORITHM_UNDEFINED;
+            Lex->create_view_suid= TRUE;
 	  }
-	  opt_view_list AS select_view_init check_option
+	  view_or_trigger
 	  {}
-        | CREATE TRIGGER_SYM sp_name trg_action_time trg_event 
-          ON table_ident FOR_SYM EACH_SYM ROW_SYM
-          {
-            LEX *lex= Lex;
-            sp_head *sp;
-           
-            if (lex->sphead)
-            {
-              my_error(ER_SP_NO_RECURSIVE_CREATE, MYF(0), "TRIGGER");
-              YYABORT;
-            }
-
-            if (!(sp= new sp_head()))
-              YYABORT;
-            sp->reset_thd_mem_root(YYTHD);
-            sp->init(lex);
-            
-            sp->m_type= TYPE_ENUM_TRIGGER;
-            lex->sphead= sp;
-            lex->spname= $3;
-            /*
-              We have to turn of CLIENT_MULTI_QUERIES while parsing a
-              stored procedure, otherwise yylex will chop it into pieces
-              at each ';'.
-            */
-            sp->m_old_cmq= YYTHD->client_capabilities & CLIENT_MULTI_QUERIES;
-            YYTHD->client_capabilities &= ~CLIENT_MULTI_QUERIES;
-            
-            bzero((char *)&lex->sp_chistics, sizeof(st_sp_chistics));
-            lex->sphead->m_chistics= &lex->sp_chistics;
-            lex->sphead->m_body_begin= lex->ptr;
-          }
-          sp_proc_stmt
-          {
-            LEX *lex= Lex;
-            sp_head *sp= lex->sphead;
-            
-            lex->sql_command= SQLCOM_CREATE_TRIGGER;
-            sp->init_strings(YYTHD, lex, $3);
-            /* Restore flag if it was cleared above */
-            if (sp->m_old_cmq)
-              YYTHD->client_capabilities |= CLIENT_MULTI_QUERIES;
-            sp->restore_thd_mem_root(YYTHD);
-
-            if (sp->is_not_allowed_in_function("trigger"))
-                YYABORT;
-
-            /*
-              We have to do it after parsing trigger body, because some of
-              sp_proc_stmt alternatives are not saving/restoring LEX, so
-              lex->query_tables can be wiped out.
-              
-              QQ: What are other consequences of this?
-              
-              QQ: Could we loosen lock type in certain cases ?
-            */
-            if (!lex->select_lex.add_table_to_list(YYTHD, $7, 
-                                                   (LEX_STRING*) 0,
-                                                   TL_OPTION_UPDATING,
-                                                   TL_WRITE))
-              YYABORT;
-          }
 	| CREATE USER clear_privileges grant_list
 	  {
 	    Lex->sql_command = SQLCOM_CREATE_USER;
@@ -3435,7 +3371,8 @@ alter:
 	    lex->sql_command= SQLCOM_ALTER_FUNCTION;
 	    lex->spname= $3;
 	  }
-	| ALTER algorithm view_user view_suid VIEW_SYM table_ident
+        | ALTER view_algorithm_opt definer view_suid
+          VIEW_SYM table_ident
 	  {
 	    THD *thd= YYTHD;
 	    LEX *lex= thd->lex;
@@ -3445,7 +3382,7 @@ alter:
 	    /* first table in list is target VIEW name */
 	    lex->select_lex.add_table_to_list(thd, $6, NULL, 0);
 	  }
-	  opt_view_list AS select_view_init check_option
+	  view_list_opt AS view_select view_check_option
 	  {}
 	;
 
@@ -4013,18 +3950,6 @@ 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
 	  {
@@ -8963,8 +8888,119 @@ subselect_end:
 	  lex->current_select = lex->current_select->return_after_parsing();
 	};
 
-opt_view_list:
-	/* empty */ {}
+definer:
+	get_definer
+	{
+	  THD *thd= YYTHD;
+	  
+	  if (! (thd->lex->definer= create_definer(thd, &$1->user, &$1->host)))
+	    YYABORT;
+	}
+	;
+
+get_definer:
+	opt_current_definer
+	{
+	  THD *thd= YYTHD;
+          
+	  if (!($$=(LEX_USER*) thd->alloc(sizeof(st_lex_user))))
+	    YYABORT;
+
+	  if (get_default_definer(thd, $$))
+	    YYABORT;
+	}
+	| DEFINER_SYM EQ ident_or_text '@' ident_or_text
+	{
+	  if (!($$=(LEX_USER*) YYTHD->alloc(sizeof(st_lex_user))))
+	    YYABORT;
+
+	  $$->user= $3;
+	  $$->host= $5;
+	}
+	;
+
+opt_current_definer:
+	/* empty */
+	| DEFINER_SYM EQ CURRENT_USER optional_braces
+	;
+
+/**************************************************************************
+
+ CREATE VIEW statement options.
+
+**************************************************************************/
+
+view_replace_or_algorithm:
+	view_replace
+	{}
+	| view_replace view_algorithm
+	{}
+	| view_algorithm
+	{}
+	;
+
+view_replace:
+	OR_SYM REPLACE
+	{ Lex->create_view_mode= VIEW_CREATE_OR_REPLACE; }
+	;
+
+view_algorithm:
+	ALGORITHM_SYM EQ UNDEFINED_SYM
+	{ Lex->create_view_algorithm= VIEW_ALGORITHM_UNDEFINED; }
+	| ALGORITHM_SYM EQ MERGE_SYM
+	{ Lex->create_view_algorithm= VIEW_ALGORITHM_MERGE; }
+	| ALGORITHM_SYM EQ TEMPTABLE_SYM
+	{ Lex->create_view_algorithm= VIEW_ALGORITHM_TMPTABLE; }
+	;
+
+view_algorithm_opt:
+	/* empty */
+	{ Lex->create_view_algorithm= VIEW_ALGORITHM_UNDEFINED; }
+	| view_algorithm
+	{}
+	;
+
+view_or_trigger:
+	definer view_or_trigger_tail
+	{}
+	| view_replace_or_algorithm definer view_tail
+	{}
+	;
+
+view_or_trigger_tail:
+	view_tail
+	{}
+	| trigger_tail
+	{}
+	;
+
+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; }
+	;
+
+view_tail:
+	view_suid VIEW_SYM table_ident
+	{
+	  THD *thd= YYTHD;
+	  LEX *lex= thd->lex;
+	  lex->sql_command= SQLCOM_CREATE_VIEW;
+	  lex->create_view_start= thd->query;
+	  /* first table in list is target VIEW name */
+	  if (!lex->select_lex.add_table_to_list(thd, $3, NULL, 0))
+	    YYABORT;
+	}
+	view_list_opt AS view_select view_check_option
+	{}
+	;
+
+view_list_opt:
+	/* empty */
+	{}
 	| '(' view_list ')'
 	;
 
@@ -8981,79 +9017,102 @@ view_list:
 	  }
 	;
 
-or_replace:
-	/* empty */	 { Lex->create_view_mode= VIEW_CREATE_NEW; }
-	| OR_SYM REPLACE { Lex->create_view_mode= VIEW_CREATE_OR_REPLACE; }
+view_select:
+	SELECT_SYM remember_name select_init2
+	{
+	  Lex->create_view_select_start= $2;
+	}
+	| '(' remember_name select_paren ')' union_opt
+	{
+	  Lex->create_view_select_start= $2;
+	}
 	;
 
-algorithm:
+view_check_option:
 	/* empty */
-	  { Lex->create_view_algorithm= VIEW_ALGORITHM_UNDEFINED; }
-	| ALGORITHM_SYM EQ UNDEFINED_SYM
-	  { Lex->create_view_algorithm= VIEW_ALGORITHM_UNDEFINED; }
-	| ALGORITHM_SYM EQ MERGE_SYM
-	  { Lex->create_view_algorithm= VIEW_ALGORITHM_MERGE; }
-	| ALGORITHM_SYM EQ TEMPTABLE_SYM
-	  { Lex->create_view_algorithm= VIEW_ALGORITHM_TMPTABLE; }
+	{ Lex->create_view_check= VIEW_CHECK_NONE; }
+	| WITH CHECK_SYM OPTION
+	{ Lex->create_view_check= VIEW_CHECK_CASCADED; }
+	| WITH CASCADED CHECK_SYM OPTION
+	{ Lex->create_view_check= VIEW_CHECK_CASCADED; }
+	| WITH LOCAL_SYM CHECK_SYM OPTION
+	{ Lex->create_view_check= VIEW_CHECK_LOCAL; }
 	;
 
-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->security_ctx,
-                                     thd->lex->create_view_definer))
-              YYABORT;
-          }
-        | DEFINER_SYM EQ 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->security_ctx,
-                                     thd->lex->create_view_definer))
-              YYABORT;
-          }
-	| DEFINER_SYM EQ ident_or_text '@' ident_or_text
+/**************************************************************************
+
+ CREATE TRIGGER statement parts.
+
+**************************************************************************/
+
+trigger_tail:
+	TRIGGER_SYM remember_name sp_name trg_action_time trg_event 
+	ON table_ident FOR_SYM EACH_SYM ROW_SYM
+	{
+	  LEX *lex= Lex;
+	  sp_head *sp;
+	 
+	  if (lex->sphead)
 	  {
-	    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 (view_user->host.length == 0)
-            {
-              my_error(ER_NO_VIEW_USER, MYF(0));
-              YYABORT;
-            }
+	    my_error(ER_SP_NO_RECURSIVE_CREATE, MYF(0), "TRIGGER");
+	    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; }
+	
+	  if (!(sp= new sp_head()))
+	    YYABORT;
+	  sp->reset_thd_mem_root(YYTHD);
+	  sp->init(lex);
+	
+	  lex->trigger_definition_begin= $2;
+	  
+	  sp->m_type= TYPE_ENUM_TRIGGER;
+	  lex->sphead= sp;
+	  lex->spname= $3;
+	  /*
+	    We have to turn of CLIENT_MULTI_QUERIES while parsing a
+	    stored procedure, otherwise yylex will chop it into pieces
+	    at each ';'.
+	  */
+	  sp->m_old_cmq= YYTHD->client_capabilities & CLIENT_MULTI_QUERIES;
+	  YYTHD->client_capabilities &= ~CLIENT_MULTI_QUERIES;
+	  
+	  bzero((char *)&lex->sp_chistics, sizeof(st_sp_chistics));
+	  lex->sphead->m_chistics= &lex->sp_chistics;
+	  lex->sphead->m_body_begin= lex->ptr;
+	}
+	sp_proc_stmt
+	{
+	  LEX *lex= Lex;
+	  sp_head *sp= lex->sphead;
+	  
+	  lex->sql_command= SQLCOM_CREATE_TRIGGER;
+	  sp->init_strings(YYTHD, lex, $3);
+	  /* Restore flag if it was cleared above */
+	  if (sp->m_old_cmq)
+	    YYTHD->client_capabilities |= CLIENT_MULTI_QUERIES;
+	  sp->restore_thd_mem_root(YYTHD);
+	
+	  if (sp->is_not_allowed_in_function("trigger"))
+	      YYABORT;
+	
+	  /*
+	    We have to do it after parsing trigger body, because some of
+	    sp_proc_stmt alternatives are not saving/restoring LEX, so
+	    lex->query_tables can be wiped out.
+	    
+	    QQ: What are other consequences of this?
+	    
+	    QQ: Could we loosen lock type in certain cases ?
+	  */
+	  if (!lex->select_lex.add_table_to_list(YYTHD, $7, 
+	                                         (LEX_STRING*) 0,
+	                                         TL_OPTION_UPDATING,
+	                                         TL_WRITE))
+	    YYABORT;
+	}
 	;
 
-check_option:
-        /* empty */
-          { Lex->create_view_check= VIEW_CHECK_NONE; }
-        | WITH CHECK_SYM OPTION
-          { Lex->create_view_check= VIEW_CHECK_CASCADED; }
-        | WITH CASCADED CHECK_SYM OPTION
-          { Lex->create_view_check= VIEW_CHECK_CASCADED; }
-        | WITH LOCAL_SYM CHECK_SYM OPTION
-          { Lex->create_view_check= VIEW_CHECK_LOCAL; }
-        ;
+/*************************************************************************/
 
 xa: XA_SYM begin_or_start xid opt_join_or_resume
       {